diff --git a/python/adapters/kafka/adapter_request_facade.py b/python/adapters/kafka/adapter_request_facade.py
index 0efb811..de8ae0b 100644
--- a/python/adapters/kafka/adapter_request_facade.py
+++ b/python/adapters/kafka/adapter_request_facade.py
@@ -22,6 +22,7 @@
 from twisted.internet.defer import inlineCallbacks
 from zope.interface import implementer
 from twisted.internet import reactor
+
 from afkak.consumer import OFFSET_LATEST, OFFSET_EARLIEST
 from python.adapters.interface import IAdapterInterface
 from python.protos.inter_container_pb2 import IntType, InterAdapterMessage, StrType, Error, ErrorCode
@@ -29,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
+    get_messaging_proxy, KAFKA_OFFSET_LATEST, KAFKA_OFFSET_EARLIEST
 
 log = structlog.get_logger()
 
@@ -72,7 +73,7 @@
         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, target_cls=self, offset=OFFSET_EARLIEST)
+        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):
diff --git a/python/adapters/kafka/kafka_inter_container_library.py b/python/adapters/kafka/kafka_inter_container_library.py
index fbb0834..5cad2e8 100644
--- a/python/adapters/kafka/kafka_inter_container_library.py
+++ b/python/adapters/kafka/kafka_inter_container_library.py
@@ -18,8 +18,6 @@
 from uuid import uuid4
 
 import structlog
-from afkak.client import KafkaClient
-from afkak.consumer import OFFSET_LATEST, Consumer
 from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks, returnValue, Deferred, \
     DeferredQueue, gatherResults
@@ -34,6 +32,9 @@
 
 log = structlog.get_logger()
 
+KAFKA_OFFSET_LATEST = 'latest'
+KAFKA_OFFSET_EARLIEST = 'earliest'
+
 
 class KafkaMessagingError(BaseException):
     def __init__(self, error):
@@ -48,6 +49,7 @@
                  kafka_host_port,
                  kv_store,
                  default_topic,
+                 group_id_prefix,
                  target_cls):
         """
         Initialize the kafka proxy.  This is a singleton (may change to
@@ -67,15 +69,15 @@
         self.kafka_host_port = kafka_host_port
         self.kv_store = kv_store
         self.default_topic = default_topic
+        self.default_group_id = "_".join((group_id_prefix, default_topic))
         self.target_cls = target_cls
         self.topic_target_cls_map = {}
-        self.topic_consumer_map = {}
         self.topic_callback_map = {}
         self.subscribers = {}
-        self.kafka_client = None
         self.kafka_proxy = None
         self.transaction_id_deferred_map = {}
         self.received_msg_queue = DeferredQueue()
+        self.stopped = False
 
         self.init_time = 0
         self.init_received_time = 0
@@ -91,11 +93,7 @@
 
     def start(self):
         try:
-            # Create the kafka client
-            # assert self.kafka_host is not None
-            # assert self.kafka_port is not None
-            # kafka_host_port = ":".join((self.kafka_host, self.kafka_port))
-            self.kafka_client = KafkaClient(self.kafka_host_port)
+            log.debug("KafkaProxy-starting")
 
             # Get the kafka proxy instance.  If it does not exist then
             # create it
@@ -110,12 +108,17 @@
             # Start the queue to handle incoming messages
             reactor.callLater(0, self._received_message_processing_loop)
 
-            # Start listening for incoming messages
-            reactor.callLater(0, self.subscribe, self.default_topic,
-                              target_cls=self.target_cls)
+            # Subscribe using the default topic and default group id.  Whenever
+            # a message is received on that topic then teh target_cls will be
+            # invoked.
+            reactor.callLater(0, self.subscribe,
+                              topic=self.default_topic,
+                              target_cls=self.target_cls,
+                              group_id=self.default_group_id)
 
             # Setup the singleton instance
             IKafkaMessagingProxy._kafka_messaging_instance = self
+            log.debug("KafkaProxy-started")
         except Exception as e:
             log.exception("Failed-to-start-proxy", e=e)
 
@@ -126,37 +129,14 @@
         """
         log.debug("Stopping-messaging-proxy ...")
         try:
