dpkp: Would a warning be more appropriate here? Is this configuration always doomed, or is it possible some power user would want do go against the grain in some circumstance?
jeffwidman: > Would a warning be more appropriate here? Is this configuration always doomed, or is it possible some power user would want do go against the grain in some circumstance?

It's a great question, and I've thought about it several times. But each time I cannot come up with a valid scenario... unlike Zookeeper, clients don't keep any session/request state that spans multiple connections.
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 531c107..f521891 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -313,11 +313,15 @@ class KafkaConsumer(six.Iterator):
                         new_config, self.config['auto_offset_reset'])
             self.config['auto_offset_reset'] = new_config
 
+        connections_max_idle_ms = self.config['connections_max_idle_ms']
         request_timeout_ms = self.config['request_timeout_ms']
         fetch_max_wait_ms = self.config['fetch_max_wait_ms']
-        if request_timeout_ms <= fetch_max_wait_ms:
-            raise KafkaConfigurationError("Request timeout (%s) must be larger than fetch-max-wait-ms (%s)" %
-                                          (request_timeout_ms, fetch_max_wait_ms))
+        if not (fetch_max_wait_ms < request_timeout_ms < connections_max_idle_ms):
+            raise KafkaConfigurationError(
+                "connections_max_idle_ms ({}) must be larger than "
+                "request_timeout_ms ({}) which must be larger than "
+                "fetch_max_wait_ms ({})."
+                .format(connections_max_idle_ms, request_timeout_ms, fetch_max_wait_ms))
 
         metrics_tags = {'client-id': self.config['client_id']}
         metric_config = MetricConfig(samples=self.config['metrics_num_samples'],

Verify these timeouts are in descending order on KafkaConsumer instantiation
When we create a `KafkaConsumer`, we should validate that the following timeouts are in descending order:
1. `connections_max_idle_ms`
2. `request_timeout_ms`
3. `session_timeout_ms` (or equivalent, depending on pre/post 0.10.1 broker)

If they are not, then raise an error... otherwise the consumer can end up in weird states.
diff --git a/test/test_consumer.py b/test/test_consumer.py
index 4ea01c8..edcc2d8 100644
--- a/test/test_consumer.py
+++ b/test/test_consumer.py
@@ -15,11 +15,15 @@ from kafka.structs import (
 class TestKafkaConsumer:
     def test_session_timeout_larger_than_request_timeout_raises(self):
         with pytest.raises(KafkaConfigurationError):
-            KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0,9), group_id='foo', session_timeout_ms=60000, request_timeout_ms=40000)
+            KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9), group_id='foo', session_timeout_ms=50000, request_timeout_ms=40000)
 
     def test_fetch_max_wait_larger_than_request_timeout_raises(self):
         with pytest.raises(KafkaConfigurationError):
-            KafkaConsumer(bootstrap_servers='localhost:9092', fetch_max_wait_ms=41000, request_timeout_ms=40000)
+            KafkaConsumer(bootstrap_servers='localhost:9092', fetch_max_wait_ms=50000, request_timeout_ms=40000)
+
+    def test_request_timeout_larger_than_connections_max_idle_ms_raises(self):
+        with pytest.raises(KafkaConfigurationError):
+            KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9), request_timeout_ms=50000, connections_max_idle_ms=40000)
 
     def test_subscription_copy(self):
         consumer = KafkaConsumer('foo', api_version=(0, 10))

pip install -e .[dev]
parse_log_pytest
requirements.txt
pytest pytest-cov pytest-pylint pytest-mock mock python-snappy lz4 xxhash crc32c pytest
apt-get update apt-get install -y gcc libsnappy-dev
3.7
requirements/base.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
astroid==2.15.8
certifi @ file:///croot/certifi_1671487769961/work/certifi
coverage==7.2.7
cramjam==2.8.3
crc32c==2.7.1
dill==0.3.7
exceptiongroup==1.2.2
importlib-metadata==6.7.0
iniconfig==2.0.0
isort==5.11.5
-e git+https://github.com/dpkp/kafka-python.git@812de351f75beefe73bd9bef55847ab61ccc951d#egg=kafka_python
lazy-object-proxy==1.9.0
lz4==4.3.2
mccabe==0.7.0
mock==5.2.0
packaging==24.0
platformdirs==4.0.0
pluggy==1.2.0
pylint==2.17.7
pytest==7.4.4
pytest-cov==4.1.0
pytest-mock==3.11.1
pytest-pylint==0.21.0
python-snappy==0.7.3
tomli==2.0.1
tomlkit==0.12.5
typed-ast==1.5.5
typing_extensions==4.7.1
wrapt==1.16.0
xxhash==3.5.0
zipp==3.15.0

name: kafka-python
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2022.12.7=py37h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=22.3.1=py37h06a4308_0
  - python=3.7.16=h7a1cb2a_0
  - readline=8.2=h5eee18b_0
  - setuptools=65.6.3=py37h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.38.4=py37h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - astroid==2.15.8
      - coverage==7.2.7
      - cramjam==2.8.3
      - crc32c==2.7.1
      - dill==0.3.7
      - exceptiongroup==1.2.2
      - importlib-metadata==6.7.0
      - iniconfig==2.0.0
      - isort==5.11.5
      - lazy-object-proxy==1.9.0
      - lz4==4.3.2
      - mccabe==0.7.0
      - mock==5.2.0
      - packaging==24.0
      - platformdirs==4.0.0
      - pluggy==1.2.0
      - pylint==2.17.7
      - pytest==7.4.4
      - pytest-cov==4.1.0
      - pytest-mock==3.11.1
      - pytest-pylint==0.21.0
      - python-snappy==0.7.3
      - tomli==2.0.1
      - tomlkit==0.12.5
      - typed-ast==1.5.5
      - typing-extensions==4.7.1
      - wrapt==1.16.0
      - xxhash==3.5.0
      - zipp==3.15.0
prefix: /opt/conda/envs/kafka-python

test/test_consumer.py::TestKafkaConsumer::test_request_timeout_larger_than_connections_max_idle_ms_raises
test/test_consumer.py::TestKafkaConsumer::test_session_timeout_larger_than_request_timeout_raises test/test_consumer.py::TestKafkaConsumer::test_fetch_max_wait_larger_than_request_timeout_raises test/test_consumer.py::TestKafkaConsumer::test_subscription_copy test/test_consumer.py::TestMultiProcessConsumer::test_partition_list test/test_consumer.py::TestSimpleConsumer::test_non_integer_partitions test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_commit_does_not_raise test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_failed_payloads test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_leader_change test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_reset_partition_offset test/test_consumer.py::TestSimpleConsumer::test_simple_consumer_unknown_topic_partition
kafka/consumer/group.py
kafka/consumer/group.py