diff options
| -rw-r--r-- | kafka/coordinator/base.py | 8 |
1 files changed, 8 insertions, 0 deletions
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index e147b6f..5f60aa3 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -242,6 +242,14 @@ class BaseCoordinator(object): while self.need_rejoin(): self.ensure_coordinator_known() + # ensure that there are no pending requests to the coordinator. + # This is important in particular to avoid resending a pending + # JoinGroup request. + if self._client.in_flight_request_count(self.coordinator_id): + while self._client.in_flight_request_count(self.coordinator_id): + self._client.poll() + continue + future = self._send_join_group_request() self._client.poll(future=future) |