Merge pull request #389 from dpkp/task_done_key_error

KafkaConsumer.task_done: warn and skip unrecognized topic-partitions

1 files changed, 10 insertions, 2 deletions

diff --git a/kafka/consumer/kafka.py b/kafka/consumer/kafka.py
index 60f1a0b..7ba83cb 100644
--- a/kafka/consumer/kafka.py
+++ b/kafka/consumer/kafka.py
@@ -448,10 +448,16 @@ class KafkaConsumer(object):
             message (KafkaMessage): the message to mark as complete
 
         Returns:
-            Nothing
-
+            True, unless the topic-partition for this message has not
+            been configured for the consumer. In normal operation, this
+            should not happen. But see github issue 364.
         """
         topic_partition = (message.topic, message.partition)
+        if topic_partition not in self._topics:
+            logger.warning('Unrecognized topic/partition in task_done message: '
+                           '{0}:{1}'.format(*topic_partition))
+            return False
+
         offset = message.offset
 
         # Warn on non-contiguous offsets
@@ -476,6 +482,8 @@ class KafkaConsumer(object):
         if self._should_auto_commit():
             self.commit()
 
+        return True
+
     def commit(self):
         """Store consumed message offsets (marked via task_done())
         to kafka cluster for this consumer_group.