KIP-735: Increase default consumer session_timeout_ms from 10s to 45s (#3030)

Author: dpkp

kafka/consumer/group.py

@@ -166,7 +166,7 @@ class KafkaConsumer:
             from the group and initiate a rebalance. Note that the value must
             be in the allowable range as configured in the broker configuration
             by group.min.session.timeout.ms and group.max.session.timeout.ms.
-            Default: 10000
+            Default: 45000 for brokers 3.0+, otherwise 30000.
         heartbeat_interval_ms (int): The expected time in milliseconds
             between heartbeats to the consumer coordinator when using
             Kafka's group management facilities. Heartbeats are used to ensure
@@ -317,7 +317,7 @@ class KafkaConsumer:
         'partition_assignment_strategy': (RangePartitionAssignor, RoundRobinPartitionAssignor),
         'max_poll_records': 500,
         'max_poll_interval_ms': 300000,
-        'session_timeout_ms': 10000,
+        'session_timeout_ms': 45000,
         'heartbeat_interval_ms': 3000,
         'receive_buffer_bytes': None,
         'send_buffer_bytes': None,
@@ -354,11 +354,7 @@ class KafkaConsumer:
         'socks5_proxy': None,  # deprecated
         'kafka_client': KafkaNetClient,
     }
-    # Pre-0.10.1 brokers don't separate session_timeout_ms from
-    # max_poll_interval_ms; both default to this value when neither is
-    # user-supplied. Kept under request_timeout_ms (30s) so the strict
-    # request > session check below doesn't fire on the default path.
-    DEFAULT_SESSION_TIMEOUT_MS_0_9 = 25000
+    DEFAULT_SESSION_TIMEOUT_MS_PRE_KIP_735 = 30000
 
     def __init__(self, *topics, **configs):
         # Only check for extra config keys in top-level class
@@ -416,21 +412,21 @@ def __init__(self, *topics, **configs):
         # Coordinator configurations are different for older brokers
         # max_poll_interval_ms is not supported directly -- it must the be
         # the same as session_timeout_ms. If the user provides one of them,
-        # use it for both. Otherwise use the old default of 30secs
-        if self.config['api_version'] < (0, 10, 1):
-            if 'session_timeout_ms' not in configs:
-                if 'max_poll_interval_ms' in configs:
-                    self.config['session_timeout_ms'] = configs['max_poll_interval_ms']
-                else:
-                    self.config['session_timeout_ms'] = self.DEFAULT_SESSION_TIMEOUT_MS_0_9
-            if 'max_poll_interval_ms' not in configs:
-                self.config['max_poll_interval_ms'] = self.config['session_timeout_ms']
+        # use it for both.
+        user_supplied_session_timeout = 'session_timeout_ms' in configs
+        user_supplied_max_poll_interval = 'max_poll_interval_ms' in configs
 
-        if self.config['group_id'] is not None:
-            if self.config['request_timeout_ms'] <= self.config['session_timeout_ms']:
-                raise KafkaConfigurationError(
-                    "Request timeout (%s) must be larger than session timeout (%s)" %
-                    (self.config['request_timeout_ms'], self.config['session_timeout_ms']))
+        if not user_supplied_session_timeout:
+            if self.config['api_version'] < (0, 10, 1) and user_supplied_max_poll_interval:
+                self.config['session_timeout_ms'] = self.config['max_poll_interval_ms']
+
+            elif self.config['api_version'] < (3, 0):
+                # Prior to 3.0 the broker-side default max session timeout was 30000
+                self.config['session_timeout_ms'] = self.DEFAULT_SESSION_TIMEOUT_MS_PRE_KIP_735
+
+        if not user_supplied_max_poll_interval:
+            if self.config['api_version'] < (0, 10, 1):
+                self.config['max_poll_interval_ms'] = self.config['session_timeout_ms']
 
         if self.config['group_instance_id'] is not None:
             if self.config['group_id'] is None:

kafka/coordinator/base.py

@@ -109,7 +109,7 @@ class BaseCoordinator(metaclass=abc.ABCMeta):
     DEFAULT_CONFIG = {
         'group_id': 'kafka-python-default-group',
         'group_instance_id': None,
-        'session_timeout_ms': 10000,
+        'session_timeout_ms': 45000,
         'heartbeat_interval_ms': 3000,
         'max_poll_interval_ms': 300000,
         'request_timeout_ms': 30000,
@@ -118,15 +118,31 @@ class BaseCoordinator(metaclass=abc.ABCMeta):
         'metrics': None,
         'metric_group_prefix': '',
     }
