Merge upstream master

Author: chinmaychandak

streamz/sources.py

@@ -515,7 +515,8 @@ def checkpoint_emit(_part):
                         continue
                     if 'auto.offset.reset' in self.consumer_params.keys():
                         if self.consumer_params['auto.offset.reset'] == 'latest' and \
-                                self.positions == [-1001] * self.npartitions:
+                                (self.positions == [-1001] * self.npartitions
+                                 or self.positions == [0] * self.npartitions):
                             self.positions[partition] = high
                     current_position = self.positions[partition]
                     lowest = max(current_position, low)