-            # Stop all the consumers
-            deferred_list = []
-            for key, values in self.topic_consumer_map.iteritems():
-                deferred_list.extend([c.stop() for c in values])
-
-            if not deferred_list:
-                d = gatherResults(deferred_list)
-                d.addCallback(lambda result: self.kafka_client.close())
+            # Stop the kafka proxy.  This will stop all the consumers
+            # and producers
+            self.stopped = True
+            self.kafka_proxy.stop()
             log.debug("Messaging-proxy-stopped.")
         except Exception as e:
             log.exception("Exception-when-stopping-messaging-proxy:", e=e)
 
-
-    @inlineCallbacks
-    def create_topic(self, topic):
-        yield self._wait_until_topic_is_ready(self.kafka_client, topic)
-
-    @inlineCallbacks
-    def _wait_until_topic_is_ready(self, client, topic):
-        e = True
-        while e:
-            yield client.load_metadata_for_topics(topic)
-            e = client.metadata_error_for_topic(topic)
-            if e:
-                log.debug("Topic-not-ready-retrying...", topic=topic)
-
-    def _clear_backoff(self):
-        if self.retries:
-            log.info('reconnected-to-consul', after_retries=self.retries)
-            self.retries = 0
-
     def get_target_cls(self):
         return self.target_cls
 
@@ -164,24 +144,13 @@
         return self.default_topic
 
     @inlineCallbacks
-    def _subscribe(self, topic, offset, callback=None, target_cls=None):
+    def _subscribe_group_consumer(self, group_id, topic, offset, callback=None,
+                                  target_cls=None):
         try:
             log.debug("subscribing-to-topic-start", topic=topic)
-            yield self._wait_until_topic_is_ready(self.kafka_client, topic)
-            partitions = self.kafka_client.topic_partitions[topic]
-            consumers = []
-
-            # First setup the generic callback - all received messages will
-            # go through that queue
-            if topic not in self.topic_consumer_map:
-                log.debug("topic-not-in-consumer-map", topic=topic)
-                consumers = [Consumer(self.kafka_client, topic, partition,
-                                      self._enqueue_received_message)
-                             for partition in partitions]
-                self.topic_consumer_map[topic] = consumers
-
-            log.debug("_subscribe", topic=topic,
-                      consumermap=self.topic_consumer_map)
+            yield self.kafka_proxy.subscribe(topic,
+                                             self._enqueue_received_group_message,
+                                             group_id, offset)
 
             if target_cls is not None and callback is None:
                 # Scenario #1
@@ -198,25 +167,6 @@
             else:
                 log.warn("invalid-parameters")
 
-            def cb_closed(result):
-                """
-                Called when a consumer cleanly stops.
-                """
-                log.debug("Consumers-cleanly-stopped")
-
-            def eb_failed(failure):
-                """
-                Called when a consumer fails due to an uncaught exception in the
-                processing callback or a network error on shutdown. In this case we
-                simply log the error.
-                """
-                log.warn("Consumers-failed", failure=failure)
-
-            for c in consumers:
-                c.start(offset).addCallbacks(cb_closed, eb_failed)
-
-            log.debug("subscribed-to-topic", topic=topic)
-
             returnValue(True)
         except Exception as e:
             log.exception("Exception-during-subscription", e=e)
