Repository: dpkp/kafka-python
Pull request #2623 (2025-05-15 20:49:08)

Update offset commit error handling; use RebalanceInProgressError if applicable
Also do not pause offset commits during metadata update.


Patch:
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py
index 8521578..c408c26 100644
--- a/kafka/coordinator/base.py
+++ b/kafka/coordinator/base.py
@@ -5,6 +5,7 @@ import copy
 import logging
 import threading
 import time
+import warnings
 import weakref
 
 from kafka.vendor import six
@@ -250,6 +251,11 @@ class BaseCoordinator(object):
         else:
             return self.coordinator_id
 
+    def connected(self):
+        """Return True iff the coordinator node is connected"""
+        with self._lock:
+            return self.coordinator_id is not None and self._client.connected(self.coordinator_id)
+
     def ensure_coordinator_ready(self, timeout_ms=None):
         """Block until the coordinator for this group is known.
 
@@ -792,7 +798,7 @@ class BaseCoordinator(object):
                         self.coordinator_id, self.group_id, error)
             self.coordinator_id = None
 
-    def generation(self):
+    def generation_if_stable(self):
         """Get the current generation state if the group is stable.
 
         Returns: the current generation or None if the group is unjoined/rebalancing
@@ -802,6 +808,15 @@ class BaseCoordinator(object):
                 return None
             return self._generation
 
+    # deprecated
+    def generation(self):
+        warnings.warn("Function coordinator.generation() has been renamed to generation_if_stable()",
+                      DeprecationWarning, stacklevel=2)
+        return self.generation_if_stable()
+
+    def rebalance_in_progress(self):
+        return self.state is MemberState.REBALANCING
+
     def reset_generation(self, member_id=UNKNOWN_MEMBER_ID):
         """Reset the generation and member_id because we have fallen out of the group."""
         with self._lock:
@@ -1058,28 +1073,28 @@ class HeartbeatThread(threading.Thread):
         self.coordinator._client._lock.acquire()
         self.coordinator._lock.acquire()
         try:
-            if self.enabled and self.coordinator.state is MemberState.STABLE:
-                # TODO: When consumer.wakeup() is implemented, we need to
-                # disable here to prevent propagating an exception to this
-                # heartbeat thread
-                # must get client._lock, or maybe deadlock at heartbeat 
-                # failure callback in consumer poll
-                self.coordinator._client.poll(timeout_ms=0)
-
             if not self.enabled:
                 heartbeat_log.debug('Heartbeat disabled. Waiting')
                 self.coordinator._client._lock.release()
                 self.coordinator._lock.wait()
-                heartbeat_log.debug('Heartbeat re-enabled.')
+                if self.enabled:
+                    heartbeat_log.debug('Heartbeat re-enabled.')
+                return
 
-            elif self.coordinator.state is not MemberState.STABLE:
+            if self.coordinator.state is not MemberState.STABLE:
                 # the group is not stable (perhaps because we left the
                 # group or because the coordinator kicked us out), so
                 # disable heartbeats and wait for the main thread to rejoin.
                 heartbeat_log.debug('Group state is not stable, disabling heartbeats')
                 self.disable()
+                return
+
+            # TODO: When consumer.wakeup() is implemented, we need to
+            # disable here to prevent propagating an exception to this
+            # heartbeat thread
+            self.coordinator._client.poll(timeout_ms=0)
 
-            elif self.coordinator.coordinator_unknown():
+            if self.coordinator.coordinator_unknown():
                 future = self.coordinator.lookup_coordinator()
                 if not future.is_done or future.failed():
                     # the immediate future check ensures that we backoff
@@ -1088,6 +1103,10 @@ class HeartbeatThread(threading.Thread):
                     self.coordinator._client._lock.release()
                     self.coordinator._lock.wait(self.coordinator.config['retry_backoff_ms'] / 1000)
 
+            elif not self.coordinator.connected():
+                self.coordinator._client._lock.release()
+                self.coordinator._lock.wait(self.coordinator.config['retry_backoff_ms'] / 1000)
+
             elif self.coordinator.heartbeat.session_timeout_expired():
                 # the session timeout has expired without seeing a
                 # successful heartbeat, so we should probably make sure
@@ -1103,11 +1122,10 @@ class HeartbeatThread(threading.Thread):
                 self.coordinator.maybe_leave_group()
 
             elif not self.coordinator.heartbeat.should_heartbeat():
-                # poll again after waiting for the retry backoff in case
-                # the heartbeat failed or the coordinator disconnected
-                heartbeat_log.log(0, 'Not ready to heartbeat, waiting')
+                next_hb = self.coordinator.heartbeat.time_to_next_heartbeat()
+                heartbeat_log.debug('Waiting %0.1f secs to send next heartbeat', next_hb)
                 self.coordinator._client._lock.release()
