Add more cleanup in consumer.stop()

1 files changed, 7 insertions, 5 deletions

diff --git a/kafka/consumer.py b/kafka/consumer.py
index 23a2f90..6ceea72 100644
--- a/kafka/consumer.py
+++ b/kafka/consumer.py
@@ -159,6 +159,11 @@ class Consumer(object):
         if self.count_since_commit > self.auto_commit_every_n:
             self.commit()
 
+    def stop(self):
+        if self.commit_timer is not None:
+            self.commit_timer.stop()
+            self.commit()
+
     def pending(self, partitions=None):
         """
         Gets the pending message count
@@ -226,11 +231,6 @@ class SimpleConsumer(Consumer):
         """
         self.partition_info = True
 
-    def stop(self):
-        if self.commit_timer is not None:
-            self.commit_timer.stop()
-            self.commit()
-
     def seek(self, offset, whence):
         """
         Alter the current offset in the consumer, similar to fseek
@@ -510,6 +510,8 @@ class MultiProcessConsumer(Consumer):
             proc.join()
             proc.terminate()
 
+        super(MultiProcessConsumer, self).stop()
+
     def __iter__(self):
         """
         Iterator to consume the messages available on this consumer