diff --git a/kafka/client.go b/kafka/client.go
index 316a4a5..9588274 100644
--- a/kafka/client.go
+++ b/kafka/client.go
@@ -40,9 +40,9 @@
 	DefaultKafkaPort                = 9092
 	DefaultGroupName                = "voltha"
 	DefaultSleepOnError             = 1
-	DefaultProducerFlushFrequency   = 5
+	DefaultProducerFlushFrequency   = 1
 	DefaultProducerFlushMessages    = 1
-	DefaultProducerFlushMaxmessages = 5
+	DefaultProducerFlushMaxmessages = 1
 	DefaultProducerReturnSuccess    = true
 	DefaultProducerReturnErrors     = true
 	DefaultProducerRetryMax         = 3
diff --git a/kafka/kafka_inter_container_library.go b/kafka/kafka_inter_container_library.go
index 1229e7a..e5e9606 100644
--- a/kafka/kafka_inter_container_library.go
+++ b/kafka/kafka_inter_container_library.go
@@ -256,7 +256,7 @@
 	// specific key, hence ensuring a single partition is used to publish the request.  This ensures that the
 	// subscriber on that topic will receive the request in the order it was sent.  The key used is the deviceId.
 	//key := GetDeviceIdFromTopic(*toTopic)
-	log.Debugw("sending-msg", log.Fields{"rpc": rpc, "toTopic": toTopic, "replyTopic": responseTopic, "key": key})
+	log.Debugw("sending-msg", log.Fields{"rpc": rpc, "toTopic": toTopic, "replyTopic": responseTopic, "key": key, "xId": protoRequest.Header.Id})
 	kp.kafkaClient.Send(protoRequest, toTopic, key)
 
 	if waitForResponse {
@@ -370,8 +370,8 @@
 }
 
 func (kp *InterContainerProxy) isTopicSubscribedForResponse(topic string) bool {
-	kp.lockTopicResponseChannelMap.Lock()
-	defer kp.lockTopicResponseChannelMap.Unlock()
+	kp.lockTopicResponseChannelMap.RLock()
+	defer kp.lockTopicResponseChannelMap.RUnlock()
 	_, exist := kp.topicToResponseChannelMap[topic]
 	return exist
 }
@@ -710,10 +710,10 @@
 			kp.kafkaClient.Send(icm, replyTopic, key)
 		}
 	} else if msg.Header.Type == ic.MessageType_RESPONSE {
-		log.Debugw("response-received", log.Fields{"msg": msg})
+		log.Debugw("response-received", log.Fields{"msg-header": msg.Header})
 		go kp.dispatchResponse(msg)
 	} else {
-		log.Warnw("unsupported-message-received", log.Fields{"msg": msg})
+		log.Warnw("unsupported-message-received", log.Fields{"msg-header": msg.Header})
 	}
 }
 
@@ -726,8 +726,8 @@
 }
 
 func (kp *InterContainerProxy) dispatchResponse(msg *ic.InterContainerMessage) {
-	kp.lockTransactionIdToChannelMap.Lock()
-	defer kp.lockTransactionIdToChannelMap.Unlock()
+	kp.lockTransactionIdToChannelMap.RLock()
+	defer kp.lockTransactionIdToChannelMap.RUnlock()
 	if _, exist := kp.transactionIdToChannelMap[msg.Header.Id]; !exist {
 		log.Debugw("no-waiting-channel", log.Fields{"transaction": msg.Header.Id})
 		return
diff --git a/kafka/sarama_client.go b/kafka/sarama_client.go
index 698fefa..b0ce502 100644
--- a/kafka/sarama_client.go
+++ b/kafka/sarama_client.go
@@ -540,8 +540,8 @@
 }
 
 func (sc *SaramaClient) getConsumerChannel(topic *Topic) *consumerChannels {
-	sc.lockTopicToConsumerChannelMap.Lock()
-	defer sc.lockTopicToConsumerChannelMap.Unlock()
+	sc.lockTopicToConsumerChannelMap.RLock()
+	defer sc.lockTopicToConsumerChannelMap.RUnlock()
 
 	if consumerCh, exist := sc.topicToConsumerChannelMap[topic.Name]; exist {
 		return consumerCh
@@ -726,8 +726,8 @@
 // topic via the unique channel each subscriber received during subscription
 func (sc *SaramaClient) dispatchToConsumers(consumerCh *consumerChannels, protoMessage *ic.InterContainerMessage) {
 	// Need to go over all channels and publish messages to them - do we need to copy msg?
-	sc.lockTopicToConsumerChannelMap.Lock()
-	defer sc.lockTopicToConsumerChannelMap.Unlock()
+	sc.lockTopicToConsumerChannelMap.RLock()
+	defer sc.lockTopicToConsumerChannelMap.RUnlock()
 	for _, ch := range consumerCh.channels {
 		go func(c chan *ic.InterContainerMessage) {
 			c <- protoMessage