+    DEFAULT_SESSION_TIMEOUT_MS_PRE_KIP_735 = 30000
 
     def __init__(self, client, **configs):
         """
         Keyword Arguments:
             group_id (str): name of the consumer group to join for dynamic
                 partition assignment (if enabled), and to use for fetching and
                 committing offsets. Default: 'kafka-python-default-group'
+            group_instance_id (str): A unique identifier of the consumer instance
+                provided by end user. Only non-empty strings are permitted. If set,
+                the consumer is treated as a static member, which means that only
+                one instance with this ID is allowed in the consumer group at any
+                time. This can be used in combination with a larger session timeout
+                to avoid group rebalances caused by transient unavailability (e.g.
+                process restarts). If not set, the consumer will join the group as
+                a dynamic member, which is the traditional behavior. Default: None
             session_timeout_ms (int): The timeout used to detect failures when
-                using Kafka's group management facilities. Default: 30000
+                using Kafka's group management facilities. The consumer sends
+                periodic heartbeats to indicate its liveness to the broker. If
+                no heartbeats are received by the broker before the expiration of
+                this session timeout, then the broker will remove this consumer
+                from the group and initiate a rebalance. Note that the value must
+                be in the allowable range as configured in the broker configuration
+                by group.min.session.timeout.ms and group.max.session.timeout.ms.
+                Default: 45000 for brokers 3.0+, otherwise 30000.
             heartbeat_interval_ms (int): The expected time in milliseconds
                 between heartbeats to the consumer coordinator when using
                 Kafka's group management feature. Heartbeats are used to ensure
@@ -144,6 +160,29 @@ def __init__(self, client, **configs):
             if key in configs:
                 self.config[key] = configs[key]
 
+        # Coordinator configurations are different for older brokers
+        # max_poll_interval_ms is not supported directly -- it must the be
+        # the same as session_timeout_ms. If the user provides one of them,
+        # use it for both.
+        user_supplied_session_timeout = 'session_timeout_ms' in configs
+        user_supplied_max_poll_interval = 'max_poll_interval_ms' in configs
+
+        if not user_supplied_session_timeout:
+            if self.config['api_version'] < (0, 10, 1) and user_supplied_max_poll_interval:
+                self.config['session_timeout_ms'] = self.config['max_poll_interval_ms']
+
+            elif self.config['api_version'] < (3, 0):
+                # Prior to 3.0 the broker-side default max session timeout was 30000
+                self.config['session_timeout_ms'] = self.DEFAULT_SESSION_TIMEOUT_MS_PRE_KIP_735
+
+        if not user_supplied_max_poll_interval:
+            if self.config['api_version'] < (0, 10, 1):
+                self.config['max_poll_interval_ms'] = self.config['session_timeout_ms']
+
+        if self.config['group_instance_id'] is not None:
+            if self.config['group_id'] is None:
+                raise Errors.KafkaConfigurationError("group_instance_id requires group_id")
+
         if self.config['api_version'] < (0, 10, 1):
             if self.config['max_poll_interval_ms'] != self.config['session_timeout_ms']:
                 raise Errors.KafkaConfigurationError("Broker version %s does not support "

kafka/coordinator/consumer.py

@@ -27,24 +27,16 @@
 
 class ConsumerCoordinator(BaseCoordinator):
     """This class manages the coordination process with the consumer coordinator."""
-    DEFAULT_CONFIG = {
-        'group_id': 'kafka-python-default-group',
-        'group_instance_id': None,
+    DEFAULT_CONFIG = BaseCoordinator.DEFAULT_CONFIG.copy()
+    DEFAULT_CONFIG.update({
         'enable_auto_commit': True,
         'auto_commit_interval_ms': 5000,
         'default_offset_commit_callback': None,
         'assignors': (RangePartitionAssignor, RoundRobinPartitionAssignor, StickyPartitionAssignor),
-        'session_timeout_ms': 10000,
-        'heartbeat_interval_ms': 3000,
-        'max_poll_interval_ms': 300000,
-        'request_timeout_ms': 30000,
-        'retry_backoff_ms': 100,
-        'api_version': (0, 10, 1),
         'exclude_internal_topics': True,
         'isolation_level': 'read_uncommitted',
-        'metrics': None,
         'metric_group_prefix': 'consumer'
-    }
+    })
 
     def __init__(self, client, subscription, **configs):
         """Initialize the coordination manager.
@@ -53,14 +45,6 @@ def __init__(self, client, subscription, **configs):
             group_id (str): name of the consumer group to join for dynamic
                 partition assignment (if enabled), and to use for fetching and
                 committing offsets. Default: 'kafka-python-default-group'
-            group_instance_id (str): A unique identifier of the consumer instance
-                provided by end user. Only non-empty strings are permitted. If set,
-                the consumer is treated as a static member, which means that only
-                one instance with this ID is allowed in the consumer group at any
-                time. This can be used in combination with a larger session timeout
-                to avoid group rebalances caused by transient unavailability (e.g.
-                process restarts). If not set, the consumer will join the group as
-                a dynamic member, which is the traditional behavior. Default: None
             enable_auto_commit (bool): If true the consumer's offset will be
                 periodically committed in the background. Default: True.
             auto_commit_interval_ms (int): milliseconds between automatic
