Change coordinator.needs_join_prepare -> coordinator.rejoining; fix consumer group test (dpkp#795)

Author: dpkp

kafka/coordinator/base.py

@@ -89,7 +89,7 @@ def __init__(self, client, metrics, **configs):
         self.group_id = self.config['group_id']
         self.coordinator_id = None
         self.rejoin_needed = True
-        self.needs_join_prepare = True
+        self.rejoining = False
         self.heartbeat = Heartbeat(**self.config)
         self.heartbeat_task = HeartbeatTask(weakref.proxy(self))
         self.sensors = GroupCoordinatorMetrics(self.heartbeat, metrics,
@@ -235,9 +235,9 @@ def ensure_active_group(self):
         if not self.need_rejoin():
             return
 
-        if self.needs_join_prepare:
+        if not self.rejoining:
             self._on_join_prepare(self.generation, self.member_id)
-            self.needs_join_prepare = False
+            self.rejoining = True
 
         while self.need_rejoin():
             self.ensure_coordinator_known()
@@ -249,7 +249,7 @@ def ensure_active_group(self):
                 member_assignment_bytes = future.value
                 self._on_join_complete(self.generation, self.member_id,
                                        self.protocol, member_assignment_bytes)
-                self.needs_join_prepare = True
+                self.rejoining = False
                 self.heartbeat_task.reset()
             else:
                 assert future.failed()

test/test_consumer_group.py

@@ -95,7 +95,12 @@ def consumer_thread(i):
                 generations = set([consumer._coordinator.generation
                                    for consumer in list(consumers.values())])
 
-                if len(generations) == 1:
+                # New generation assignment is not complete until
+                # coordinator.rejoining = False
+                rejoining = any([consumer._coordinator.rejoining
+                                 for consumer in list(consumers.values())])
+
+                if not rejoining and len(generations) == 1:
                     for c, consumer in list(consumers.items()):
                         logging.info("[%s] %s %s: %s", c,
                                      consumer._coordinator.generation,