diff --git a/tests/kafka/kafka_client_test.go b/tests/kafka/kafka_client_test.go
index 76d63c6..12f0ae4 100644
--- a/tests/kafka/kafka_client_test.go
+++ b/tests/kafka/kafka_client_test.go
@@ -22,7 +22,7 @@
 	"github.com/google/uuid"
 	"github.com/opencord/voltha-go/common/log"
 	kk "github.com/opencord/voltha-go/kafka"
-	ca "github.com/opencord/voltha-go/protos/core_adapter"
+	ic "github.com/opencord/voltha-go/protos/inter_container"
 	"github.com/stretchr/testify/assert"
 	"os"
 	"testing"
@@ -63,7 +63,7 @@
 	numMessageToSend = 1
 }
 
-func waitForMessage(ch <-chan *ca.InterContainerMessage, doneCh chan string, maxMessages int) {
+func waitForMessage(ch <-chan *ic.InterContainerMessage, doneCh chan string, maxMessages int) {
 	totalTime = 0
 	totalMessageReceived = 0
 	mytime := time.Now()
@@ -92,17 +92,17 @@
 func sendMessages(topic *kk.Topic, numMessages int, fn sendToKafka) error {
 	// Loop for numMessages
 	for i := 0; i < numMessages; i++ {
-		msg := &ca.InterContainerMessage{}
-		msg.Header = &ca.Header{
+		msg := &ic.InterContainerMessage{}
+		msg.Header = &ic.Header{
 			Id:        uuid.New().String(),
-			Type:      ca.MessageType_REQUEST,
+			Type:      ic.MessageType_REQUEST,
 			FromTopic: topic.Name,
 			ToTopic:   topic.Name,
 			Timestamp: time.Now().UnixNano(),
 		}
 		var marshalledArg *any.Any
 		var err error
-		body := &ca.InterContainerRequestBody{Rpc: "testRPC", Args: []*ca.Argument{}}
+		body := &ic.InterContainerRequestBody{Rpc: "testRPC", Args: []*ic.Argument{}}
 		if marshalledArg, err = ptypes.MarshalAny(body); err != nil {
 			log.Warnw("cannot-marshal-request", log.Fields{"error": err})
 			return err
@@ -116,7 +116,7 @@
 }
 
 func runWithPartionConsumer(topic *kk.Topic, numMessages int, doneCh chan string) error {
-	var ch <-chan *ca.InterContainerMessage
+	var ch <-chan *ic.InterContainerMessage
 	var err error
 	if ch, err = partionClient.Subscribe(topic); err != nil {
 		return nil
@@ -130,7 +130,7 @@
 }
 
 func runWithGroupConsumer(topic *kk.Topic, numMessages int, doneCh chan string) error {
-	var ch <-chan *ca.InterContainerMessage
+	var ch <-chan *ic.InterContainerMessage
 	var err error
 	if ch, err = groupClient.Subscribe(topic); err != nil {
 		return nil
diff --git a/tests/kafka/kafka_inter_container_messaging_test.go b/tests/kafka/kafka_inter_container_messaging_test.go
index 0293d6d..57e7ab9 100644
--- a/tests/kafka/kafka_inter_container_messaging_test.go
+++ b/tests/kafka/kafka_inter_container_messaging_test.go
@@ -21,10 +21,11 @@
 	"github.com/google/uuid"
 	"github.com/opencord/voltha-go/common/log"
 	kk "github.com/opencord/voltha-go/kafka"
-	ca "github.com/opencord/voltha-go/protos/core_adapter"
+	ic "github.com/opencord/voltha-go/protos/inter_container"
 	"github.com/opencord/voltha-go/protos/voltha"
 	rhp "github.com/opencord/voltha-go/rw_core/core"
 	"github.com/stretchr/testify/assert"
+	"os"
 	"testing"
 	"time"
 )
@@ -35,27 +36,42 @@
 
 var coreKafkaProxy *kk.InterContainerProxy
 var adapterKafkaProxy *kk.InterContainerProxy
+var kafkaPartitionClient kk.Client
+var affinityRouterTopic string
+var hostIP string
+var kafkaClient kk.Client
 
 func init() {
 	log.AddPackage(log.JSON, log.ErrorLevel, nil)
 	log.UpdateAllLoggers(log.Fields{"instanceId": "testing"})
 	log.SetAllLogLevel(log.ErrorLevel)
-	kafkaClient := kk.NewSaramaClient(
-		kk.Host("10.176.212.108"),
+	affinityRouterTopic = "AffinityRouter"
+	hostIP = os.Getenv("DOCKER_HOST_IP")
+	kafkaClient = kk.NewSaramaClient(
+		kk.Host(hostIP),
 		kk.Port(9092))
 
 	coreKafkaProxy, _ = kk.NewInterContainerProxy(
-		kk.InterContainerHost("10.176.212.108"),
+		kk.InterContainerHost(hostIP),
 		kk.InterContainerPort(9092),
 		kk.DefaultTopic(&kk.Topic{Name: "Core"}),
-		kk.MsgClient(kafkaClient))
+		kk.MsgClient(kafkaClient),
+		kk.DeviceDiscoveryTopic(&kk.Topic{Name: affinityRouterTopic}))
 
 	adapterKafkaProxy, _ = kk.NewInterContainerProxy(
-		kk.InterContainerHost("10.176.212.108"),
+		kk.InterContainerHost(hostIP),
 		kk.InterContainerPort(9092),
 		kk.DefaultTopic(&kk.Topic{Name: "Adapter"}),
 		kk.MsgClient(kafkaClient))
 
+	kafkaPartitionClient = kk.NewSaramaClient(
+		kk.ConsumerType(kk.PartitionConsumer),
+		kk.Host(hostIP),
+		kk.Port(9092),
+		kk.AutoCreateTopic(true),
+		kk.ProducerFlushFrequency(5))
+	kafkaPartitionClient.Start()
+
 	coreKafkaProxy.Start()
 	adapterKafkaProxy.Start()
 	subscribeTarget(coreKafkaProxy)
@@ -67,11 +83,11 @@
 	kmp.SubscribeWithRequestHandlerInterface(topic, requestProxy)
 }
 
-func waitForRPCMessage(topic kk.Topic, ch <-chan *ca.InterContainerMessage, doneCh chan string) {
+func waitForRPCMessage(topic kk.Topic, ch <-chan *ic.InterContainerMessage, doneCh chan string) {
 	for msg := range ch {
 		log.Debugw("Got-RPC-message", log.Fields{"msg": msg})
 		//	Unpack message
-		requestBody := &ca.InterContainerRequestBody{}
+		requestBody := &ic.InterContainerRequestBody{}
 		if err := ptypes.UnmarshalAny(msg.Body, requestBody); err != nil {
 			doneCh <- "Error"
 		} else {
@@ -106,8 +122,8 @@
 //	// First subscribe to the specific topic
 //	//log.SetPackageLogLevel("github.com/opencord/voltha-go/kafka", log.DebugLevel)
 //	var err error
-//	var ch1 <-chan *ca.InterContainerMessage
-//	var ch2 <-chan *ca.InterContainerMessage
+//	var ch1 <-chan *ic.InterContainerMessage
+//	var ch2 <-chan *ic.InterContainerMessage
 //	topic := kk.Topic{Name: "Core"}
 //	ch1, err = coreKafkaProxy.Subscribe(topic)
 //	assert.NotNil(t, ch1)
@@ -159,7 +175,7 @@
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, false)
 	//Unpack the result into the actual proto object
-	unpackResult := &ca.Error{}
+	unpackResult := &ic.Error{}
 	if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
 		log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
 	}
@@ -177,12 +193,12 @@
 	rpc := "GetDevice"
 	topic := kk.Topic{Name: "Core"}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, false)
 	//Unpack the result into the actual proto object
-	unpackResult := &ca.Error{}
+	unpackResult := &ic.Error{}
 	if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
 		log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
 	}
@@ -204,7 +220,7 @@
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(ctx, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(ctx, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, true)
@@ -229,11 +245,11 @@
 	ctx, cancel := context.WithTimeout(context.Background(), timeout)
 	defer cancel()
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(ctx, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(ctx, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, false)
-	unpackResult := &ca.Error{}
+	unpackResult := &ic.Error{}
 	if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
 		log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
 	}
@@ -252,7 +268,7 @@
 	topic := kk.Topic{Name: "Core"}
 	expectedResponse := &voltha.Device{Id: trnsId}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, true)
@@ -275,7 +291,7 @@
 	topic := kk.Topic{Name: "Core"}
 	expectedResponse := &voltha.Device{Id: trnsId}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, true)
@@ -294,7 +310,7 @@
 		Key:   "deviceID",
 		Value: protoArg1,
 	}
-	protoArg2 := &ca.IntType{Val: 1}
+	protoArg2 := &ic.IntType{Val: 1}
 	args[1] = &kk.KVArg{
 		Key:   "portType",
 		Value: protoArg2,
@@ -302,7 +318,7 @@
 	rpc := "GetPorts"
 	topic := kk.Topic{Name: "Core"}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, true)
@@ -325,12 +341,12 @@
 	rpc := "GetPorts"
 	topic := kk.Topic{Name: "Core"}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, false)
 	//Unpack the result into the actual proto object
-	unpackResult := &ca.Error{}
+	unpackResult := &ic.Error{}
 	if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
 		log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
 	}
@@ -339,18 +355,18 @@
 
 func TestChildDeviceDetected(t *testing.T) {
 	trnsId := uuid.New().String()
-	protoArg1 := &ca.StrType{Val: trnsId}
+	protoArg1 := &ic.StrType{Val: trnsId}
 	args := make([]*kk.KVArg, 5)
 	args[0] = &kk.KVArg{
 		Key:   "deviceID",
 		Value: protoArg1,
 	}
-	protoArg2 := &ca.IntType{Val: 1}
+	protoArg2 := &ic.IntType{Val: 1}
 	args[1] = &kk.KVArg{
 		Key:   "parentPortNo",
 		Value: protoArg2,
 	}
-	protoArg3 := &ca.StrType{Val: "great_onu"}
+	protoArg3 := &ic.StrType{Val: "great_onu"}
 	args[2] = &kk.KVArg{
 		Key:   "childDeviceType",
 		Value: protoArg3,
@@ -360,7 +376,7 @@
 		Key:   "proxyAddress",
 		Value: protoArg4,
 	}
-	protoArg5 := &ca.IntType{Val: 1}
+	protoArg5 := &ic.IntType{Val: 1}
 	args[4] = &kk.KVArg{
 		Key:   "portType",
 		Value: protoArg5,
@@ -369,7 +385,7 @@
 	rpc := "ChildDeviceDetected"
 	topic := kk.Topic{Name: "Core"}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, true)
@@ -378,18 +394,18 @@
 
 func TestChildDeviceDetectedNoWait(t *testing.T) {
 	trnsId := uuid.New().String()
-	protoArg1 := &ca.StrType{Val: trnsId}
+	protoArg1 := &ic.StrType{Val: trnsId}
 	args := make([]*kk.KVArg, 5)
 	args[0] = &kk.KVArg{
 		Key:   "deviceID",
 		Value: protoArg1,
 	}
-	protoArg2 := &ca.IntType{Val: 1}
+	protoArg2 := &ic.IntType{Val: 1}
 	args[1] = &kk.KVArg{
 		Key:   "parentPortNo",
 		Value: protoArg2,
 	}
-	protoArg3 := &ca.StrType{Val: "great_onu"}
+	protoArg3 := &ic.StrType{Val: "great_onu"}
 	args[2] = &kk.KVArg{
 		Key:   "childDeviceType",
 		Value: protoArg3,
@@ -399,7 +415,7 @@
 		Key:   "proxyAddress",
 		Value: protoArg4,
 	}
-	protoArg5 := &ca.IntType{Val: 1}
+	protoArg5 := &ic.IntType{Val: 1}
 	args[4] = &kk.KVArg{
 		Key:   "portType",
 		Value: protoArg5,
@@ -408,7 +424,7 @@
 	rpc := "ChildDeviceDetected"
 	topic := kk.Topic{Name: "Core"}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,false, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, false, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, true)
@@ -417,18 +433,18 @@
 
 func TestChildDeviceDetectedMissingArgs(t *testing.T) {
 	trnsId := uuid.New().String()
-	protoArg1 := &ca.StrType{Val: trnsId}
+	protoArg1 := &ic.StrType{Val: trnsId}
 	args := make([]*kk.KVArg, 4)
 	args[0] = &kk.KVArg{
 		Key:   "deviceID",
 		Value: protoArg1,
 	}
-	protoArg2 := &ca.IntType{Val: 1}
+	protoArg2 := &ic.IntType{Val: 1}
 	args[1] = &kk.KVArg{
 		Key:   "parentPortNo",
 		Value: protoArg2,
 	}
-	protoArg3 := &ca.StrType{Val: "great_onu"}
+	protoArg3 := &ic.StrType{Val: "great_onu"}
 	args[2] = &kk.KVArg{
 		Key:   "childDeviceType",
 		Value: protoArg3,
@@ -437,11 +453,11 @@
 	rpc := "ChildDeviceDetected"
 	topic := kk.Topic{Name: "Core"}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, false)
-	unpackResult := &ca.Error{}
+	unpackResult := &ic.Error{}
 	if err := ptypes.UnmarshalAny(result, unpackResult); err != nil {
 		log.Warnw("cannot-unmarshal-response", log.Fields{"error": err})
 	}
@@ -457,12 +473,12 @@
 		Key:   "device_id",
 		Value: protoArg1,
 	}
-	protoArg2 := &ca.IntType{Val: 1}
+	protoArg2 := &ic.IntType{Val: 1}
 	args[1] = &kk.KVArg{
 		Key:   "oper_status",
 		Value: protoArg2,
 	}
-	protoArg3 := &ca.IntType{Val: 1}
+	protoArg3 := &ic.IntType{Val: 1}
 	args[2] = &kk.KVArg{
 		Key:   "connect_status",
 		Value: protoArg3,
@@ -471,13 +487,58 @@
 	rpc := "DeviceStateUpdate"
 	topic := kk.Topic{Name: "Core"}
 	start := time.Now()
-	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic,true, args...)
+	status, result := adapterKafkaProxy.InvokeRPC(nil, rpc, &topic, &topic, true, args...)
 	elapsed := time.Since(start)
 	log.Infow("Result", log.Fields{"status": status, "result": result, "time": elapsed})
 	assert.Equal(t, status, true)
 	assert.Nil(t, result)
 }
 
+func subscribeToTopic(topic *kk.Topic, waitingChannel chan *ic.InterContainerMessage) error {
+	var ch <-chan *ic.InterContainerMessage
+	var err error
+	if ch, err = kafkaPartitionClient.Subscribe(topic); err != nil {
+		return nil
+	}
+	msg := <-ch
+
+	log.Debugw("msg-received", log.Fields{"msg": msg})
+	waitingChannel <- msg
+	return nil
+}
+
+func TestDeviceDiscovery(t *testing.T) {
+	// Create an intercontainer proxy - similar to the Core
+	testProxy, _ := kk.NewInterContainerProxy(
+		kk.InterContainerHost(hostIP),
+		kk.InterContainerPort(9092),
+		kk.DefaultTopic(&kk.Topic{Name: "Test"}),
+		kk.MsgClient(kafkaClient),
+		kk.DeviceDiscoveryTopic(&kk.Topic{Name: affinityRouterTopic}))
+
+	//	First start to wait for the message
+	waitingChannel := make(chan *ic.InterContainerMessage)
+	go subscribeToTopic(&kk.Topic{Name: affinityRouterTopic}, waitingChannel)
+
+	// Sleep to make sure the consumer is ready
+	time.Sleep(time.Millisecond * 100)
+
+	// Send the message
+	go testProxy.DeviceDiscovered("TestDeviceId", "TestDevicetype", "TestParentId")
+
+	msg := <-waitingChannel
+	totalTime := (time.Now().UnixNano() - msg.Header.Timestamp) / int64(time.Millisecond)
+	assert.Equal(t, msg.Header.Type, ic.MessageType_DEVICE_DISCOVERED)
+	//	Unpack message
+	dd := &ic.DeviceDiscovered{}
+	err := ptypes.UnmarshalAny(msg.Body, dd)
+	assert.Nil(t, err)
+	assert.Equal(t, dd.Id, "TestDeviceId")
+	assert.Equal(t, dd.DeviceType, "TestDevicetype")
+	assert.Equal(t, dd.ParentId, "TestParentId")
+	log.Debugw("TotalTime", log.Fields{"time": totalTime})
+}
+
 func TestStopKafkaProxy(t *testing.T) {
 	adapterKafkaProxy.Stop()
 	coreKafkaProxy.Stop()