@@ -72,17 +56,6 @@ def __init__(self, client, subscription, **configs):
             assignors (list): List of objects to use to distribute partition
                 ownership amongst consumer instances when group management is
                 used. Default: [RangePartitionAssignor, RoundRobinPartitionAssignor, StickyPartitionAssignor]
-            heartbeat_interval_ms (int): The expected time in milliseconds
-                between heartbeats to the consumer coordinator when using
-                Kafka's group management feature. Heartbeats are used to ensure
-                that the consumer's session stays active and to facilitate
-                rebalancing when new consumers join or leave the group. The
-                value must be set lower than session_timeout_ms, but typically
-                should be set no higher than 1/3 of that value. It can be
-                adjusted even lower to control the expected time for normal
-                rebalances. Default: 3000
-            session_timeout_ms (int): The timeout used to detect failures when
-                using Kafka's group management facilities. Default: 30000
             retry_backoff_ms (int): Milliseconds to backoff when retrying on
                 errors. Default: 100.
             exclude_internal_topics (bool): Whether records from internal topics
@@ -121,11 +94,6 @@ def __init__(self, client, subscription, **configs):
             if self._use_group_apis:
                 if not self.config['assignors']:
                     raise Errors.KafkaConfigurationError('Coordinator requires assignors')
-            if self.config['api_version'] < (0, 10, 1):
-                if self.config['max_poll_interval_ms'] != self.config['session_timeout_ms']:
-                    raise Errors.KafkaConfigurationError("Broker version %s does not support "
-                                                         "different values for max_poll_interval_ms "
-                                                         "and session_timeout_ms")
 
         if self.config['enable_auto_commit']:
             if not self._use_offset_apis:
@@ -264,7 +232,7 @@ def _lookup_assignor(self, name):
     # warning. Sync listeners on the IO loop will block heartbeats while
     # they run; even async ones delay rebalance progress. 1s is a soft
     # ceiling: well below default heartbeat_interval_ms (3s) and
-    # session_timeout_ms (10s).
+    # session_timeout_ms (45s).
     _REBALANCE_LISTENER_WARN_SECS = 1.0
 
     async def _invoke_rebalance_listener_async(self, method_name, arg):

test/consumer/test_consumer.py

@@ -4,9 +4,9 @@
 from kafka.errors import KafkaConfigurationError, IllegalStateError
 
 
-def test_session_timeout_larger_than_request_timeout_raises():
+def test_session_timeout_different_from_max_poll_timeout_raises():
     with pytest.raises(KafkaConfigurationError):
-        KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9), group_id='foo', session_timeout_ms=50000, request_timeout_ms=40000)
+        KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9), group_id='foo', session_timeout_ms=50000, max_poll_timeout_ms=40000)
 
 
 def test_fetch_max_wait_larger_than_request_timeout_raises():

test/consumer/test_coordinator.py

@@ -34,9 +34,7 @@
 def coordinator(broker, client, metrics):
     coord = ConsumerCoordinator(client, SubscriptionState(),
                                 metrics=metrics,
-                                api_version=broker.broker_version,
-                                max_poll_interval_ms=300000 if broker.broker_version >= (0, 10, 1) else 10000,
-                                session_timeout_ms=10000)
+                                api_version=broker.broker_version)
     try:
         yield coord
     finally:
@@ -560,8 +558,6 @@ def test_maybe_auto_commit_offsets_sync(mocker, client, api_version, group_id, e
     mock_exc = mocker.patch('kafka.coordinator.consumer.log.exception')
     coordinator = ConsumerCoordinator(client, SubscriptionState(),
                                       api_version=api_version,
-                                      session_timeout_ms=30000,
-                                      max_poll_interval_ms=30000,
                                       enable_auto_commit=enable,
                                       group_id=group_id)
     commit_sync = mocker.patch.object(coordinator, 'commit_offsets_sync',
@@ -864,9 +860,7 @@ def test_send_offset_fetch_request_sets_require_stable(
     coord = ConsumerCoordinator(client, SubscriptionState(),
                                 metrics=metrics,
                                 api_version=broker.broker_version,
-                                isolation_level=isolation_level,
-                                max_poll_interval_ms=300000,
-                                session_timeout_ms=10000)
+                                isolation_level=isolation_level)
     try:
         client._manager.bootstrap(timeout_ms=5000)
         coord._subscription.subscribe(topics=['foobar'])
@@ -1571,8 +1565,6 @@ def _cooperative_coordinator(client, metrics):
         client, SubscriptionState(),
         metrics=metrics,
         api_version=(2, 4),
-        max_poll_interval_ms=300000,
-        session_timeout_ms=10000,
         assignors=(CooperativeStickyAssignor,))
 
 
@@ -1605,8 +1597,6 @@ def test_rejects_mixed_protocols(self, client, metrics):
                 client, SubscriptionState(),
                 metrics=metrics,
                 api_version=(2, 4),
-                max_poll_interval_ms=300000,
-                session_timeout_ms=10000,
                 assignors=(RangePartitionAssignor, CooperativeStickyAssignor))