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

Author: dpkp

kafka/producer/record_accumulator.py

@@ -68,7 +68,10 @@ def done(self, base_offset=None, exception=None):
         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)