Ensure timeout is checked after each fetch position update in Consumer.position()

Author: k61n

kafka/consumer/group.py

@@ -781,7 +781,7 @@ def position(self, partition, timeout_ms=None):
             # batch update fetch positions for any partitions without a valid position
             if self._update_fetch_positions(timeout_ms=timer.timeout_ms):
                 position = self._subscription.assignment[partition].position
-            elif timer.expired:
+            if timer.expired:
                 return None
         else:
             return position.offset