Stability fixes

README.rst

@@ -47,7 +47,8 @@ To install manually, first clone this here repo and:
 .. parsed-literal::
 
     cd kiel
-    python setup.py install
+    # optionally update version in __init__.py
+    pip install .
 
 
 Documentation

kiel/__init__.py

@@ -1,3 +1,3 @@
-version_info = (0, 9, 4)
+version_info = (0, 9, 4, 1)
 
-__version__ = ".".join(map(str, version_info)) + "-dev"
+__version__ = ".".join(map(str, version_info)) #+ "-dev"

kiel/clients/consumer.py

@@ -14,7 +14,7 @@
 
 
 log = logging.getLogger(__name__)
-
+# log.setLevel(logging.DEBUG) # uncomment for troubleshooting
 
 class BaseConsumer(Client):
     """
@@ -128,9 +128,11 @@ def consume(self, topic, start=None):
             )
 
         results = yield self.send(requests)
+#        log.debug(results)
         raise gen.Return([
             msg for messageset in results.values() for msg in messageset
-            if messageset
+            if msg # Skip empty messages
+            # BUG: troubleshoot why there are None values
         ])
 
     def handle_fetch_response(self, response):
@@ -195,6 +197,9 @@ def deserialize_messages(self, topic_name, partition):
                 continue
 
             messages.append(value)
-            self.offsets[topic_name][partition.partition_id] = offset + 1
+#            log.debug("offset %d => ", self.offsets[topic_name][partition.partition_id])
+            if offset>=0: # FIX: offset may have special value (e.g., def. END(-1))
+                self.offsets[topic_name][partition.partition_id] = offset + 1
+#            log.debug("%d", self.offsets[topic_name][partition.partition_id])
 
         return messages

kiel/clients/single.py

@@ -32,6 +32,7 @@ class SingleConsumer(BaseConsumer):
     BEGINNING = -2
     #: special offset api value for 'very latest offset'
     END = -1
+    #: For api ref. see https://github.com/kafka-dev/kafka/blob/master/core/src/main/scala/kafka/api/OffsetRequest.scala
 
     @property
     def allocation(self):
@@ -104,7 +105,10 @@ def handle_offset_response(self, response):
             code = partition.error_code
             if code == errors.no_error:
                 offset = partition.offsets[0]
-                self.offsets[topic][partition.partition_id] = offset
+#                log.debug("offset %d ==> ", self.offsets[topic][partition.partition_id])
+                if offset>=0:
+                    self.offsets[topic][partition.partition_id] = offset
+#                log.debug("%d", self.offsets[topic][partition.partition_id])
             elif code in errors.retriable:
                 self.heal_cluster = True
                 self.synced_offsets.discard(topic)

kiel/cluster.py

@@ -137,6 +137,7 @@ def heal(self, response=None):
 
         missing_conns = yield self.process_brokers(response.brokers)
         missing_topics = self.process_topics(response.topics)
+
         while missing_conns or missing_topics:
             response = yield self.get_metadata(topics=list(missing_topics))
             missing_conns = yield self.process_brokers(response.brokers)

kiel/iterables.py

@@ -1,3 +1,5 @@
+from __future__ import generator_stop
+
 def drain(iterable):
     """
     Helper method that empties an iterable as it is iterated over.
@@ -23,4 +25,5 @@ def next_item(coll):
         try:
             yield next_item(iterable)
         except (IndexError, KeyError):
-            raise StopIteration
+            return
+        # merge from fork https://github.com/denissmirnov/kiel/commit/fa80aa1ccd790c0fbbd8cc46a72162195e1aed69