diff --git a/adapters/common/core_proxy.go b/adapters/common/core_proxy.go
index f076910..a503c97 100644
--- a/adapters/common/core_proxy.go
+++ b/adapters/common/core_proxy.go
@@ -25,12 +25,16 @@
 	"github.com/opencord/voltha-go/protos/voltha"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"sync"
 )
 
 type CoreProxy struct {
 	kafkaICProxy *kafka.InterContainerProxy
 	adapterTopic string
 	coreTopic    string
+	deviceIdCoreMap map[string]string
+	lockDeviceIdCoreMap sync.RWMutex
+
 }
 
 func NewCoreProxy(kafkaProxy *kafka.InterContainerProxy, adapterTopic string, coreTopic string) *CoreProxy {
@@ -38,6 +42,8 @@
 	proxy.kafkaICProxy = kafkaProxy
 	proxy.adapterTopic = adapterTopic
 	proxy.coreTopic = coreTopic
+	proxy.deviceIdCoreMap = make(map[string]string)
+	proxy.lockDeviceIdCoreMap = sync.RWMutex{}
 	log.Debugw("TOPICS", log.Fields{"core": proxy.coreTopic, "adapter": proxy.adapterTopic})
 
 	return &proxy
@@ -58,11 +64,40 @@
 	}
 }
 
