Optionally skip auto-commit during consumer.close (#1031)

kafka/consumer/group.py

@@ -376,13 +376,13 @@ class KafkaConsumer(six.Iterator):
         """
         return self._subscription.assigned_partitions()
 
-    def close(self):
+    def close(self, autocommit=True):
         """Close the consumer, waiting indefinitely for any needed cleanup."""
         if self._closed:
             return
         log.debug("Closing the KafkaConsumer.")
         self._closed = True
-        self._coordinator.close()
+        self._coordinator.close(autocommit=autocommit)
         self._metrics.close()
         self._client.close()
         try:

kafka/coordinator/consumer.py

@@ -325,8 +325,9 @@ class ConsumerCoordinator(BaseCoordinator):
 
             time.sleep(self.config['retry_backoff_ms'] / 1000.0)
 
-    def close(self):
+    def close(self, autocommit=True):
         try:
+            if autocommit:
                 self._maybe_auto_commit_offsets_sync()
         finally:
             super(ConsumerCoordinator, self).close()