@@ -224,7 +174,7 @@
 
     @inlineCallbacks
     def subscribe(self, topic, callback=None, target_cls=None,
-                  max_retry=3, offset=OFFSET_LATEST):
+                  max_retry=3, group_id=None, offset=KAFKA_OFFSET_LATEST):
         """
         Scenario 1:  invoked to subscribe to a specific topic with a
         target_cls to invoke when a message is received on that topic.  This
@@ -245,6 +195,8 @@
         :param max_retry:  the number of retries before reporting failure
         to subscribe.  This caters for scenario where the kafka topic is not
         ready.
+        :param group_id:  The ID of the group the consumer is subscribing to
+        :param offset: The topic offset on the kafka bus from where message consumption will start
         :return: True on success, False on failure
         """
         RETRY_BACKOFF = [0.05, 0.1, 0.2, 0.5, 1, 2, 5]
@@ -253,13 +205,20 @@
             wait_time = RETRY_BACKOFF[min(retries,
                                           len(RETRY_BACKOFF) - 1)]
             log.info(msg, retry_in=wait_time)
-            return asleep(wait_time)
+            return asleep.asleep(wait_time)
+
+        log.debug("subscribing", topic=topic, group_id=group_id,
+                  callback=callback, target=target_cls)
 
         retry = 0
         subscribed = False
+        if group_id is None:
+            group_id = self.default_group_id
         while not subscribed:
-            subscribed = yield self._subscribe(topic, callback=callback,
-                                               target_cls=target_cls, offset=offset)
+            subscribed = yield self._subscribe_group_consumer(group_id, topic,
+                                                              callback=callback,
+                                                              target_cls=target_cls,
+                                                              offset=offset)
             if subscribed:
                 returnValue(True)
             elif retry > max_retry:
@@ -268,51 +227,56 @@
                 _backoff("subscription-not-complete", retry)
                 retry += 1
 
-        # while not self._subscribe(topic, callback=callback,
-        #                           target_cls=target_cls):
-        #     if retry > max_retry:
-        #         return False
-        #     else:
-        #         _backoff("subscription-not-complete", retry)
-        #         retry += 1
-        # return True
-
-    def unsubscribe(self, topic):
+    def unsubscribe(self, topic, callback=None, target_cls=None):
         """
         Invoked when unsubscribing to a topic
-        :param topic: topic to unsubscibe from
+        :param topic: topic to unsubscribe from
+        :param callback:  the callback used when subscribing to the topic, if any
+        :param target_cls: the targert class used when subscribing to the topic, if any
         :return: None on success or Exception on failure
         """
         log.debug("Unsubscribing-to-topic", topic=topic)
 
-        def remove_topic(topic):
-            if topic in self.topic_consumer_map:
-                del self.topic_consumer_map[topic]
-
         try:
-            if topic in self.topic_consumer_map:
-                consumers = self.topic_consumer_map[topic]
-                d = gatherResults([c.stop() for c in consumers])
-                d.addCallback(remove_topic, topic)
-                log.debug("Unsubscribed-to-topic.", topic=topic)
-            else:
-                log.debug("Topic-does-not-exist.", topic=topic)
+            self.kafka_proxy.unsubscribe(topic,
+                                         self._enqueue_received_group_message)
+
+            if callback is None and target_cls is None:
+                log.error("both-call-and-target-cls-cannot-be-none",
+                          topic=topic)
+                raise KafkaMessagingError(
+                    error="both-call-and-target-cls-cannot-be-none")
+
+            if target_cls is not None and topic in self.topic_target_cls_map:
+                del self.topic_target_cls_map[topic]
+
+            if callback is not None and topic in self.topic_callback_map:
+                index = 0
+                for cb in self.topic_callback_map[topic]:
+                    if cb == callback:
+                        break
+                    index += 1
+                if index < len(self.topic_callback_map[topic]):
+                    self.topic_callback_map[topic].pop(index)
+
+                if len(self.topic_callback_map[topic]) == 0:
+                    del self.topic_callback_map[topic]
         except Exception as e:
-            log.exception("Exception-when-stopping-messaging-proxy:", e=e)
+            log.exception("Exception-when-unsubscribing-to-topic", topic=topic,
+                          e=e)
+            return e
 
     @inlineCallbacks
