Catch duplicate batch.done() calls -- this can happen if we maybe_expire then process a response errback

1 files changed, 4 insertions, 1 deletions

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)