From f6d216856f044de7b54450e00858190bc6dafc4f Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Thu, 4 Aug 2016 20:09:13 -0700 Subject: Drain pending requests to the coordinator before initiating group rejoin (#798) --- kafka/coordinator/base.py | 8 ++++++++ 1 file changed, 8 insertions(+) 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) -- cgit v1.2.1