diff --git a/rw_core/core/adapter_proxy.go b/rw_core/core/adapter_proxy.go
index d3907bb..1e18ba4 100755
--- a/rw_core/core/adapter_proxy.go
+++ b/rw_core/core/adapter_proxy.go
@@ -30,14 +30,16 @@
 	deviceTopicRegistered bool
 	corePairTopic         string
 	kafkaICProxy          kafka.InterContainerProxy
+	endpointManager       kafka.EndpointManager
 }
 
 // NewAdapterProxy will return adapter proxy instance
-func NewAdapterProxy(kafkaProxy kafka.InterContainerProxy, corePairTopic string) *AdapterProxy {
+func NewAdapterProxy(kafkaProxy kafka.InterContainerProxy, corePairTopic string, endpointManager kafka.EndpointManager) *AdapterProxy {
 	return &AdapterProxy{
 		kafkaICProxy:          kafkaProxy,
 		corePairTopic:         corePairTopic,
 		deviceTopicRegistered: false,
+		endpointManager:       endpointManager,
 	}
 }
 
@@ -45,8 +47,14 @@
 	return kafka.Topic{Name: ap.corePairTopic}
 }
 
-func (ap *AdapterProxy) getAdapterTopic(adapterName string) kafka.Topic {
-	return kafka.Topic{Name: adapterName}
+func (ap *AdapterProxy) getAdapterTopic(deviceID string, adapterType string) (*kafka.Topic, error) {
+
+	endpoint, err := ap.endpointManager.GetEndpoint(deviceID, adapterType)
+	if err != nil {
+		return nil, err
+	}
+
+	return &kafka.Topic{Name: string(endpoint)}, nil
 }
 
 func (ap *AdapterProxy) sendRPC(ctx context.Context, rpc string, toTopic *kafka.Topic, replyToTopic *kafka.Topic,
@@ -69,167 +77,210 @@
 func (ap *AdapterProxy) adoptDevice(ctx context.Context, device *voltha.Device) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("adoptDevice", log.Fields{"device-id": device.Id})
 	rpc := "adopt_device"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 	}
 	replyToTopic := ap.getCoreTopic()
 	ap.deviceTopicRegistered = true
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	logger.Debugw("adoptDevice-send-request", log.Fields{"device-id": device.Id, "deviceType": device.Type, "serialNumber": device.SerialNumber})
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // disableDevice invokes disable device rpc
 func (ap *AdapterProxy) disableDevice(ctx context.Context, device *voltha.Device) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("disableDevice", log.Fields{"device-id": device.Id})
 	rpc := "disable_device"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // reEnableDevice invokes reenable device rpc
 func (ap *AdapterProxy) reEnableDevice(ctx context.Context, device *voltha.Device) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("reEnableDevice", log.Fields{"device-id": device.Id})
 	rpc := "reenable_device"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // rebootDevice invokes reboot device rpc
 func (ap *AdapterProxy) rebootDevice(ctx context.Context, device *voltha.Device) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("rebootDevice", log.Fields{"device-id": device.Id})
 	rpc := "reboot_device"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // deleteDevice invokes delete device rpc
 func (ap *AdapterProxy) deleteDevice(ctx context.Context, device *voltha.Device) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("deleteDevice", log.Fields{"device-id": device.Id})
 	rpc := "delete_device"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // getOfpDeviceInfo invokes get ofp device info rpc
 func (ap *AdapterProxy) getOfpDeviceInfo(ctx context.Context, device *voltha.Device) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("getOfpDeviceInfo", log.Fields{"device-id": device.Id})
 	rpc := "get_ofp_device_info"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // getOfpPortInfo invokes get ofp port info rpc
 func (ap *AdapterProxy) getOfpPortInfo(ctx context.Context, device *voltha.Device, portNo uint32) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("getOfpPortInfo", log.Fields{"device-id": device.Id, "port-no": portNo})
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 		{Key: "port_no", Value: &ic.IntType{Val: int64(portNo)}},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, "get_ofp_port_info", &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, "get_ofp_port_info", toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // reconcileDevice invokes reconcile device rpc
 func (ap *AdapterProxy) reconcileDevice(ctx context.Context, device *voltha.Device) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("reconcileDevice", log.Fields{"device-id": device.Id})
 	rpc := "reconcile_device"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // downloadImage invokes download image rpc
 func (ap *AdapterProxy) downloadImage(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("downloadImage", log.Fields{"device-id": device.Id, "image": download.Name})
 	rpc := "download_image"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 		{Key: "request", Value: download},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // getImageDownloadStatus invokes get image download status rpc
 func (ap *AdapterProxy) getImageDownloadStatus(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("getImageDownloadStatus", log.Fields{"device-id": device.Id, "image": download.Name})
 	rpc := "get_image_download_status"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 		{Key: "request", Value: download},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // cancelImageDownload invokes cancel image download rpc
 func (ap *AdapterProxy) cancelImageDownload(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("cancelImageDownload", log.Fields{"device-id": device.Id, "image": download.Name})
 	rpc := "cancel_image_download"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 		{Key: "request", Value: download},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // activateImageUpdate invokes activate image update rpc
 func (ap *AdapterProxy) activateImageUpdate(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("activateImageUpdate", log.Fields{"device-id": device.Id, "image": download.Name})
 	rpc := "activate_image_update"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 		{Key: "request", Value: download},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // revertImageUpdate invokes revert image update rpc
 func (ap *AdapterProxy) revertImageUpdate(ctx context.Context, device *voltha.Device, download *voltha.ImageDownload) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("revertImageUpdate", log.Fields{"device-id": device.Id, "image": download.Name})
 	rpc := "revert_image_update"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 		{Key: "request", Value: download},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 func (ap *AdapterProxy) packetOut(ctx context.Context, deviceType string, deviceID string, outPort uint32, packet *openflow_13.OfpPacketOut) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("packetOut", log.Fields{"device-id": deviceID, "device-type": deviceType, "out-port": outPort})
-	toTopic := ap.getAdapterTopic(deviceType)
+	toTopic, err := ap.getAdapterTopic(deviceID, deviceType)
+	if err != nil {
+		return nil, err
+	}
 	rpc := "receive_packet_out"
 	args := []*kafka.KVArg{
 		{Key: "deviceId", Value: &ic.StrType{Val: deviceID}},
@@ -237,13 +288,16 @@
 		{Key: "packet", Value: packet},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, deviceID, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, deviceID, args...)
 }
 
 // updateFlowsBulk invokes update flows bulk rpc
 func (ap *AdapterProxy) updateFlowsBulk(ctx context.Context, device *voltha.Device, flows *voltha.Flows, groups *voltha.FlowGroups, flowMetadata *voltha.FlowMetadata) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("updateFlowsBulk", log.Fields{"device-id": device.Id, "flow-count": len(flows.Items), "group-count": len(groups.Items), "flow-metadata": flowMetadata})
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	rpc := "update_flows_bulk"
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
@@ -252,7 +306,7 @@
 		{Key: "flow_metadata", Value: flowMetadata},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(context.TODO(), rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(context.TODO(), rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // updateFlowsIncremental invokes update flows incremental rpc
@@ -266,7 +320,10 @@
 			"group-to-delete-count": len(groupChanges.ToRemove.Items),
 			"group-to-update-count": len(groupChanges.ToUpdate.Items),
 		})
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	rpc := "update_flows_incrementally"
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
@@ -275,83 +332,101 @@
 		{Key: "flow_metadata", Value: flowMetadata},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(context.TODO(), rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(context.TODO(), rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // updatePmConfigs invokes update pm configs rpc
 func (ap *AdapterProxy) updatePmConfigs(ctx context.Context, device *voltha.Device, pmConfigs *voltha.PmConfigs) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("updatePmConfigs", log.Fields{"device-id": device.Id, "pm-configs-id": pmConfigs.Id})
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	rpc := "Update_pm_config"
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 		{Key: "pm_configs", Value: pmConfigs},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // simulateAlarm invokes simulate alarm rpc
 func (ap *AdapterProxy) simulateAlarm(ctx context.Context, device *voltha.Device, simulateReq *voltha.SimulateAlarmRequest) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("simulateAlarm", log.Fields{"device-id": device.Id, "simulate-req-id": simulateReq.Id})
 	rpc := "simulate_alarm"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "device", Value: device},
 		{Key: "request", Value: simulateReq},
 	}
 	replyToTopic := ap.getCoreTopic()
 	ap.deviceTopicRegistered = true
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 func (ap *AdapterProxy) disablePort(ctx context.Context, device *voltha.Device, port *voltha.Port) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("disablePort", log.Fields{"device-id": device.Id, "port-no": port.PortNo})
 	rpc := "disable_port"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "deviceId", Value: &ic.StrType{Val: device.Id}},
 		{Key: "port", Value: port},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 func (ap *AdapterProxy) enablePort(ctx context.Context, device *voltha.Device, port *voltha.Port) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("enablePort", log.Fields{"device-id": device.Id, "port-no": port.PortNo})
 	rpc := "enable_port"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
 		{Key: "deviceId", Value: &ic.StrType{Val: device.Id}},
 		{Key: "port", Value: port},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id, args...)
 }
 
 // childDeviceLost invokes child device_lost rpc
-func (ap *AdapterProxy) childDeviceLost(ctx context.Context, deviceType string, pDeviceID string, pPortNo uint32, onuID uint32) (chan *kafka.RpcResponse, error) {
-	logger.Debugw("childDeviceLost", log.Fields{"parent-device-id": pDeviceID, "parent-port-no": pPortNo, "onu-id": onuID})
+func (ap *AdapterProxy) childDeviceLost(ctx context.Context, deviceType string, deviceID string, pPortNo uint32, onuID uint32) (chan *kafka.RpcResponse, error) {
+	logger.Debugw("childDeviceLost", log.Fields{"device-id": deviceID, "parent-port-no": pPortNo, "onu-id": onuID})
 	rpc := "child_device_lost"
-	toTopic := ap.getAdapterTopic(deviceType)
+	toTopic, err := ap.getAdapterTopic(deviceID, deviceType)
+	if err != nil {
+		return nil, err
+	}
 	args := []*kafka.KVArg{
-		{Key: "pDeviceId", Value: &ic.StrType{Val: pDeviceID}},
+		{Key: "pDeviceId", Value: &ic.StrType{Val: deviceID}},
 		{Key: "pPortNo", Value: &ic.IntType{Val: int64(pPortNo)}},
 		{Key: "onuID", Value: &ic.IntType{Val: int64(onuID)}},
 	}
 	replyToTopic := ap.getCoreTopic()
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, pDeviceID, args...)
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, deviceID, args...)
 }
 
 func (ap *AdapterProxy) startOmciTest(ctx context.Context, device *voltha.Device, omcitestrequest *voltha.OmciTestRequest) (chan *kafka.RpcResponse, error) {
 	logger.Debugw("Omci_test_Request_adapter_proxy", log.Fields{"device": device, "omciTestRequest": omcitestrequest})
 	rpc := "start_omci_test"
-	toTopic := ap.getAdapterTopic(device.Adapter)
+	toTopic, err := ap.getAdapterTopic(device.Id, device.Adapter)
+	if err != nil {
+		return nil, err
+	}
 	// Use a device specific topic as we are the only core handling requests for this device
 	replyToTopic := ap.getCoreTopic()
 	// TODO: Perhaps this should have used omcitestrequest.uuid as the second argument rather
 	//   than including the whole request, which is (deviceid, uuid)
-	return ap.sendRPC(ctx, rpc, &toTopic, &replyToTopic, true, device.Id,
+	return ap.sendRPC(ctx, rpc, toTopic, &replyToTopic, true, device.Id,
 		&kafka.KVArg{Key: "device", Value: device},
 		&kafka.KVArg{Key: "omcitestrequest", Value: omcitestrequest})
 }