Error if connections_max_idle_ms not larger than request_timeout_ms

`connections_max_idle_ms` must always be larger than `request_timeout_ms`
to avoid potentially unexpected behavior.

Fix #1680.

Author: jeffwidman

kafka/consumer/group.py

@@ -323,9 +323,12 @@ def __init__(self, *topics, **configs):
 
         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 connections_max_idle_ms <= request_timeout_ms <= fetch_max_wait_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'],

test/test_consumer.py

@@ -21,6 +21,10 @@ 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)
 
+    def test_connections_max_idle_ms_smaller_than_request_timeout_raises(self):
+        with pytest.raises(KafkaConfigurationError):
+            KafkaConsumer(bootstrap_servers='localhost:9092', connections_max_idle_ms=39000, request_timeout_ms=40000)
+
     def test_subscription_copy(self):
         consumer = KafkaConsumer('foo', api_version=(0, 10))
         sub = consumer.subscription()