-    def _enqueue_received_message(self, reactor, message_list):
+    def _enqueue_received_group_message(self, msg):
         """
         Internal method to continuously queue all received messaged
         irrespective of topic
-        :param reactor: A requirement by the Twisted Python kafka library
-        :param message_list: Received list of messages
+        :param msg: Received message
         :return: None on success, Exception on failure
         """
         try:
-            for m in message_list:
-                log.debug("received-msg", msg=m)
-                yield self.received_msg_queue.put(m)
+            log.debug("received-msg", msg=msg)
+            yield self.received_msg_queue.put(msg)
         except Exception as e:
             log.exception("Failed-enqueueing-received-message", e=e)
 
@@ -327,6 +291,8 @@
             try:
                 message = yield self.received_msg_queue.get()
                 yield self._process_message(message)
+                if self.stopped:
+                    break
             except Exception as e:
                 log.exception("Failed-dequeueing-received-message", e=e)
 
@@ -457,16 +423,18 @@
         current_time = int(round(time.time() * 1000))
         # log.debug("Got Message", message=m)
         try:
-            val = m.message.value
+            val = m.value()
+            # val = m.message.value
             # print m.topic
 
             # Go over customized callbacks first
-            if m.topic in self.topic_callback_map:
-                for c in self.topic_callback_map[m.topic]:
+            m_topic = m.topic()
+            if m_topic in self.topic_callback_map:
+                for c in self.topic_callback_map[m_topic]:
                     yield c(val)
 
             #  Check whether we need to process request/response scenario
-            if m.topic not in self.topic_target_cls_map:
+            if m_topic not in self.topic_target_cls_map:
                 return
 
             # Process request/response scenario
@@ -506,7 +474,8 @@
                                 response.header.to_topic)
                             self._send_kafka_message(res_topic, response)
 
-                        log.debug("Response-sent", response=response.body, to_topic=res_topic)
+                        log.debug("Response-sent", response=response.body,
+                                  to_topic=res_topic)
             elif message.header.type == MessageType.Value("RESPONSE"):
                 trns_id = self._to_string(message.header.id)
                 if trns_id in self.transaction_id_deferred_map:
diff --git a/python/adapters/kafka/kafka_proxy.py b/python/adapters/kafka/kafka_proxy.py
index d596334..cefc590 100644
--- a/python/adapters/kafka/kafka_proxy.py
+++ b/python/adapters/kafka/kafka_proxy.py
@@ -14,18 +14,19 @@
 # limitations under the License.
 #
 
-from afkak.client import KafkaClient as _KafkaClient
-from afkak.common import (
-    PRODUCER_ACK_NOT_REQUIRED
-)
-from afkak.producer import Producer as _kafkaProducer
+
+from confluent_kafka import Producer as _kafkaProducer
 from structlog import get_logger
+from twisted.internet import reactor
 from twisted.internet.defer import inlineCallbacks, returnValue
+from twisted.internet.threads import deferToThread
 from zope.interface import implementer
 
 from python.common.utils.consulhelpers import get_endpoint_from_consul
-from python.common.utils.registry import IComponent
 from event_bus_publisher import EventBusPublisher
+from python.common.utils.registry import IComponent
+from confluent_kafka import Consumer, KafkaError
+import threading
 
 log = get_logger()
 
@@ -33,7 +34,11 @@
 @implementer(IComponent)
 class KafkaProxy(object):
     """
-    This is a singleton proxy kafka class to hide the kafka client details.
+    This is a singleton proxy kafka class to hide the kafka client details. This
+    proxy uses confluent-kafka-python as the kafka client. Since that client is
+    not a Twisted client then requests to that client are wrapped with
+    twisted.internet.threads.deferToThread to avoid any potential blocking of
+    the Twisted loop.
     """
     _kafka_instance = None
 
@@ -42,6 +47,7 @@
                  kafka_endpoint='localhost:9092',
                  ack_timeout=1000,
                  max_req_attempts=10,
+                 consumer_poll_timeout=10,
                  config={}):
 
         # return an exception if the object already exist