-                self.coordinator._lock.wait(self.coordinator.config['retry_backoff_ms'] / 1000)
+                self.coordinator._lock.wait(next_hb)
 
             else:
                 self.coordinator.heartbeat.sent_heartbeat()
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 3db00d7..ddd413b 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -608,6 +608,11 @@ class ConsumerCoordinator(BaseCoordinator):
         if node_id is None:
             return Future().failure(Errors.CoordinatorNotAvailableError)
 
+        # Verify node is ready
+        if not self._client.ready(node_id, metadata_priority=False):
+            log.debug("Node %s not ready -- failing offset commit request",
+                      node_id)
+            return Future().failure(Errors.NodeNotReadyError)
 
         # create the offset commit request
         offset_data = collections.defaultdict(dict)
@@ -616,7 +621,7 @@ class ConsumerCoordinator(BaseCoordinator):
 
         version = self._client.api_version(OffsetCommitRequest, max_version=6)
         if version > 1 and self._subscription.partitions_auto_assigned():
-            generation = self.generation()
+            generation = self.generation_if_stable()
         else:
             generation = Generation.NO_GENERATION
 
@@ -625,7 +630,18 @@ class ConsumerCoordinator(BaseCoordinator):
         # and let the user rejoin the group in poll()
         if generation is None:
             log.info("Failing OffsetCommit request since the consumer is not part of an active group")
-            return Future().failure(Errors.CommitFailedError('Group rebalance in progress'))
+            if self.rebalance_in_progress():
+                # if the client knows it is already rebalancing, we can use RebalanceInProgressError instead of
+                # CommitFailedError to indicate this is not a fatal error
+                return Future().failure(Errors.RebalanceInProgressError(
+                    "Offset commit cannot be completed since the"
+                    " consumer is undergoing a rebalance for auto partition assignment. You can try completing the rebalance"
+                    " by calling poll() and then retry the operation."))
+            else:
+                return Future().failure(Errors.CommitFailedError(
+                    "Offset commit cannot be completed since the"
+                    " consumer is not part of an active group for auto partition assignment; it is likely that the consumer"
+                    " was kicked out of the group."))
 
         if version == 0:
             request = OffsetCommitRequest[version](
@@ -756,7 +772,7 @@ class ConsumerCoordinator(BaseCoordinator):
                     # However, we do not need to reset generations and just request re-join, such that
                     # if the caller decides to proceed and poll, it would still try to proceed and re-join normally.
                     self.request_rejoin()
-                    future.failure(Errors.CommitFailedError('Group rebalance in progress'))
+                    future.failure(Errors.CommitFailedError(error_type()))
                     return
                 elif error_type in (Errors.UnknownMemberIdError,
                                     Errors.IllegalGenerationError):
@@ -765,7 +781,7 @@ class ConsumerCoordinator(BaseCoordinator):
                     log.warning("OffsetCommit for group %s failed: %s",
                                 self.group_id, error)
                     self.reset_generation()
-                    future.failure(Errors.CommitFailedError())
+                    future.failure(Errors.CommitFailedError(error_type()))
                     return
                 else:
                     log.error("Group %s failed to commit partition %s at offset"
@@ -804,7 +820,7 @@ class ConsumerCoordinator(BaseCoordinator):
             return Future().failure(Errors.CoordinatorNotAvailableError)
 
         # Verify node is ready
-        if not self._client.ready(node_id):
+        if not self._client.ready(node_id, metadata_priority=False):
             log.debug("Node %s not ready -- failing offset fetch request",
                       node_id)
             return Future().failure(Errors.NodeNotReadyError)
diff --git a/kafka/errors.py b/kafka/errors.py
index 8985826..ac4eadf 100644
--- a/kafka/errors.py
+++ b/kafka/errors.py
@@ -24,14 +24,7 @@ class CommitFailedError(KafkaError):
     def __init__(self, *args):
         if not args:
             args = ("Commit cannot be completed since the group has already"
-                    " rebalanced and assigned the partitions to another member."
-                    " This means that the time between subsequent calls to poll()"
-                    " was longer than the configured max_poll_interval_ms, which"
-                    " typically implies that the poll loop is spending too much"
-                    " time message processing. You can address this either by"
-                    " increasing the rebalance timeout with max_poll_interval_ms,"
-                    " or by reducing the maximum size of batches returned in poll()"
-                    " with max_poll_records.",)
+                    " rebalanced and assigned the partitions to another member.",)
         super(CommitFailedError, self).__init__(*args)
 
 


Test patch:
diff --git a/test/test_coordinator.py b/test/test_coordinator.py
index 251de56..4ffe1d2 100644
--- a/test/test_coordinator.py
+++ b/test/test_coordinator.py
@@ -658,6 +658,7 @@ def test_heartbeat(mocker, patched_coord):
     heartbeat.enable()
     patched_coord.state = MemberState.STABLE
     mocker.spy(patched_coord, '_send_heartbeat_request')
+    mocker.patch.object(patched_coord, 'connected', return_value=True)
     mocker.patch.object(patched_coord.heartbeat, 'should_heartbeat', return_value=True)
     heartbeat._run_once()
     assert patched_coord._send_heartbeat_request.call_count == 1