From 99bc503b1a549bd0877706ec8f04f7cb35445cda Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Thu, 18 Feb 2016 22:46:04 -0800 Subject: Catch duplicate batch.done() calls -- this can happen if we maybe_expire then process a response errback --- kafka/producer/record_accumulator.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/kafka/producer/record_accumulator.py b/kafka/producer/record_accumulator.py index c404e9e..24cf8af 100644 --- a/kafka/producer/record_accumulator.py +++ b/kafka/producer/record_accumulator.py @@ -68,7 +68,10 @@ class RecordBatch(object): log.debug("Produced messages to topic-partition %s with base offset" " %s and error %s.", self.topic_partition, base_offset, exception) # trace - if exception is None: + if self.produce_future.is_done: + log.warning('Batch is already closed -- ignoring batch.done()') + return + elif exception is None: self.produce_future.success(base_offset) else: self.produce_future.failure(exception) -- cgit v1.2.1