@@ -59,6 +65,10 @@
         self.event_bus_publisher = None
         self.stopping = False
         self.faulty = False
+        self.consumer_poll_timeout = consumer_poll_timeout
+        self.topic_consumer_map = {}
+        self.topic_callbacks_map = {}
+        self.topic_any_map_lock = threading.Lock()
         log.debug('initialized', endpoint=kafka_endpoint)
 
     @inlineCallbacks
@@ -77,6 +87,7 @@
     def stop(self):
         try:
             log.debug('stopping-kafka-proxy')
+            self.stopping = True
             try:
                 if self.kclient:
                     yield self.kclient.close()
@@ -84,16 +95,38 @@
                     log.debug('stopped-kclient-kafka-proxy')
             except Exception, e:
                 log.exception('failed-stopped-kclient-kafka-proxy', e=e)
-                pass
 
             try:
                 if self.kproducer:
-                    yield self.kproducer.stop()
+                    yield self.kproducer.flush()
                     self.kproducer = None
                     log.debug('stopped-kproducer-kafka-proxy')
             except Exception, e:
                 log.exception('failed-stopped-kproducer-kafka-proxy', e=e)
-                pass
+
+            # Stop all consumers
+            try:
+                self.topic_any_map_lock.acquire()
+                log.debug('stopping-consumers-kafka-proxy')
+                for _, c in self.topic_consumer_map.iteritems():
+                    yield deferToThread(c.close)
+                self.topic_consumer_map.clear()
+                self.topic_callbacks_map.clear()
+                log.debug('stopped-consumers-kafka-proxy')
+            except Exception, e:
+                log.exception('failed-stopped-consumers-kafka-proxy', e=e)
+            finally:
+                self.topic_any_map_lock.release()
+                log.debug('stopping-consumers-kafka-proxy-released-lock')
+
+            # try:
+            #    if self.event_bus_publisher:
+            #        yield self.event_bus_publisher.stop()
+            #        self.event_bus_publisher = None
+            #        log.debug('stopped-event-bus-publisher-kafka-proxy')
+            # except Exception, e:
+            #    log.debug('failed-stopped-event-bus-publisher-kafka-proxy')
+            #    pass
 
             log.debug('stopped-kafka-proxy')
 
@@ -105,8 +138,7 @@
             pass
 
     def _get_kafka_producer(self):
-        # PRODUCER_ACK_LOCAL_WRITE : server will wait till the data is written
-        #  to a local log before sending response
+
         try:
 
             if self.kafka_endpoint.startswith('@'):
@@ -124,39 +156,125 @@
                     return
             else:
                 _k_endpoint = self.kafka_endpoint
-
-            self.kclient = _KafkaClient(_k_endpoint)
-            self.kproducer = _kafkaProducer(self.kclient,
-                                            req_acks=PRODUCER_ACK_NOT_REQUIRED,
-                                            # req_acks=PRODUCER_ACK_LOCAL_WRITE,
-                                            # ack_timeout=self.ack_timeout,
-                                            # max_req_attempts=self.max_req_attempts)
-                                            )
+            self.kproducer = _kafkaProducer(
+                {'bootstrap.servers': _k_endpoint,
+                 }
+            )
+            pass
         except Exception, e:
             log.exception('failed-get-kafka-producer', e=e)
             return
 
+    @inlineCallbacks
+    def _wait_for_messages(self, consumer, topic):
+        while True:
+            try:
+                msg = yield deferToThread(consumer.poll,
+                                          self.consumer_poll_timeout)
 
-    # @inlineCallbacks
-    # def wait_until_topic_is_ready(self, topic):
-    #     #  Assumes "auto.create.topics.enable" is set in the broker
-    #     #  configuration
-    #     e = True
-    #     while e:
-    #         yield self.kclient.load_metadata_for_topics(topic)
-    #         e = self.kclient.metadata_error_for_topic(topic)
-    #         if e:
-    #             log.debug("Topic-not-ready-retrying...", topic=topic)
+                if self.stopping:
+                    log.debug("stop-request-recieved", topic=topic)
+                    break
 
