diff --git a/python/adapters/kafka/adapter_request_facade.py b/python/adapters/kafka/adapter_request_facade.py
index 978f57d..cf7afd8 100644
--- a/python/adapters/kafka/adapter_request_facade.py
+++ b/python/adapters/kafka/adapter_request_facade.py
@@ -18,11 +18,11 @@
 This facade handles kafka-formatted messages from the Core, extracts the kafka
 formatting and forwards the request to the concrete handler.
 """
-
+import structlog
 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.core_adapter_pb2 import IntType, InterAdapterMessage, StrType, Error, ErrorCode
 from python.protos.device_pb2 import Device
@@ -31,6 +31,7 @@
 from python.adapters.kafka.kafka_inter_container_library import IKafkaMessagingProxy, \
     get_messaging_proxy
 
+log = structlog.get_logger()
 
 class MacAddressError(BaseException):
     def __init__(self, error):
@@ -59,31 +60,34 @@
 
     @inlineCallbacks
     def start(self):
-        self.log.debug('starting')
+        log.debug('starting')
 
     @inlineCallbacks
     def stop(self):
-        self.log.debug('stopping')
+        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
-        kafka_proxy.subscribe(topic=device_topic, target_cls=self)
+        # yield kafka_proxy.create_topic(topic=device_topic)
+        yield kafka_proxy.subscribe(topic=device_topic, target_cls=self, offset=OFFSET_EARLIEST)
+        log.debug("subscribed-to-topic", topic=device_topic)
 
     def adopt_device(self, device):
         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)
+
             result = self.adapter.adopt_device(d)
             # return True, self.adapter.adopt_device(d)
 
-            # Before we return, create 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)
-
             return True, result
         else:
             return False, Error(code=ErrorCode.INVALID_PARAMETERS,
diff --git a/python/adapters/kafka/kafka_inter_container_library.py b/python/adapters/kafka/kafka_inter_container_library.py
index aaa0c3c..a3de4f1 100644
--- a/python/adapters/kafka/kafka_inter_container_library.py
+++ b/python/adapters/kafka/kafka_inter_container_library.py
@@ -138,6 +138,11 @@
         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
@@ -159,8 +164,9 @@
         return self.default_topic
 
     @inlineCallbacks
-    def _subscribe(self, topic, callback=None, target_cls=None):
+    def _subscribe(self, 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 = []
@@ -168,6 +174,7 @@
             # 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]
@@ -206,15 +213,18 @@
                 log.warn("Consumers-failed", failure=failure)
 
             for c in consumers:
-                c.start(OFFSET_LATEST).addCallbacks(cb_closed, eb_failed)
+                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)
             returnValue(False)
 
+    @inlineCallbacks
     def subscribe(self, topic, callback=None, target_cls=None,
-                  max_retry=3):
+                  max_retry=3, offset=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
@@ -246,14 +256,26 @@
             return asleep(wait_time)
 
         retry = 0
-        while not self._subscribe(topic, callback=callback,
-                                  target_cls=target_cls):
-            if retry > max_retry:
-                return False
+        subscribed = False
+        while not subscribed:
+            subscribed = yield self._subscribe(topic, callback=callback,
+                                               target_cls=target_cls, offset=offset)
+            if subscribed:
+                returnValue(True)
+            elif retry > max_retry:
+                returnValue(False)
             else:
                 _backoff("subscription-not-complete", retry)
                 retry += 1
-        return True
+
+        # 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):
         """
diff --git a/python/adapters/kafka/kafka_proxy.py b/python/adapters/kafka/kafka_proxy.py
index 6dcb10f..d596334 100644
--- a/python/adapters/kafka/kafka_proxy.py
+++ b/python/adapters/kafka/kafka_proxy.py
@@ -136,6 +136,25 @@
             log.exception('failed-get-kafka-producer', e=e)
             return
 
+
+    # @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)
+
+
+    @inlineCallbacks
+    def create_topic(self, topic):
+        # Assume auto create is enabled on the broker configuration
+        yield self.wait_until_topic_is_ready(topic)
+
+
     @inlineCallbacks
     def send_message(self, topic, msg):
         assert topic is not None