megre/fix

Author: Martin Durant

streamz/sources.py

@@ -449,10 +449,13 @@ def start(self):
             self.stopped = False
             self.consumer = ck.Consumer(self.cpars)
             self.consumer.subscribe(self.topics)
-            weakref.finalize(self, lambda consumer=self.consumer: _close_consumer(consumer))
+            weakref.finalize(
+                self, lambda consumer=self.consumer: _close_consumer(consumer)
+            )
             tp = ck.TopicPartition(self.topics[0], 0, 0)
 
-             # blocks for consumer thread to come up and invoke poll to establish connection with broker to fetch oauth token for kafka
+            # blocks for consumer thread to come up and invoke poll to
+            # establish connection with broker to fetch oauth token for kafka
             self.consumer.poll()
             self.consumer.get_watermark_offsets(tp)
             self.loop.add_callback(self.poll_kafka)
@@ -480,7 +483,8 @@ def __init__(self, topic, consumer_params, poll_interval='1s',
                  max_batch_size=10000, keys=False,
                  engine=None, **kwargs):
         self.consumer_params = consumer_params
-        # Override the auto-commit config to enforce custom streamz checkpointing
+        # Override the auto-commit config to enforce custom streamz
+        # checkpointing
         self.consumer_params['enable.auto.commit'] = 'false'
         if 'auto.offset.reset' not in self.consumer_params.keys():
             consumer_params['auto.offset.reset'] = 'latest'