+                if msg is None:
+                    continue
+                if msg.error():
+                    # This typically is received when there are no more messages
+                    # to read from kafka. Ignore.
+                    continue
+
+                # Invoke callbacks
+                for cb in self.topic_callbacks_map[topic]:
+                    yield cb(msg)
+            except Exception as e:
+                log.debug("exception-receiving-msg", topic=topic, e=e)
 
     @inlineCallbacks
-    def create_topic(self, topic):
-        # Assume auto create is enabled on the broker configuration
-        yield self.wait_until_topic_is_ready(topic)
+    def subscribe(self, topic, callback, groupId, offset='latest'):
+        """
+        subscribe allows a caller to subscribe to a given kafka topic.  This API
+        always create a group consumer.
+        :param topic - the topic to subscribe to
+        :param callback - the callback to invoke whenever a message is received
+        on that topic
+        :param groupId - the groupId for this consumer.  In the current
+        implementation there is a one-to-one mapping between a topic and a
+        groupId.  In other words, once a groupId is used for a given topic then
+        we won't be able to create another groupId for the same topic.
+        :param offset:  the kafka offset from where the consumer will start
+        consuming messages
+        """
+        try:
+            self.topic_any_map_lock.acquire()
+            if topic in self.topic_consumer_map:
+                # Just add the callback
+                if topic in self.topic_callbacks_map:
+                    self.topic_callbacks_map[topic].append(callback)
+                else:
+                    self.topic_callbacks_map[topic] = [callback]
+                return
 
+            # Create consumer for that topic
+            c = Consumer({
+                'bootstrap.servers': self.kafka_endpoint,
+                'group.id': groupId,
+                'auto.offset.reset': offset
+            })
+            yield deferToThread(c.subscribe, [topic])
+            # c.subscribe([topic])
+            self.topic_consumer_map[topic] = c
+            self.topic_callbacks_map[topic] = [callback]
+            # Start the consumer
+            reactor.callLater(0, self._wait_for_messages, c, topic)
+        except Exception, e:
+            log.exception("topic-subscription-error", e=e)
+        finally:
+            self.topic_any_map_lock.release()
 
     @inlineCallbacks
-    def send_message(self, topic, msg):
+    def unsubscribe(self, topic, callback):
+        """
+        Unsubscribe to a given topic.  Since there they be multiple callers
+        consuming from the same topic then to ensure only the relevant caller
+        gets unsubscribe then the callback is used as a differentiator.   The
+        kafka consumer will be closed when there are no callbacks required.
+        :param topic: topic to unsubscribe
+        :param callback: callback the caller used when subscribing to the topic.
+        If multiple callers have subscribed to a topic using the same callback
+        then the first callback on the list will be removed.
+        :return:None
+        """
+        try:
+            self.topic_any_map_lock.acquire()
+            log.debug("unsubscribing-to-topic", topic=topic)
+            if topic in self.topic_callbacks_map:
+                index = 0
+                for cb in self.topic_callbacks_map[topic]:
+                    if cb == callback:
+                        break
+                    index += 1
+                if index < len(self.topic_callbacks_map[topic]):
+                    self.topic_callbacks_map[topic].pop(index)
+
+                if len(self.topic_callbacks_map[topic]) == 0:
+                    # Stop the consumer
+                    if topic in self.topic_consumer_map:
+                        yield deferToThread(
+                            self.topic_consumer_map[topic].close)
+                        del self.topic_consumer_map[topic]
+                    del self.topic_callbacks_map[topic]
+                    log.debug("unsubscribed-to-topic", topic=topic)
+                else:
+                    log.debug("consumers-for-topic-still-exist", topic=topic,
+                              num=len(self.topic_callbacks_map[topic]))
+        except Exception, e:
+            log.exception("topic-unsubscription-error", e=e)
+        finally:
+            self.topic_any_map_lock.release()
+            log.debug("unsubscribing-to-topic-release-lock", topic=topic)
+
+    @inlineCallbacks
+    def send_message(self, topic, msg, key=None):
         assert topic is not None
         assert msg is not None
 
