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

2 files changed, 5 insertions, 4 deletions

diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 32f4556..7d451b3 100644
--- a/kafka/consumer/group.py
+++ b/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:
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index fac8144..fdbb995 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -325,9 +325,10 @@ class ConsumerCoordinator(BaseCoordinator):
 
             time.sleep(self.config['retry_backoff_ms'] / 1000.0)
 
-    def close(self):
+    def close(self, autocommit=True):
         try:
-            self._maybe_auto_commit_offsets_sync()
+            if autocommit:
+                self._maybe_auto_commit_offsets_sync()
         finally:
             super(ConsumerCoordinator, self).close()