Merge pull request #827 from Parsely/bugfix/cleanup_fetch

call cleanup() in fetch instead of autocommitter

pykafka/simpleconsumer.py

@@ -424,7 +424,6 @@ def autocommitter():
                     # surface all exceptions to the main thread
                     self._worker_exception = sys.exc_info()
                     break
-            self.cleanup()
             log.debug("Autocommitter thread exiting")
         log.debug("Starting autocommitter thread")
         return self._cluster.handler.spawn(autocommitter, name="pykafka.SimpleConsumer.autocommiter")
@@ -447,6 +446,11 @@ def fetcher():
                     # surface all exceptions to the main thread
                     self._worker_exception = sys.exc_info()
                     break
+            try:
+                self.cleanup()
+            except ReferenceError as e:
+                log.debug("Attempt to cleanup consumer failed")
+                log.exception(e)
             log.debug("Fetcher thread exiting")
         log.info("Starting %s fetcher threads", self._num_consumer_fetchers)
         return [self._cluster.handler.spawn(fetcher, name="pykafka.SimpleConsumer.fetcher")