@@ -174,22 +292,23 @@
                                   endpoint=self.kafka_endpoint)
                         return
 
-                # log.debug('sending-kafka-msg', topic=topic, msg=msg)
+                log.debug('sending-kafka-msg', topic=topic, kafka_msg=msg)
                 msgs = [msg]
 
-                if self.kproducer and self.kclient and \
-                        self.event_bus_publisher and self.faulty is False:
-                    # log.debug('sending-kafka-msg-I-am-here0', time=int(round(time.time() * 1000)))
-
-                    yield self.kproducer.send_messages(topic, msgs=msgs)
-                    # self.kproducer.send_messages(topic, msgs=msgs)
-                    # log.debug('sent-kafka-msg', topic=topic, msg=msg)
+                if self.kproducer is not None and self.event_bus_publisher and self.faulty is False:
+                    d = deferToThread(self.kproducer.produce, topic, msg, key)
+                    yield d
+                    log.debug('sent-kafka-msg', topic=topic, kafka_msg=msg)
+                    # send a lightweight poll to avoid an exception after 100k messages.
+                    d1 = deferToThread(self.kproducer.poll, 0)
+                    yield d1
                 else:
                     return
 
         except Exception, e:
             self.faulty = True
-            log.error('failed-to-send-kafka-msg', topic=topic, msg=msg, e=e)
+            log.error('failed-to-send-kafka-msg', topic=topic, kafka_msg=msg,
+                      e=e)
 
             # set the kafka producer to None.  This is needed if the
             # kafka docker went down and comes back up with a different
diff --git a/python/adapters/ponsim_olt/main.py b/python/adapters/ponsim_olt/main.py
index 09b78fc..86b6be0 100755
--- a/python/adapters/ponsim_olt/main.py
+++ b/python/adapters/ponsim_olt/main.py
@@ -63,8 +63,8 @@
     core_topic=os.environ.get('CORE_TOPIC', 'rwcore'),
     interface=os.environ.get('INTERFACE', get_my_primary_interface()),
     instance_id=os.environ.get('INSTANCE_ID', os.environ.get('HOSTNAME', '1')),
-    kafka_adapter=os.environ.get('KAFKA_ADAPTER', '192.168.0.20:9092'),
-    kafka_cluster=os.environ.get('KAFKA_CLUSTER', '10.100.198.220:9092'),
+    kafka_adapter=os.environ.get('KAFKA_ADAPTER', '172.20.10.3:9092'),
+    kafka_cluster=os.environ.get('KAFKA_CLUSTER', '172.20.10.3:9092'),
     backend=os.environ.get('BACKEND', 'none'),
     retry_interval=os.environ.get('RETRY_INTERVAL', 2),
     heartbeat_topic=os.environ.get('HEARTBEAT_TOPIC', "adapters.heartbeat"),
@@ -380,7 +380,7 @@
                     # TODO: Add KV Store object reference
                     kv_store=self.args.backend,
                     default_topic=self.args.name,
-                    # Needs to assign a real class
+                    group_id_prefix=self.args.instance_id,
                     target_cls=ponsim_request_handler
                 )
             ).start()
diff --git a/python/adapters/ponsim_onu/main.py b/python/adapters/ponsim_onu/main.py
index d6418e9..c4c12db 100755
--- a/python/adapters/ponsim_onu/main.py
+++ b/python/adapters/ponsim_onu/main.py
@@ -379,6 +379,7 @@
                     # TODO: Add KV Store object reference
                     kv_store=self.args.backend,
                     default_topic=self.args.name,
+                    group_id_prefix=self.args.instance_id,
                     target_cls=ponsim_request_handler
                 )
             ).start()