diff options
| -rw-r--r-- | kafka/consumer/kafka.py | 12 |
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. |