+// UpdateCoreReference adds or update a core reference (really the topic name) for a given device Id
+func (ap *CoreProxy) UpdateCoreReference(deviceId string, coreReference string) {
+	ap.lockDeviceIdCoreMap.Lock()
+	defer ap.lockDeviceIdCoreMap.Unlock()
+	ap.deviceIdCoreMap[deviceId] = coreReference
+}
+
+// DeleteCoreReference removes a core reference (really the topic name) for a given device Id
+func (ap *CoreProxy) DeleteCoreReference(deviceId string) {
+	ap.lockDeviceIdCoreMap.Lock()
+	defer ap.lockDeviceIdCoreMap.Unlock()
+	delete(ap.deviceIdCoreMap, deviceId)
+}
+
+func (ap *CoreProxy) getCoreTopic(deviceId string) kafka.Topic {
+	ap.lockDeviceIdCoreMap.Lock()
+	defer ap.lockDeviceIdCoreMap.Unlock()
+
+	if t, exist := ap.deviceIdCoreMap[deviceId]; exist {
+		return kafka.Topic{Name: t}
+	}
+
+	return kafka.Topic{Name: ap.coreTopic}
+}
+
+func (ap *CoreProxy) getAdapterTopic(args ...string) kafka.Topic {
+	return kafka.Topic{Name: ap.adapterTopic}
+}
+
 func (ap *CoreProxy) RegisterAdapter(ctx context.Context, adapter *voltha.Adapter, deviceTypes *voltha.DeviceTypes) error {
 	log.Debugw("registering-adapter", log.Fields{"coreTopic": ap.coreTopic, "adapterTopic": ap.adapterTopic})
 	rpc := "Register"
 	topic := kafka.Topic{Name: ap.coreTopic}
-	replyToTopic := kafka.Topic{Name: ap.adapterTopic}
+	replyToTopic := ap.getAdapterTopic()
 	args := make([]*kafka.KVArg, 2)
 	args[0] = &kafka.KVArg{
 		Key:   "adapter",
@@ -81,16 +116,14 @@
 func (ap *CoreProxy) DeviceUpdate(ctx context.Context, device *voltha.Device) error {
 	log.Debugw("DeviceUpdate", log.Fields{"deviceId": device.Id})
 	rpc := "DeviceUpdate"
-	// Use a device specific topic to send the request.  The adapter handling the device creates a device
-	// specific topic
-	toTopic := kafka.CreateSubTopic(ap.coreTopic, device.Id)
+	toTopic := ap.getCoreTopic(device.Id)
 	args := make([]*kafka.KVArg, 1)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
 		Value: device,
 	}
 	// Use a device specific topic as we are the only adaptercore handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.adapterTopic, device.Id)
+	replyToTopic := ap.getAdapterTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(nil, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("DeviceUpdate-response", log.Fields{"deviceId": device.Id, "success": success})
 	return unPackResponse(rpc, device.Id, success, result)
@@ -101,7 +134,7 @@
 	rpc := "PortCreated"
 	// Use a device specific topic to send the request.  The adapter handling the device creates a device
 	// specific topic
-	toTopic := kafka.CreateSubTopic(ap.coreTopic, deviceId)
+	toTopic := ap.getCoreTopic(deviceId)
 	args := make([]*kafka.KVArg, 2)
 	id := &voltha.ID{Id: deviceId}
 	args[0] = &kafka.KVArg{
@@ -114,7 +147,7 @@
 	}
 
 	// Use a device specific topic as we are the only adaptercore handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.adapterTopic, deviceId)
+	replyToTopic := ap.getAdapterTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(nil, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("PortCreated-response", log.Fields{"deviceId": deviceId, "success": success})
 	return unPackResponse(rpc, deviceId, success, result)
@@ -126,7 +159,7 @@
 	rpc := "DeviceStateUpdate"
 	// Use a device specific topic to send the request.  The adapter handling the device creates a device
 	// specific topic
-	toTopic := kafka.CreateSubTopic(ap.coreTopic, deviceId)
+	toTopic := ap.getCoreTopic(deviceId)
 	args := make([]*kafka.KVArg, 3)
 	id := &voltha.ID{Id: deviceId}
 	oStatus := &ic.IntType{Val: int64(operStatus)}
@@ -145,7 +178,7 @@
 		Value: cStatus,
 	}
 	// Use a device specific topic as we are the only adaptercore handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.adapterTopic, deviceId)
+	replyToTopic := ap.getAdapterTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(nil, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("DeviceStateUpdate-response", log.Fields{"deviceId": deviceId, "success": success})
 	return unPackResponse(rpc, deviceId, success, result)
@@ -157,8 +190,8 @@
 	rpc := "ChildDeviceDetected"
 	// Use a device specific topic to send the request.  The adapter handling the device creates a device
 	// specific topic
-	toTopic := kafka.CreateSubTopic(ap.coreTopic, parentDeviceId)
-	replyToTopic := kafka.CreateSubTopic(ap.adapterTopic, parentDeviceId)
+	toTopic := ap.getCoreTopic(parentDeviceId)
+	replyToTopic := ap.getAdapterTopic()
 
 	args := make([]*kafka.KVArg, 4)
 	id := &voltha.ID{Id: parentDeviceId}
diff --git a/adapters/common/request_handler.go b/adapters/common/request_handler.go
index b3606b0..4d65d89 100644
--- a/adapters/common/request_handler.go
+++ b/adapters/common/request_handler.go
@@ -32,12 +32,14 @@
 	TestMode       bool
 	coreInstanceId string
 	adapter        adapters.IAdapter
+	coreProxy *CoreProxy
 }
 
-func NewRequestHandlerProxy(coreInstanceId string, iadapter adapters.IAdapter) *RequestHandlerProxy {
+func NewRequestHandlerProxy(coreInstanceId string, iadapter adapters.IAdapter, cProxy *CoreProxy) *RequestHandlerProxy {
 	var proxy RequestHandlerProxy
 	proxy.coreInstanceId = coreInstanceId
 	proxy.adapter = iadapter
+	proxy.coreProxy = cProxy
 	return &proxy
 }
 
@@ -54,13 +56,14 @@
 }
 
 func (rhp *RequestHandlerProxy) Adopt_device(args []*ic.Argument) (*empty.Empty, error) {
-	if len(args) != 2 {
+	if len(args) < 3 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
 	}
 	device := &voltha.Device{}
 	transactionID := &ic.StrType{}
+	fromTopic := &ic.StrType{}
 	for _, arg := range args {
 		switch arg.Key {
 		case "device":
@@ -73,11 +76,19 @@
 				log.Warnw("cannot-unmarshal-transaction-ID", log.Fields{"error": err})
 				return nil, err
 			}
+		case kafka.FromTopic:
+			if err := ptypes.UnmarshalAny(arg.Value, fromTopic); err != nil {
+				log.Warnw("cannot-unmarshal-from-topic", log.Fields{"error": err})
+				return nil, err
+			}
 		}
 	}
 
 	log.Debugw("Adopt_device", log.Fields{"deviceId": device.Id})
 
+	//Update the core reference for that device
+	rhp.coreProxy.UpdateCoreReference(device.Id, fromTopic.Val)
+
 	//Invoke the adopt device on the adapter
 	if err := rhp.adapter.Adopt_device(device); err != nil {
 		return nil, status.Errorf(codes.NotFound, "%s", err.Error())
@@ -143,7 +154,7 @@
 }
 
 func (rhp *RequestHandlerProxy) Get_ofp_device_info(args []*ic.Argument) (*ic.SwitchCapability, error) {
-	if len(args) != 2 {
+	if len(args) < 2 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
@@ -177,7 +188,7 @@
 }
 
 func (rhp *RequestHandlerProxy) Get_ofp_port_info(args []*ic.Argument) (*ic.PortCapability, error) {
-	if len(args) != 3 {
+	if len(args) < 3 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
@@ -214,7 +225,7 @@
 }
 
 func (rhp *RequestHandlerProxy) Process_inter_adapter_message(args []*ic.Argument) (*empty.Empty, error) {
-	if len(args) != 2 {
+	if len(args) < 2 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
diff --git a/adapters/simulated_olt/adaptercore/simulated_olt.go b/adapters/simulated_olt/adaptercore/simulated_olt.go
index edf3135..529b89d 100644
--- a/adapters/simulated_olt/adaptercore/simulated_olt.go
+++ b/adapters/simulated_olt/adaptercore/simulated_olt.go
@@ -96,16 +96,6 @@
 	return nil
 }
 
-func (so *SimulatedOLT) createDeviceTopic(device *voltha.Device) error {
-	log.Infow("create-device-topic", log.Fields{"deviceId": device.Id})
-	deviceTopic := kafka.Topic{Name: so.kafkaICProxy.DefaultTopic.Name + "_" + device.Id}
-	if err := so.kafkaICProxy.SubscribeWithDefaultRequestHandler(deviceTopic, kafka.OffsetOldest); err != nil {
-		log.Infow("create-device-topic-failed", log.Fields{"deviceId": device.Id, "error": err})
-		return err
-	}
-	return nil
-}
-
 func (so *SimulatedOLT) Adopt_device(device *voltha.Device) error {
 	if device == nil {
 		log.Warn("device-is-nil")
@@ -117,8 +107,6 @@
 		handler := NewDeviceHandler(so.coreProxy, device, so)
 		so.addDeviceHandlerToMap(handler)
 		go handler.AdoptDevice(device)
-		// Launch the creation of the device topic
-		go so.createDeviceTopic(device)
 	}
 	return nil
 }
diff --git a/adapters/simulated_olt/main.go b/adapters/simulated_olt/main.go
index 01234a9..ea6fb6c 100644
--- a/adapters/simulated_olt/main.go
+++ b/adapters/simulated_olt/main.go
@@ -91,7 +91,7 @@
 	}
 
 	// Register the core request handler
-	if err = a.setupRequestHandler(a.instanceId, a.iAdapter); err != nil {
+	if err = a.setupRequestHandler(a.instanceId, a.iAdapter, a.coreProxy); err != nil {
 		log.Fatal("error-setting-core-request-handler")
 	}
 
@@ -218,9 +218,9 @@
 	return sOLT, nil
 }
 
-func (a *adapter) setupRequestHandler(coreInstanceId string, iadapter adapters.IAdapter) error {
+func (a *adapter) setupRequestHandler(coreInstanceId string, iadapter adapters.IAdapter, coreProxy *com.CoreProxy) error {
 	log.Info("setting-request-handler")
-	requestProxy := com.NewRequestHandlerProxy(coreInstanceId, iadapter)
+	requestProxy := com.NewRequestHandlerProxy(coreInstanceId, iadapter, coreProxy)
 	if err := a.kip.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: a.config.Topic}, requestProxy); err != nil {
 		log.Errorw("request-handler-setup-failed", log.Fields{"error": err})
 		return err
diff --git a/adapters/simulated_onu/adaptercore/simulated_onu.go b/adapters/simulated_onu/adaptercore/simulated_onu.go
index 2d43944..c10b844 100644
--- a/adapters/simulated_onu/adaptercore/simulated_onu.go
+++ b/adapters/simulated_onu/adaptercore/simulated_onu.go
@@ -94,16 +94,6 @@
 	return nil
 }
 
-func (so *SimulatedONU) createDeviceTopic(device *voltha.Device) error {
-	log.Infow("create-device-topic", log.Fields{"deviceId": device.Id})
-	deviceTopic := kafka.Topic{Name: so.kafkaICProxy.DefaultTopic.Name + "_" + device.Id}
-	if err := so.kafkaICProxy.SubscribeWithDefaultRequestHandler(deviceTopic, kafka.OffsetOldest); err != nil {
-		log.Infow("create-device-topic-failed", log.Fields{"deviceId": device.Id, "error": err})
-		return err
-	}
-	return nil
-}
-
 func (so *SimulatedONU) Adopt_device(device *voltha.Device) error {
 	if device == nil {
 		log.Warn("device-is-nil")
@@ -115,8 +105,6 @@
 		handler := NewDeviceHandler(so.coreProxy, device, so)
 		so.addDeviceHandlerToMap(handler)
 		go handler.AdoptDevice(device)
-		// Launch the creation of the device topic
-		go so.createDeviceTopic(device)
 	}
 	return nil
 }
diff --git a/adapters/simulated_onu/main.go b/adapters/simulated_onu/main.go
index 62cf2db..a5c18aa 100644
--- a/adapters/simulated_onu/main.go
+++ b/adapters/simulated_onu/main.go
@@ -91,7 +91,7 @@
 	}
 
 	// Register the core request handler
-	if err = a.setupRequestHandler(a.instanceId, a.iAdapter); err != nil {
+	if err = a.setupRequestHandler(a.instanceId, a.iAdapter, a.coreProxy); err != nil {
 		log.Fatal("error-setting-core-request-handler")
 	}
 
@@ -218,9 +218,9 @@
 	return sOLT, nil
 }
 
-func (a *adapter) setupRequestHandler(coreInstanceId string, iadapter adapters.IAdapter) error {
+func (a *adapter) setupRequestHandler(coreInstanceId string, iadapter adapters.IAdapter, coreProxy *com.CoreProxy) error {
 	log.Info("setting-request-handler")
-	requestProxy := com.NewRequestHandlerProxy(coreInstanceId, iadapter)
+	requestProxy := com.NewRequestHandlerProxy(coreInstanceId, iadapter, coreProxy)
 	if err := a.kip.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: a.config.Topic}, requestProxy); err != nil {
 		log.Errorw("adaptercore-request-handler-setup-failed", log.Fields{"error": err})
 		return err
diff --git a/common/core/northbound/grpc/default_api_handler.go b/common/core/northbound/grpc/default_api_handler.go
index 0cc76c5..30320a2 100644
--- a/common/core/northbound/grpc/default_api_handler.go
+++ b/common/core/northbound/grpc/default_api_handler.go
@@ -42,6 +42,11 @@
 	return nil, errors.New("UnImplemented")
 }
 
+func (handler *DefaultAPIHandler) UpdateMembership(ctx context.Context, membership *voltha.Membership) (*empty.Empty, error) {
+	log.Debugw("UpdateMembership-request", log.Fields{"membership": membership})
+	return nil, errors.New("UnImplemented")
+}
+
 func (handler *DefaultAPIHandler) GetVoltha(ctx context.Context, empty *empty.Empty) (*voltha.Voltha, error) {
 	log.Debug("GetVoltha-request")
 	return nil, errors.New("UnImplemented")
diff --git a/kafka/kafka_inter_container_library.go b/kafka/kafka_inter_container_library.go
index 099d0d8..c5e0772 100644
--- a/kafka/kafka_inter_container_library.go
+++ b/kafka/kafka_inter_container_library.go
@@ -43,6 +43,7 @@
 
 const (
 	TransactionKey = "transactionID"
+	FromTopic      = "fromTopic"
 )
 
 // requestHandlerChannel represents an interface associated with a channel.  Whenever, an event is
@@ -331,7 +332,7 @@
 	kp.defaultRequestHandlerInterface = handler
 	kp.addToTopicRequestHandlerChannelMap(topic.Name, &requestHandlerChannel{requesthandlerInterface: handler, ch: ch})
 	// Launch a go routine to receive and process kafka messages
-	go kp.waitForRequest(ch, topic, handler)
+	go kp.waitForMessages(ch, topic, handler)
 
 	return nil
 }
@@ -342,14 +343,14 @@
 	// Subscribe to receive messages for that topic
 	var ch <-chan *ic.InterContainerMessage
 	var err error
-	if ch, err = kp.kafkaClient.Subscribe(&topic, &KVArg{Key:Offset, Value:initialOffset}); err != nil {
+	if ch, err = kp.kafkaClient.Subscribe(&topic, &KVArg{Key: Offset, Value: initialOffset}); err != nil {
 		log.Errorw("failed-to-subscribe", log.Fields{"error": err, "topic": topic.Name})
 		return err
 	}
 	kp.addToTopicRequestHandlerChannelMap(topic.Name, &requestHandlerChannel{requesthandlerInterface: kp.defaultRequestHandlerInterface, ch: ch})
 
 	// Launch a go routine to receive and process kafka messages
-	go kp.waitForRequest(ch, topic, kp.defaultRequestHandlerInterface)
+	go kp.waitForMessages(ch, topic, kp.defaultRequestHandlerInterface)
 
 	return nil
 }
@@ -615,7 +616,21 @@
 	return append(currentArgs, protoArg)
 }
 
-func (kp *InterContainerProxy) handleRequest(msg *ic.InterContainerMessage, targetInterface interface{}) {
+func (kp *InterContainerProxy) addFromTopic(fromTopic string, currentArgs []*ic.Argument) []*ic.Argument {
+	var marshalledArg *any.Any
+	var err error
+	if marshalledArg, err = ptypes.MarshalAny(&ic.StrType{Val: fromTopic}); err != nil {
+		log.Warnw("cannot-add-transactionId", log.Fields{"error": err})
+		return currentArgs
+	}
+	protoArg := &ic.Argument{
+		Key:   FromTopic,
+		Value: marshalledArg,
+	}
+	return append(currentArgs, protoArg)
+}
+
+func (kp *InterContainerProxy) handleMessage(msg *ic.InterContainerMessage, targetInterface interface{}) {
 
 	// First extract the header to know whether this is a request - responses are handled by a different handler
 	if msg.Header.Type == ic.MessageType_REQUEST {
@@ -633,6 +648,11 @@
 			// Augment the requestBody with the message Id as it will be used in scenarios where cores
 			// are set in pairs and competing
 			requestBody.Args = kp.addTransactionId(msg.Header.Id, requestBody.Args)
+
+			// Augment the requestBody with the From topic name as it will be used in scenarios where a container
+			// needs to send an unsollicited message to the currently requested container
+			requestBody.Args = kp.addFromTopic(msg.Header.FromTopic, requestBody.Args)
+
 			out, err = CallFuncByName(targetInterface, requestBody.Rpc, requestBody.Args)
 			if err != nil {
 				log.Warn(err)
@@ -660,7 +680,7 @@
 						returnError = &ic.Error{Reason: "incorrect-error-returns"}
 						returnedValues = append(returnedValues, returnError)
 					}
-				} else if len(out) == 2 && reflect.ValueOf(out[0].Interface()).IsValid() && reflect.ValueOf(out[0].Interface()).IsNil()  {
+				} else if len(out) == 2 && reflect.ValueOf(out[0].Interface()).IsValid() && reflect.ValueOf(out[0].Interface()).IsNil() {
 					return // Ignore case - when core is in competing mode
 				} else { // Non-error case
 					success = true
@@ -688,15 +708,19 @@
 			// TODO: handle error response.
 			kp.kafkaClient.Send(icm, replyTopic, key)
 		}
-
+	} else if msg.Header.Type == ic.MessageType_RESPONSE {
+		log.Debugw("response-received", log.Fields{"msg": msg})
+		go kp.dispatchResponse(msg)
+	} else {
+		log.Warnw("unsupported-message-received", log.Fields{"msg": msg})
 	}
 }
 
-func (kp *InterContainerProxy) waitForRequest(ch <-chan *ic.InterContainerMessage, topic Topic, targetInterface interface{}) {
+func (kp *InterContainerProxy) waitForMessages(ch <-chan *ic.InterContainerMessage, topic Topic, targetInterface interface{}) {
 	//	Wait for messages
 	for msg := range ch {
 		//log.Debugw("request-received", log.Fields{"msg": msg, "topic": topic.Name, "target": targetInterface})
-		go kp.handleRequest(msg, targetInterface)
+		go kp.handleMessage(msg, targetInterface)
 	}
 }
 
@@ -710,33 +734,6 @@
 	kp.transactionIdToChannelMap[msg.Header.Id].ch <- msg
 }
 
-// waitForResponse listens for messages on the subscribedCh, ensure we get a response with the transaction ID,
-// and then dispatches to the consumers
-func (kp *InterContainerProxy) waitForResponseLoop(subscribedCh <-chan *ic.InterContainerMessage, topic *Topic) {
-	log.Debugw("starting-response-loop-for-topic", log.Fields{"topic": topic.Name})
-startloop:
-	for {
-		select {
-		case msg, ok := <-subscribedCh:
-			if !ok {
-				log.Debugw("channel-closed", log.Fields{"topic": topic.Name})
-				break startloop
-			}
-			log.Debugw("message-received", log.Fields{"msg": msg})
-			//log.Debugw("message-received", log.Fields{"msg": msg, "fromTopic": msg.Header.FromTopic})
-			if msg.Header.Type == ic.MessageType_RESPONSE {
-				go kp.dispatchResponse(msg)
-			}
-		case <-kp.doneCh:
-			log.Infow("received-exit-signal", log.Fields{"topic": topic.Name})
-			break startloop
-		}
-	}
-	//log.Infow("received-exit-signal-out-of-for-loop", log.Fields{"topic": topic.Name})
-	//	We got an exit signal.  Unsubscribe to the channel
-	//kp.kafkaClient.UnSubscribe(topic, subscribedCh)
-}
-
 // subscribeForResponse allows a caller to subscribe to a given topic when waiting for a response.
 // This method is built to prevent all subscribers to receive all messages as is the case of the Subscribe
 // API. There is one response channel waiting for kafka messages before dispatching the message to the
@@ -744,26 +741,6 @@
 func (kp *InterContainerProxy) subscribeForResponse(topic Topic, trnsId string) (chan *ic.InterContainerMessage, error) {
 	log.Debugw("subscribeForResponse", log.Fields{"topic": topic.Name, "trnsid": trnsId})
 
-	// First check whether we already have a channel listening for response on that topic.  If there is
-	// already one then it will be reused.  If not, it will be created.
-	if !kp.isTopicSubscribedForResponse(topic.Name) {
-		log.Debugw("not-subscribed-for-response", log.Fields{"topic": topic.Name, "trnsid": trnsId})
-		var subscribedCh <-chan *ic.InterContainerMessage
-		var err error
-		if subscribedCh, err = kp.kafkaClient.Subscribe(&topic); err != nil {
-			log.Debugw("subscribe-failure", log.Fields{"topic": topic.Name})
-			return nil, err
-		}
-		kp.setupTopicResponseChannelMap(topic.Name, subscribedCh)
-		go kp.waitForResponseLoop(subscribedCh, &topic)
-
-		//	Wait until topic is ready - it takes on average 300 ms for a topic to be created.  This is a one time
-		//	delay everything a device is created.
-		// TODO:  Implement a mechanism to determine when a topic is ready instead of relying on a timeout
-		//kp.kafkaClient.WaitForTopicToBeReady
-		time.Sleep(400 * time.Millisecond)
-	}
-
 	// Create a specific channel for this consumers.  We cannot use the channel from the kafkaclient as it will
 	// broadcast any message for this topic to all channels waiting on it.
 	ch := make(chan *ic.InterContainerMessage)
diff --git a/kafka/utils.go b/kafka/utils.go
index beac9f9..0cb9535 100644
--- a/kafka/utils.go
+++ b/kafka/utils.go
@@ -15,10 +15,7 @@
  */
 package kafka
 
-import (
-	"fmt"
-	"strings"
-)
+import "strings"
 
 const (
 	TopicSeparator = "_"
@@ -39,19 +36,7 @@
 	Value interface{}
 }
 
-//CreateSubTopic concatenate a list of arguments together using underscores.
-func CreateSubTopic(args ...string) Topic {
-	topic := ""
-	for index, arg := range args {
-		if index == 0 {
-			topic = arg
-		} else {
-			topic = fmt.Sprintf("%s%s%s", topic, TopicSeparator, arg)
-		}
-	}
-	return Topic{Name: topic}
-}
-
+// TODO:  Remove and provide better may to get the device id
 // GetDeviceIdFromTopic extract the deviceId from the topic name.  The topic name is formatted either as:
 //			<any string> or <any string>_<deviceId>.  The device Id is 24 characters long.
 func GetDeviceIdFromTopic(topic Topic) string {
diff --git a/protos/voltha.proto b/protos/voltha.proto
index fdff7f6..ef6eedf 100644
--- a/protos/voltha.proto
+++ b/protos/voltha.proto
@@ -137,6 +137,15 @@
     string voltha_id = 2;
 }
 
+// Identifies a membership group a Core belongs to
+message Membership {
+    //  Group name
+    string group_name = 1;
+
+    // Unique ID of a container within that group
+    string id = 2;
+}
+
 /*
  * Voltha APIs
  *
@@ -150,6 +159,14 @@
         };
     }
 
+    // Set the membership group of a Voltha Core
+    rpc UpdateMembership(Membership) returns(google.protobuf.Empty) {
+        option (google.api.http) = {
+            post: "/api/v1/membership"
+            body: "*"
+        };
+    }
+
     // Get high level information on the Voltha cluster
     rpc GetVoltha(google.protobuf.Empty) returns(Voltha) {
         option (google.api.http) = {
diff --git a/python/adapters/kafka/adapter_request_facade.py b/python/adapters/kafka/adapter_request_facade.py
index fccb049..68344fc 100644
--- a/python/adapters/kafka/adapter_request_facade.py
+++ b/python/adapters/kafka/adapter_request_facade.py
@@ -30,7 +30,7 @@
 from python.protos.openflow_13_pb2 import FlowChanges, FlowGroups, Flows, \
     FlowGroupChanges, ofp_packet_out
 from python.adapters.kafka.kafka_inter_container_library import IKafkaMessagingProxy, \
-    get_messaging_proxy, KAFKA_OFFSET_LATEST, KAFKA_OFFSET_EARLIEST
+    get_messaging_proxy, KAFKA_OFFSET_LATEST, KAFKA_OFFSET_EARLIEST, ARG_FROM_TOPIC
 
 log = structlog.get_logger()
 
@@ -56,8 +56,9 @@
     defined in
     """
 
-    def __init__(self, adapter):
+    def __init__(self, adapter, core_proxy):
         self.adapter = adapter
+        self.core_proxy = core_proxy
 
     @inlineCallbacks
     def start(self):
@@ -67,24 +68,32 @@
     def stop(self):
         log.debug('stopping')
 
-    @inlineCallbacks
-    def createKafkaDeviceTopic(self, deviceId):
-        log.debug("subscribing-to-topic", device_id=deviceId)
-        kafka_proxy = get_messaging_proxy()
-        device_topic = kafka_proxy.get_default_topic() + "_" + deviceId
-        # yield kafka_proxy.create_topic(topic=device_topic)
-        yield kafka_proxy.subscribe(topic=device_topic, group_id=device_topic, target_cls=self, offset=KAFKA_OFFSET_EARLIEST)
-        log.debug("subscribed-to-topic", topic=device_topic)
+    # @inlineCallbacks
+    # def createKafkaDeviceTopic(self, deviceId):
+    #     log.debug("subscribing-to-topic", device_id=deviceId)
+    #     kafka_proxy = get_messaging_proxy()
+    #     device_topic = kafka_proxy.get_default_topic() + "_" + deviceId
+    #     # yield kafka_proxy.create_topic(topic=device_topic)
+    #     yield kafka_proxy.subscribe(topic=device_topic, group_id=device_topic, target_cls=self, offset=KAFKA_OFFSET_EARLIEST)
+    #     log.debug("subscribed-to-topic", topic=device_topic)
 
-    def adopt_device(self, device):
+    def adopt_device(self, device, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
 
-            # Start the creation of a device specific topic to handle all
-            # subsequent requests from the Core. This adapter instance will
-            # handle all requests for that device.
-            reactor.callLater(0, self.createKafkaDeviceTopic, d.id)
+            # Update the core reference for that device as it will be used
+            # by the adapter to send async messages to the Core.
+            if ARG_FROM_TOPIC in kwargs:
+                t = StrType()
+                kwargs[ARG_FROM_TOPIC].Unpack(t)
+                # Update the core reference for that device
+                self.core_proxy.update_device_core_reference(d.id, t.val)
+
+            # # Start the creation of a device specific topic to handle all
+            # # subsequent requests from the Core. This adapter instance will
+            # # handle all requests for that device.
+            # reactor.callLater(0, self.createKafkaDeviceTopic, d.id)
 
             result = self.adapter.adopt_device(d)
             # return True, self.adapter.adopt_device(d)
@@ -94,7 +103,7 @@
             return False, Error(code=ErrorCode.INVALID_PARAMETERS,
                                 reason="device-invalid")
 
-    def get_ofp_device_info(self, device):
+    def get_ofp_device_info(self, device, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -103,7 +112,7 @@
             return False, Error(code=ErrorCode.INVALID_PARAMETERS,
                                 reason="device-invalid")
 
-    def get_ofp_port_info(self, device, port_no):
+    def get_ofp_port_info(self, device, port_no, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -119,13 +128,13 @@
 
         return True, self.adapter.get_ofp_port_info(d, p.val)
 
-    def reconcile_device(self, device):
+    def reconcile_device(self, device, **kwargs):
         return self.adapter.reconcile_device(device)
 
-    def abandon_device(self, device):
+    def abandon_device(self, device, **kwargs):
         return self.adapter.abandon_device(device)
 
-    def disable_device(self, device):
+    def disable_device(self, device, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -134,7 +143,7 @@
             return False, Error(code=ErrorCode.INVALID_PARAMETERS,
                                 reason="device-invalid")
 
-    def reenable_device(self, device):
+    def reenable_device(self, device, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -143,7 +152,7 @@
             return False, Error(code=ErrorCode.INVALID_PARAMETERS,
                                 reason="device-invalid")
 
-    def reboot_device(self, device):
+    def reboot_device(self, device, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -152,7 +161,7 @@
             return False, Error(code=ErrorCode.INVALID_PARAMETERS,
                                 reason="device-invalid")
 
-    def download_image(self, device, request):
+    def download_image(self, device, request, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -168,7 +177,7 @@
 
         return True, self.adapter.download_image(device, request)
 
-    def get_image_download_status(self, device, request):
+    def get_image_download_status(self, device, request, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -184,7 +193,7 @@
 
         return True, self.adapter.get_image_download_status(device, request)
 
-    def cancel_image_download(self, device, request):
+    def cancel_image_download(self, device, request, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -200,7 +209,7 @@
 
         return True, self.adapter.cancel_image_download(device, request)
 
-    def activate_image_update(self, device, request):
+    def activate_image_update(self, device, request, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -216,7 +225,7 @@
 
         return True, self.adapter.activate_image_update(device, request)
 
-    def revert_image_update(self, device, request):
+    def revert_image_update(self, device, request, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -233,10 +242,10 @@
         return True, self.adapter.revert_image_update(device, request)
 
 
-    def self_test(self, device):
+    def self_test(self, device, **kwargs):
         return self.adapter.self_test_device(device)
 
-    def delete_device(self, device):
+    def delete_device(self, device, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -254,10 +263,10 @@
             return False, Error(code=ErrorCode.INVALID_PARAMETERS,
                                 reason="device-invalid")
 
-    def get_device_details(self, device):
+    def get_device_details(self, device, **kwargs):
         return self.adapter.get_device_details(device)
 
-    def update_flows_bulk(self, device, flows, groups):
+    def update_flows_bulk(self, device, flows, groups, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -274,7 +283,7 @@
 
         return (True, self.adapter.update_flows_bulk(d, f, g))
 
-    def update_flows_incrementally(self, device, flow_changes, group_changes):
+    def update_flows_incrementally(self, device, flow_changes, group_changes, **kwargs):
         d = Device()
         if device:
             device.Unpack(d)
@@ -291,13 +300,13 @@
 
         return (True, self.adapter.update_flows_incrementally(d, f, g))
 
-    def suppress_alarm(self, filter):
+    def suppress_alarm(self, filter, **kwargs):
         return self.adapter.suppress_alarm(filter)
 
-    def unsuppress_alarm(self, filter):
+    def unsuppress_alarm(self, filter, **kwargs):
         return self.adapter.unsuppress_alarm(filter)
 
-    def process_inter_adapter_message(self, msg):
+    def process_inter_adapter_message(self, msg, **kwargs):
         m = InterAdapterMessage()
         if msg:
             msg.Unpack(m)
@@ -308,7 +317,7 @@
         return (True, self.adapter.process_inter_adapter_message(m))
 
 
-    def receive_packet_out(self, deviceId, outPort, packet):
+    def receive_packet_out(self, deviceId, outPort, packet, **kwargs):
         try:
             d_id = StrType()
             if deviceId:
diff --git a/python/adapters/kafka/container_proxy.py b/python/adapters/kafka/container_proxy.py
index efcf558..bd873d8 100644
--- a/python/adapters/kafka/container_proxy.py
+++ b/python/adapters/kafka/container_proxy.py
@@ -38,10 +38,10 @@
 @implementer(IComponent)
 class ContainerProxy(object):
 
-    def __init__(self, kafka_proxy, core_topic, my_listening_topic):
+    def __init__(self, kafka_proxy, remote_topic, my_listening_topic):
         self.kafka_proxy = kafka_proxy
         self.listening_topic = my_listening_topic
-        self.core_topic = core_topic
+        self.remote_topic = remote_topic
         self.default_timeout = 3
 
     def start(self):
@@ -92,7 +92,7 @@
                           to_topic=to_topic,
                           reply_topic=reply_topic)
                 if to_topic is None:
-                    to_topic = self.core_topic
+                    to_topic = self.remote_topic
                 if reply_topic is None:
                     reply_topic = self.listening_topic
                 result = yield self.kafka_proxy.send_request(rpc=rpc,
@@ -130,4 +130,4 @@
                     raise e
                 retry += 1
                 if retry == max_retry:
-                    to_topic = self.core_topic
+                    to_topic = self.remote_topic
diff --git a/python/adapters/kafka/core_proxy.py b/python/adapters/kafka/core_proxy.py
index b897188..8d252a3 100644
--- a/python/adapters/kafka/core_proxy.py
+++ b/python/adapters/kafka/core_proxy.py
@@ -35,9 +35,27 @@
 
 class CoreProxy(ContainerProxy):
 
-    def __init__(self, kafka_proxy, core_topic, my_listening_topic):
-        super(CoreProxy, self).__init__(kafka_proxy, core_topic,
+    def __init__(self, kafka_proxy, default_core_topic, my_listening_topic):
+        super(CoreProxy, self).__init__(kafka_proxy, default_core_topic,
                                         my_listening_topic)
+        self.core_default_topic = default_core_topic
+        self.deviceId_to_core_map = dict()
+
+    def update_device_core_reference(self, device_id, core_topic):
+        log.debug("update_device_core_reference")
+        self.deviceId_to_core_map[device_id] = core_topic
+
+    def delete_device_core_reference(self, device_id, core_topic):
+        log.debug("delete_device_core_reference")
+        del self.deviceId_to_core_map[device_id]
+
+    def get_adapter_topic(self, **kwargs):
+        return self.listening_topic
+
+    def get_core_topic(self, device_id):
+        if device_id in self.deviceId_to_core_map:
+            return self.deviceId_to_core_map[device_id]
+        return self.core_default_topic
 
     @ContainerProxy.wrap_request(CoreInstance)
     @inlineCallbacks
@@ -62,8 +80,11 @@
         # Once we have a device being managed, all communications between the
         # the adapter and the core occurs over a topic associated with that
         # device
-        to_topic = createSubTopic(self.core_topic, device_id)
-        reply_topic = createSubTopic(self.listening_topic, device_id)
+        to_topic = self.get_core_topic(device_id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, device_id)
+        # reply_topic = createSubTopic(self.listening_topic, device_id)
         res = yield self.invoke(rpc="GetDevice",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -82,8 +103,11 @@
         id.id = device_id
         p_type = IntType()
         p_type.val = port_type
-        to_topic = createSubTopic(self.core_topic, device_id)
-        reply_topic = createSubTopic(self.listening_topic, device_id)
+        to_topic = self.get_core_topic(device_id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, device_id)
+        # reply_topic = createSubTopic(self.listening_topic, device_id)
         res = yield self.invoke(rpc="GetPorts",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -132,8 +156,11 @@
         cdt.val = child_device_type
         channel = IntType()
         channel.val = channel_id
-        to_topic = createSubTopic(self.core_topic, parent_device_id)
-        reply_topic = createSubTopic(self.listening_topic, parent_device_id)
+        to_topic = self.get_core_topic(parent_device_id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, parent_device_id)
+        # reply_topic = createSubTopic(self.listening_topic, parent_device_id)
         args = self._to_proto(**kw)
         res = yield self.invoke(rpc="ChildDeviceDetected",
                                 to_topic=to_topic,
@@ -149,8 +176,11 @@
     @inlineCallbacks
     def device_update(self, device):
         log.debug("device_update")
-        to_topic = createSubTopic(self.core_topic, device.id)
-        reply_topic = createSubTopic(self.listening_topic, device.id)
+        to_topic = self.get_core_topic(device.id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, device.id)
+        # reply_topic = createSubTopic(self.listening_topic, device.id)
         res = yield self.invoke(rpc="DeviceUpdate",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -178,8 +208,11 @@
         else:
             c_status.val = -1
 
-        to_topic = createSubTopic(self.core_topic, device_id)
-        reply_topic = createSubTopic(self.listening_topic, device_id)
+        to_topic = self.get_core_topic(device_id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, device_id)
+        #     reply_topic = createSubTopic(self.listening_topic, device_id)
         res = yield self.invoke(rpc="DeviceStateUpdate",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -206,8 +239,11 @@
         else:
             c_status.val = -1
 
-        to_topic = createSubTopic(self.core_topic, device_id)
-        reply_topic = createSubTopic(self.listening_topic, device_id)
+        to_topic = self.get_core_topic(device_id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, device_id)
+        # reply_topic = createSubTopic(self.listening_topic, device_id)
         res = yield self.invoke(rpc="ChildrenStateUpdate",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -232,8 +268,11 @@
         o_status = IntType()
         o_status.val = oper_status
 
-        to_topic = createSubTopic(self.core_topic, device_id)
-        reply_topic = createSubTopic(self.listening_topic, device_id)
+        to_topic = self.get_core_topic(device_id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, device_id)
+        # reply_topic = createSubTopic(self.listening_topic, device_id)
         res = yield self.invoke(rpc="PortStateUpdate",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -262,8 +301,11 @@
         else:
             c_status.val = -1
 
-        to_topic = createSubTopic(self.core_topic, parent_device_id)
-        reply_topic = createSubTopic(self.listening_topic, parent_device_id)
+        to_topic = self.get_core_topic(parent_device_id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, parent_device_id)
+        # reply_topic = createSubTopic(self.listening_topic, parent_device_id)
         res = yield self.invoke(rpc="child_devices_state_update",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -281,8 +323,11 @@
         log.debug("device_pm_config_update")
         b = BoolType()
         b.val = init
-        to_topic = createSubTopic(self.core_topic, device_pm_config.id)
-        reply_topic = createSubTopic(self.listening_topic, device_pm_config.id)
+        to_topic = self.get_core_topic(device_pm_config.id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, device_pm_config.id)
+        # reply_topic = createSubTopic(self.listening_topic, device_pm_config.id)
         res = yield self.invoke(rpc="DevicePMConfigUpdate",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -296,8 +341,11 @@
         log.debug("port_created")
         proto_id = ID()
         proto_id.id = device_id
-        to_topic = createSubTopic(self.core_topic, device_id)
-        reply_topic = createSubTopic(self.listening_topic, device_id)
+        to_topic = self.get_core_topic(device_id)
+        reply_topic = self.get_adapter_topic()
+
+        # to_topic = createSubTopic(self.core_topic, device_id)
+        # reply_topic = createSubTopic(self.listening_topic, device_id)
         res = yield self.invoke(rpc="PortCreated",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
@@ -333,8 +381,10 @@
         p.val = port
         pac = Packet()
         pac.payload = packet
-        to_topic = createSubTopic(self.core_topic, device_id)
-        reply_topic = createSubTopic(self.listening_topic, device_id)
+        to_topic = self.get_core_topic(device_id)
+        reply_topic = self.get_adapter_topic()
+        # to_topic = createSubTopic(self.core_topic, device_id)
+        # reply_topic = createSubTopic(self.listening_topic, device_id)
         res = yield self.invoke(rpc="PacketIn",
                                 to_topic=to_topic,
                                 reply_topic=reply_topic,
diff --git a/python/adapters/kafka/kafka_inter_container_library.py b/python/adapters/kafka/kafka_inter_container_library.py
index 5cad2e8..12a5dbe 100644
--- a/python/adapters/kafka/kafka_inter_container_library.py
+++ b/python/adapters/kafka/kafka_inter_container_library.py
@@ -28,12 +28,13 @@
 from kafka_proxy import KafkaProxy, get_kafka_proxy
 from python.protos.inter_container_pb2 import MessageType, Argument, \
     InterContainerRequestBody, InterContainerMessage, Header, \
-    InterContainerResponseBody
+    InterContainerResponseBody, StrType
 
 log = structlog.get_logger()
 
 KAFKA_OFFSET_LATEST = 'latest'
 KAFKA_OFFSET_EARLIEST = 'earliest'
+ARG_FROM_TOPIC = 'fromTopic'
 
 
 class KafkaMessagingError(BaseException):
@@ -406,6 +407,13 @@
         from Kafka.
         """
 
+        def _augment_args_with_FromTopic(args, from_topic):
+            arg = Argument(key=ARG_FROM_TOPIC)
+            t = StrType(val=from_topic)
+            arg.value.Pack(t)
+            args.extend([arg])
+            return args
+
         def _toDict(args):
             """
             Convert a repeatable Argument type into a python dictionary
@@ -451,12 +459,15 @@
                     log.debug("unsupported-msg", msg_type=type(message.body))
                     return
                 if targetted_topic in self.topic_target_cls_map:
-                    if msg_body.args:
+                    # Augment the request arguments with the from_topic
+                    augmented_args = _augment_args_with_FromTopic(msg_body.args,
+                                                        msg_body.reply_to_topic)
+                    if augmented_args:
                         log.debug("message-body-args-present", body=msg_body)
                         (status, res) = yield getattr(
                             self.topic_target_cls_map[targetted_topic],
                             self._to_string(msg_body.rpc))(
-                            **_toDict(msg_body.args))
+                            **_toDict(augmented_args))
                     else:
                         log.debug("message-body-args-absent", body=msg_body,
                                   rpc=msg_body.rpc)
diff --git a/python/adapters/ponsim_olt/main.py b/python/adapters/ponsim_olt/main.py
index 86b6be0..e3c5b7f 100755
--- a/python/adapters/ponsim_olt/main.py
+++ b/python/adapters/ponsim_olt/main.py
@@ -359,7 +359,7 @@
 
             self.core_proxy = CoreProxy(
                 kafka_proxy=None,
-                core_topic=self.core_topic,
+                default_core_topic=self.core_topic,
                 my_listening_topic=self.listening_topic)
 
             self.adapter_proxy = AdapterProxy(
@@ -371,7 +371,8 @@
                                             adapter_proxy=self.adapter_proxy,
                                             config=config)
 
-            ponsim_request_handler = AdapterRequestFacade(adapter=self.adapter)
+            ponsim_request_handler = AdapterRequestFacade(adapter=self.adapter,
+                                                    core_proxy=self.core_proxy)
 
             yield registry.register(
                 'kafka_adapter_proxy',
diff --git a/python/adapters/ponsim_olt/ponsim_olt.py b/python/adapters/ponsim_olt/ponsim_olt.py
index ff5ee00..4f5af65 100644
--- a/python/adapters/ponsim_olt/ponsim_olt.py
+++ b/python/adapters/ponsim_olt/ponsim_olt.py
@@ -315,6 +315,7 @@
                 hw_desc='ponsim pon',
                 sw_desc='ponsim pon',
                 serial_num=device.serial_number,
+                mfr_desc="VOLTHA Project",
                 dp_desc='n/a'
             ),
             switch_features=ofp_switch_features(
diff --git a/python/adapters/ponsim_onu/main.py b/python/adapters/ponsim_onu/main.py
index c4c12db..79320c2 100755
--- a/python/adapters/ponsim_onu/main.py
+++ b/python/adapters/ponsim_onu/main.py
@@ -358,7 +358,7 @@
 
             self.core_proxy = CoreProxy(
                 kafka_proxy=None,
-                core_topic=self.core_topic,
+                default_core_topic=self.core_topic,
                 my_listening_topic=self.listening_topic)
 
             self.adapter_proxy = AdapterProxy(
@@ -370,7 +370,7 @@
                 core_proxy=self.core_proxy, adapter_proxy=self.adapter_proxy,
                 config=config)
             ponsim_request_handler = AdapterRequestFacade(
-                adapter=self.adapter)
+                adapter=self.adapter, core_proxy=self.core_proxy)
 
             yield registry.register(
                 'kafka_adapter_proxy',
diff --git a/rw_core/core/adapter_proxy.go b/rw_core/core/adapter_proxy.go
index ed09c9d..b3ba10d 100644
--- a/rw_core/core/adapter_proxy.go
+++ b/rw_core/core/adapter_proxy.go
@@ -30,12 +30,15 @@
 
 type AdapterProxy struct {
 	TestMode     bool
+	deviceTopicRegistered bool
+	coreTopic *kafka.Topic
 	kafkaICProxy *kafka.InterContainerProxy
 }
 
 func NewAdapterProxy(kafkaProxy *kafka.InterContainerProxy) *AdapterProxy {
 	var proxy AdapterProxy
 	proxy.kafkaICProxy = kafkaProxy
+	proxy.deviceTopicRegistered = false
 	return &proxy
 }
 
@@ -55,22 +58,42 @@
 	}
 }
 
+func (ap *AdapterProxy) updateCoreTopic(coreTopic *kafka.Topic) {
+	ap.coreTopic = coreTopic
+}
+
+func (ap *AdapterProxy) getCoreTopic() kafka.Topic{
+	if ap.coreTopic != nil {
+		return *ap.coreTopic
+	}
+	return kafka.Topic{Name:ap.kafkaICProxy.DefaultTopic.Name}
+}
+
+func (ap *AdapterProxy) getAdapterTopic(adapterName string) kafka.Topic{
+	return kafka.Topic{Name: adapterName}
+}
+
+
 func (ap *AdapterProxy) AdoptDevice(ctx context.Context, device *voltha.Device) error {
 	log.Debugw("AdoptDevice", log.Fields{"device": device})
 	rpc := "adopt_device"
-	topic := kafka.Topic{Name: device.Adapter}
+	toTopic := ap.getAdapterTopic(device.Adapter)
+	//topic := kafka.Topic{Name: device.Adapter}
 	args := make([]*kafka.KVArg, 1)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
 		Value: device,
 	}
 	// Use a device topic for the response as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
-	if err := ap.kafkaICProxy.SubscribeWithDefaultRequestHandler(replyToTopic, kafka.OffsetOldest); err != nil {
-		log.Errorw("Unable-to-subscribe-new-topic", log.Fields{"topic": replyToTopic, "error": err})
-		return err
-	}
-	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &topic, &replyToTopic, true, args...)
+	replyToTopic := ap.getCoreTopic()
+	//if !ap.deviceTopicRegistered {
+	//	if err := ap.kafkaICProxy.SubscribeWithDefaultRequestHandler(replyToTopic, kafka.OffsetOldest); err != nil {
+	//		log.Errorw("Unable-to-subscribe-new-topic", log.Fields{"topic": replyToTopic, "error": err})
+	//		return err
+	//	}
+	//}
+	ap.deviceTopicRegistered = true
+	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("AdoptDevice-response", log.Fields{"replyTopic": replyToTopic, "deviceid": device.Id, "success": success})
 	return unPackResponse(rpc, device.Id, success, result)
 }
@@ -78,16 +101,19 @@
 func (ap *AdapterProxy) DisableDevice(ctx context.Context, device *voltha.Device) error {
 	log.Debugw("DisableDevice", log.Fields{"deviceId": device.Id})
 	rpc := "disable_device"
+	toTopic := ap.getAdapterTopic(device.Adapter)
+
 	// Use a device specific topic to send the request.  The adapter handling the device creates a device
 	// specific topic
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	//toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
 	args := make([]*kafka.KVArg, 1)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
 		Value: device,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	//replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(nil, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("DisableDevice-response", log.Fields{"deviceId": device.Id, "success": success})
 	return unPackResponse(rpc, device.Id, success, result)
@@ -96,14 +122,14 @@
 func (ap *AdapterProxy) ReEnableDevice(ctx context.Context, device *voltha.Device) error {
 	log.Debugw("ReEnableDevice", log.Fields{"deviceId": device.Id})
 	rpc := "reenable_device"
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 1)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
 		Value: device,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("ReEnableDevice-response", log.Fields{"deviceid": device.Id, "success": success})
 	return unPackResponse(rpc, device.Id, success, result)
@@ -112,14 +138,14 @@
 func (ap *AdapterProxy) RebootDevice(ctx context.Context, device *voltha.Device) error {
 	log.Debugw("RebootDevice", log.Fields{"deviceId": device.Id})
 	rpc := "reboot_device"
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 1)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
 		Value: device,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("RebootDevice-response", log.Fields{"deviceid": device.Id, "success": success})
 	return unPackResponse(rpc, device.Id, success, result)
@@ -128,14 +154,14 @@
 func (ap *AdapterProxy) DeleteDevice(ctx context.Context, device *voltha.Device) error {
 	log.Debugw("DeleteDevice", log.Fields{"deviceId": device.Id})
 	rpc := "delete_device"
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 1)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
 		Value: device,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("DeleteDevice-response", log.Fields{"deviceid": device.Id, "success": success})
 
@@ -150,14 +176,14 @@
 
 func (ap *AdapterProxy) GetOfpDeviceInfo(ctx context.Context, device *voltha.Device) (*ic.SwitchCapability, error) {
 	log.Debugw("GetOfpDeviceInfo", log.Fields{"deviceId": device.Id})
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 1)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
 		Value: device,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, "get_ofp_device_info", &toTopic, &replyToTopic, true, args...)
 	log.Debugw("GetOfpDeviceInfo-response", log.Fields{"deviceId": device.Id, "success": success, "result": result})
 	if success {
@@ -181,7 +207,7 @@
 
 func (ap *AdapterProxy) GetOfpPortInfo(ctx context.Context, device *voltha.Device, portNo uint32) (*ic.PortCapability, error) {
 	log.Debugw("GetOfpPortInfo", log.Fields{"deviceId": device.Id})
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 2)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
@@ -193,7 +219,7 @@
 		Value: pNo,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, "get_ofp_port_info", &toTopic, &replyToTopic, true, args...)
 	log.Debugw("GetOfpPortInfo-response", log.Fields{"deviceid": device.Id, "success": success})
 	if success {
@@ -250,7 +276,7 @@
 func (ap *AdapterProxy) DownloadImage(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) error {
 	log.Debugw("DownloadImage", log.Fields{"deviceId": device.Id, "image": download.Name})
 	rpc := "download_image"
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 2)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
@@ -261,7 +287,7 @@
 		Value: download,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("DownloadImage-response", log.Fields{"deviceId": device.Id, "success": success})
 
@@ -271,7 +297,7 @@
 func (ap *AdapterProxy) GetImageDownloadStatus(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) (*voltha.ImageDownload, error) {
 	log.Debugw("GetImageDownloadStatus", log.Fields{"deviceId": device.Id, "image": download.Name})
 	rpc := "get_image_download_status"
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 2)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
@@ -282,7 +308,7 @@
 		Value: download,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("GetImageDownloadStatus-response", log.Fields{"deviceId": device.Id, "success": success})
 
@@ -308,7 +334,7 @@
 func (ap *AdapterProxy) CancelImageDownload(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) error {
 	log.Debugw("CancelImageDownload", log.Fields{"deviceId": device.Id, "image": download.Name})
 	rpc := "cancel_image_download"
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 2)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
@@ -319,7 +345,7 @@
 		Value: download,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("CancelImageDownload-response", log.Fields{"deviceId": device.Id, "success": success})
 
@@ -329,7 +355,7 @@
 func (ap *AdapterProxy) ActivateImageUpdate(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) error {
 	log.Debugw("ActivateImageUpdate", log.Fields{"deviceId": device.Id, "image": download.Name})
 	rpc := "activate_image_update"
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 2)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
@@ -340,7 +366,7 @@
 		Value: download,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("ActivateImageUpdate-response", log.Fields{"deviceId": device.Id, "success": success})
 
@@ -350,7 +376,7 @@
 func (ap *AdapterProxy) RevertImageUpdate(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) error {
 	log.Debugw("RevertImageUpdate", log.Fields{"deviceId": device.Id, "image": download.Name})
 	rpc := "revert_image_update"
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	args := make([]*kafka.KVArg, 2)
 	args[0] = &kafka.KVArg{
 		Key:   "device",
@@ -361,7 +387,7 @@
 		Value: download,
 	}
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(ctx, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("RevertImageUpdate-response", log.Fields{"deviceId": device.Id, "success": success})
 
@@ -375,7 +401,7 @@
 
 func (ap *AdapterProxy) packetOut(deviceType string, deviceId string, outPort uint32, packet *openflow_13.OfpPacketOut) error {
 	log.Debugw("packetOut", log.Fields{"deviceId": deviceId})
-	toTopic := kafka.CreateSubTopic(deviceType, deviceId)
+	toTopic := ap.getAdapterTopic(deviceType)
 	rpc := "receive_packet_out"
 	dId := &ic.StrType{Val: deviceId}
 	args := make([]*kafka.KVArg, 3)
@@ -395,15 +421,15 @@
 
 	// TODO:  Do we need to wait for an ACK on a packet Out?
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, deviceId)
-	success, result := ap.kafkaICProxy.InvokeRPC(nil, rpc, &toTopic, &replyToTopic, false, args...)
+	replyToTopic := ap.getCoreTopic()
+	success, result := ap.kafkaICProxy.InvokeRPC(nil, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("packetOut", log.Fields{"deviceid": deviceId, "success": success})
 	return unPackResponse(rpc, deviceId, success, result)
 }
 
 func (ap *AdapterProxy) UpdateFlowsBulk(device *voltha.Device, flows *voltha.Flows, groups *voltha.FlowGroups) error {
 	log.Debugw("UpdateFlowsBulk", log.Fields{"deviceId": device.Id})
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	rpc := "update_flows_bulk"
 	args := make([]*kafka.KVArg, 3)
 	args[0] = &kafka.KVArg{
@@ -420,7 +446,7 @@
 	}
 
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(nil, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("UpdateFlowsBulk-response", log.Fields{"deviceid": device.Id, "success": success})
 	return unPackResponse(rpc, device.Id, success, result)
@@ -428,7 +454,7 @@
 
 func (ap *AdapterProxy) UpdateFlowsIncremental(device *voltha.Device, flowChanges *openflow_13.FlowChanges, groupChanges *openflow_13.FlowGroupChanges) error {
 	log.Debugw("UpdateFlowsIncremental", log.Fields{"deviceId": device.Id})
-	toTopic := kafka.CreateSubTopic(device.Adapter, device.Id)
+	toTopic := ap.getAdapterTopic(device.Adapter)
 	rpc := "update_flows_bulk"
 	args := make([]*kafka.KVArg, 3)
 	args[0] = &kafka.KVArg{
@@ -445,7 +471,7 @@
 	}
 
 	// Use a device specific topic as we are the only core handling requests for this device
-	replyToTopic := kafka.CreateSubTopic(ap.kafkaICProxy.DefaultTopic.Name, device.Id)
+	replyToTopic := ap.getCoreTopic()
 	success, result := ap.kafkaICProxy.InvokeRPC(nil, rpc, &toTopic, &replyToTopic, true, args...)
 	log.Debugw("UpdateFlowsIncremental-response", log.Fields{"deviceid": device.Id, "success": success})
 	return unPackResponse(rpc, device.Id, success, result)
diff --git a/rw_core/core/adapter_request_handler.go b/rw_core/core/adapter_request_handler.go
index 88b622e..1be0cff 100644
--- a/rw_core/core/adapter_request_handler.go
+++ b/rw_core/core/adapter_request_handler.go
@@ -81,7 +81,7 @@
 }
 
 func (rhp *AdapterRequestHandlerProxy) Register(args []*ic.Argument) (*voltha.CoreInstance, error) {
-	if len(args) != 3 {
+	if len(args) < 3 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
@@ -130,7 +130,7 @@
 }
 
 func (rhp *AdapterRequestHandlerProxy) GetDevice(args []*ic.Argument) (*voltha.Device, error) {
-	if len(args) != 2 {
+	if len(args) < 2 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
@@ -198,7 +198,7 @@
 }
 
 func (rhp *AdapterRequestHandlerProxy) DeviceUpdate(args []*ic.Argument) (*empty.Empty, error) {
-	if len(args) != 2 {
+	if len(args) < 2 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
@@ -292,7 +292,7 @@
 }
 
 func (rhp *AdapterRequestHandlerProxy) GetPorts(args []*ic.Argument) (*voltha.Ports, error) {
-	if len(args) != 3 {
+	if len(args) < 3 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
@@ -330,7 +330,7 @@
 }
 
 func (rhp *AdapterRequestHandlerProxy) GetChildDevices(args []*ic.Argument) (*voltha.Devices, error) {
-	if len(args) != 2 {
+	if len(args) < 2 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
@@ -623,7 +623,7 @@
 }
 
 func (rhp *AdapterRequestHandlerProxy) PortCreated(args []*ic.Argument) (*empty.Empty, error) {
-	if len(args) != 3 {
+	if len(args) < 3 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
@@ -674,7 +674,7 @@
 }
 
 func (rhp *AdapterRequestHandlerProxy) DevicePMConfigUpdate(args []*ic.Argument) (*empty.Empty, error) {
-	if len(args) != 3 {
+	if len(args) < 3 {
 		log.Warn("invalid-number-of-args", log.Fields{"args": args})
 		err := errors.New("invalid-number-of-args")
 		return nil, err
diff --git a/rw_core/core/core.go b/rw_core/core/core.go
index 838235d..d576a6e 100644
--- a/rw_core/core/core.go
+++ b/rw_core/core/core.go
@@ -17,6 +17,8 @@
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	grpcserver "github.com/opencord/voltha-go/common/grpc"
 	"github.com/opencord/voltha-go/common/log"
 	"github.com/opencord/voltha-go/db/kvstore"
@@ -43,6 +45,7 @@
 	exitChannel       chan int
 	kvClient          kvstore.Client
 	kafkaClient       kafka.Client
+	coreMembershipRegistered bool
 }
 
 func init() {
@@ -72,6 +75,7 @@
 	core.localDataRoot = model.NewRoot(&voltha.CoreInstance{}, &backend)
 	core.clusterDataProxy = core.clusterDataRoot.CreateProxy("/", false)
 	core.localDataProxy = core.localDataRoot.CreateProxy("/", false)
+	core.coreMembershipRegistered = false
 	return &core
 }
 
@@ -84,7 +88,8 @@
 	core.adapterMgr = newAdapterManager(core.clusterDataProxy, core.instanceId)
 	core.deviceMgr = newDeviceManager(core.kmp, core.clusterDataProxy, core.adapterMgr, core.instanceId)
 	core.logicalDeviceMgr = newLogicalDeviceManager(core.deviceMgr, core.kmp, core.clusterDataProxy)
-	if err := core.registerAdapterRequestHandler(ctx, core.instanceId, core.deviceMgr, core.logicalDeviceMgr, core.adapterMgr, core.clusterDataProxy, core.localDataProxy); err != nil {
+
+	if err := core.registerAdapterRequestHandlers(ctx, core.instanceId, core.deviceMgr, core.logicalDeviceMgr, core.adapterMgr, core.clusterDataProxy, core.localDataProxy); err != nil {
 		log.Fatal("Failure-registering-adapterRequestHandler")
 	}
 	go core.startDeviceManager(ctx)
@@ -113,7 +118,8 @@
 	core.grpcServer = grpcserver.NewGrpcServer(core.config.GrpcHost, core.config.GrpcPort, nil, false)
 	log.Info("grpc-server-created")
 
-	core.grpcNBIAPIHandler = NewAPIHandler(core.deviceMgr, core.logicalDeviceMgr, core.adapterMgr, core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
+	//core.grpcNBIAPIHandler = NewAPIHandler(core.deviceMgr, core.logicalDeviceMgr, core.adapterMgr, core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
+	core.grpcNBIAPIHandler = NewAPIHandler(core)
 	core.logicalDeviceMgr.setGrpcNbiHandler(core.grpcNBIAPIHandler)
 	//	Create a function to register the core GRPC service with the GRPC server
 	f := func(gs *grpc.Server) {
@@ -154,17 +160,57 @@
 	return nil
 }
 
-func (core *Core) registerAdapterRequestHandler(ctx context.Context, coreInstanceId string, dMgr *DeviceManager,
+//func (core *Core) registerAdapterRequestHandler(ctx context.Context, coreInstanceId string, dMgr *DeviceManager,
+//	ldMgr *LogicalDeviceManager, aMgr *AdapterManager, cdProxy *model.Proxy, ldProxy *model.Proxy,
+//	) error {
+//	requestProxy := NewAdapterRequestHandlerProxy(coreInstanceId, dMgr, ldMgr, aMgr, cdProxy, ldProxy,
+//		core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
+//	core.kmp.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: core.config.CoreTopic}, requestProxy)
+//
+//	log.Info("request-handlers")
+//	return nil
+//}
+
+func (core *Core) registerAdapterRequestHandlers(ctx context.Context, coreInstanceId string, dMgr *DeviceManager,
 	ldMgr *LogicalDeviceManager, aMgr *AdapterManager, cdProxy *model.Proxy, ldProxy *model.Proxy,
-	) error {
+) error {
 	requestProxy := NewAdapterRequestHandlerProxy(coreInstanceId, dMgr, ldMgr, aMgr, cdProxy, ldProxy,
 		core.config.InCompetingMode, core.config.LongRunningRequestTimeout, core.config.DefaultRequestTimeout)
-	core.kmp.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: core.config.CoreTopic}, requestProxy)
 
-	log.Info("request-handlers")
+	// Register the broadcast topic to handle any core-bound broadcast requests
+	if err := core.kmp.SubscribeWithRequestHandlerInterface(kafka.Topic{Name: core.config.CoreTopic}, requestProxy); err != nil {
+		log.Fatalw("Failed-registering-broadcast-handler", log.Fields{"topic": core.config.CoreTopic})
+		return err
+	}
+
+	log.Info("request-handler-registered")
 	return nil
 }
 
+
+func (core *Core) registerCoreTopic(ctx context.Context, coreTopicSuffix string) error {
+	// Sanity check - can only register once
+	if core.coreMembershipRegistered {
+		return errors.New("Can-only-set-once")
+	}
+
+	go func(coreTopicSuffix string) {
+		// Register the core-pair topic to handle core-bound requests aimed to the core pair
+		pTopic := kafka.Topic{Name: fmt.Sprintf("%s_%s", core.config.CoreTopic, coreTopicSuffix)}
+		// TODO: Set a retry here to ensure this subscription happens
+		if err := core.kmp.SubscribeWithDefaultRequestHandler(pTopic, kafka.OffsetNewest); err != nil {
+			log.Fatalw("Failed-registering-pair-handler", log.Fields{"topic": core.config.CoreTopic})
+		}
+		// Update the CoreTopic to use by the adapter proxy
+		core.deviceMgr.adapterProxy.updateCoreTopic(&pTopic)
+	}(coreTopicSuffix)
+
+	core.coreMembershipRegistered = true
+	log.Info("request-handlers-registered")
+	return nil
+}
+
+
 func (core *Core) startDeviceManager(ctx context.Context) {
 	// TODO: Interaction between the logicaldevicemanager and devicemanager should mostly occur via
 	// callbacks.  For now, until the model is ready, devicemanager will keep a reference to the
diff --git a/rw_core/core/grpc_nbi_api_handler.go b/rw_core/core/grpc_nbi_api_handler.go
index f72d615..46382c2 100644
--- a/rw_core/core/grpc_nbi_api_handler.go
+++ b/rw_core/core/grpc_nbi_api_handler.go
@@ -60,18 +60,34 @@
 	longRunningRequestTimeout int64
 	defaultRequestTimeout int64
 	da.DefaultAPIHandler
+	core *Core
 }
 
-func NewAPIHandler(deviceMgr *DeviceManager, lDeviceMgr *LogicalDeviceManager, adapterMgr *AdapterManager, inCompetingMode bool, longRunningRequestTimeout int64, defaultRequestTimeout int64 ) *APIHandler {
+//func NewAPIHandler(deviceMgr *DeviceManager, lDeviceMgr *LogicalDeviceManager, adapterMgr *AdapterManager, inCompetingMode bool, longRunningRequestTimeout int64, defaultRequestTimeout int64 ) *APIHandler {
+//	handler := &APIHandler{
+//		deviceMgr:        deviceMgr,
+//		logicalDeviceMgr: lDeviceMgr,
+//		adapterMgr:adapterMgr,
+//		coreInCompetingMode:inCompetingMode,
+//		longRunningRequestTimeout:longRunningRequestTimeout,
+//		defaultRequestTimeout:defaultRequestTimeout,
+//		// TODO: Figure out what the 'hint' parameter to queue.New does
+//		packetInQueue: queue.New(10),
+//	}
+//	return handler
+//}
+
+func NewAPIHandler(core *Core) *APIHandler {
 	handler := &APIHandler{
-		deviceMgr:        deviceMgr,
-		logicalDeviceMgr: lDeviceMgr,
-		adapterMgr:adapterMgr,
-		coreInCompetingMode:inCompetingMode,
-		longRunningRequestTimeout:longRunningRequestTimeout,
-		defaultRequestTimeout:defaultRequestTimeout,
+		deviceMgr:        core.deviceMgr,
+		logicalDeviceMgr: core.logicalDeviceMgr,
+		adapterMgr: core.adapterMgr,
+		coreInCompetingMode: core.config.InCompetingMode,
+		longRunningRequestTimeout:core.config.LongRunningRequestTimeout,
+		defaultRequestTimeout:core.config.DefaultRequestTimeout,
 		// TODO: Figure out what the 'hint' parameter to queue.New does
 		packetInQueue: queue.New(10),
+		core: core,
 	}
 	return handler
 }
@@ -186,6 +202,16 @@
 	return out, nil
 }
 
+
+func (handler *APIHandler) UpdateMembership(ctx context.Context, membership *voltha.Membership) (*empty.Empty, error) {
+	log.Debugw("UpdateMembership-request", log.Fields{"membership": membership})
+	out := new(empty.Empty)
+	if err := handler.core.registerCoreTopic(ctx, membership.GroupName); err != nil {
+		return out, err
+	}
+	return out, nil
+}
+
 func (handler *APIHandler) EnableLogicalDevicePort(ctx context.Context, id *voltha.LogicalPortId) (*empty.Empty, error) {
 	log.Debugw("EnableLogicalDevicePort-request", log.Fields{"id": id, "test": common.TestModeKeys_api_test.String()})
 	if isTestMode(ctx) {
diff --git a/rw_core/core/logical_device_manager.go b/rw_core/core/logical_device_manager.go
index 1b2fb1e..e313b77 100644
--- a/rw_core/core/logical_device_manager.go
+++ b/rw_core/core/logical_device_manager.go
@@ -306,7 +306,7 @@
 }
 
 func (ldMgr *LogicalDeviceManager) addUNILogicalPort(ctx context.Context, childDevice *voltha.Device) error {
-	log.Debugw("AddUNILogicalPort", log.Fields{"deviceId": childDevice.Id})
+	log.Debugw("AddUNILogicalPort", log.Fields{"childDeviceId": childDevice.Id, "parentDeviceId": childDevice.ParentId})
 	// Sanity check
 	if childDevice.Root {
 		return errors.New("Device-root")
diff --git a/rw_core/main.go b/rw_core/main.go
index 2a29499..41d595b 100644
--- a/rw_core/main.go
+++ b/rw_core/main.go
@@ -77,7 +77,7 @@
 			kafka.NumPartitions(3),
 			kafka.ConsumerGroupName(instanceID),
 			kafka.ConsumerGroupPrefix(instanceID),
-			kafka.AutoCreateTopic(false),
+			kafka.AutoCreateTopic(true),
 			kafka.ProducerFlushFrequency(5),
 			kafka.ProducerRetryBackoff(time.Millisecond*30)), nil
 	}