diff options
| -rw-r--r-- | kafka/coordinator/base.py | 8 | ||||
| -rw-r--r-- | test/test_consumer_group.py | 7 |
2 files changed, 10 insertions, 5 deletions
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index d6ea6c0..e147b6f 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -89,7 +89,7 @@ class BaseCoordinator(object): 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 @@ class BaseCoordinator(object): 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 @@ class BaseCoordinator(object): 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() diff --git a/test/test_consumer_group.py b/test/test_consumer_group.py index 9fb057e..1acde5e 100644 --- a/test/test_consumer_group.py +++ b/test/test_consumer_group.py @@ -95,7 +95,12 @@ def test_group(kafka_broker, topic): 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, |