From db47136671a05283d9801c5a3ec74b3e0f38004e Mon Sep 17 00:00:00 2001
From: eastlondoner <eastlondoner@users.noreply.github.com>
Date: Wed, 29 Jun 2016 18:49:28 +0100
Subject: [PATCH] allow client.check_version timeout to be set in Producer and
 Consumer constructors (#647)

* allow client.check_version timeout to be set in Producer and Consumer constructors
---
 .gitignore              | 1 +
 kafka/consumer/group.py | 6 +++++-
 kafka/producer/kafka.py | 6 +++++-
 3 files changed, 11 insertions(+), 2 deletions(-)

diff --git a/.gitignore b/.gitignore
index 13be5912f..7d9069cd6 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,3 +11,4 @@ servers/*/resources/ssl*
 .noseids
 docs/_build
 .cache*
+.idea/
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index 106e96b3c..1e9b05252 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -149,6 +149,9 @@ class KafkaConsumer(six.Iterator):
             offset commits; 0.8.0 is what is left. If set to 'auto', will
             attempt to infer the broker version by probing various APIs.
             Default: auto
+        api_version_auto_timeout_ms (int): number of milliseconds to throw a
+            timeout exception from the constructor when checking the broker
+            api version. Only applies if api_version set to 'auto'
         metric_reporters (list): A list of classes to use as metrics reporters.
             Implementing the AbstractMetricsReporter interface allows plugging
             in classes that will be notified of new metric creation. Default: []
@@ -194,6 +197,7 @@ class KafkaConsumer(six.Iterator):
         'ssl_keyfile': None,
         'ssl_crlfile': None,
         'api_version': 'auto',
+        'api_version_auto_timeout_ms': 2000,
         'connections_max_idle_ms': 9 * 60 * 1000, # not implemented yet
         'metric_reporters': [],
         'metrics_num_samples': 2,
@@ -230,7 +234,7 @@ def __init__(self, *topics, **configs):
 
         # Check Broker Version if not set explicitly
         if self.config['api_version'] == 'auto':
-            self.config['api_version'] = self._client.check_version()
+            self.config['api_version'] = self._client.check_version(timeout=(self.config['api_version_auto_timeout_ms']/1000))
         assert self.config['api_version'] in ('0.10', '0.9', '0.8.2', '0.8.1', '0.8.0'), 'Unrecognized api version'
 
         # Convert api_version config to tuple for easy comparisons
diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py
index 7c55b276e..e1200f4d9 100644
--- a/kafka/producer/kafka.py
+++ b/kafka/producer/kafka.py
@@ -216,6 +216,9 @@ class KafkaProducer(object):
         api_version (str): specify which kafka API version to use.
             If set to 'auto', will attempt to infer the broker version by
             probing various APIs. Default: auto
+        api_version_auto_timeout_ms (int): number of milliseconds to throw a
+            timeout exception from the constructor when checking the broker
+            api version. Only applies if api_version set to 'auto'
 
     Note:
         Configuration parameters are described in more detail at
@@ -251,6 +254,7 @@ class KafkaProducer(object):
         'ssl_keyfile': None,
         'ssl_crlfile': None,
         'api_version': 'auto',
+        'api_version_auto_timeout_ms': 2000
     }
 
     def __init__(self, **configs):
@@ -274,7 +278,7 @@ def __init__(self, **configs):
 
         # Check Broker Version if not set explicitly
         if self.config['api_version'] == 'auto':
-            self.config['api_version'] = client.check_version()
+            self.config['api_version'] = client.check_version(timeout=(self.config['api_version_auto_timeout_ms']/1000))
         assert self.config['api_version'] in ('0.10', '0.9', '0.8.2', '0.8.1', '0.8.0')
 
         # Convert api_version config to tuple for easy comparisons