Drain pending requests to the coordinator before initiating group rejoin (#798)

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)