diff options
| author | eastlondoner <eastlondoner@users.noreply.github.com> | 2016-06-29 18:49:28 +0100 |
|---|---|---|
| committer | Dana Powers <dana.powers@gmail.com> | 2016-06-29 10:49:28 -0700 |
| commit | db47136671a05283d9801c5a3ec74b3e0f38004e (patch) | |
| tree | e5b740a2c4eeb195a95cd1af9aea4ccfa24b87d4 | |
| parent | 229ac5d1c989d14bae3591c8b1fb1a93dc5e4b1c (diff) | |
| download | kafka-python-db47136671a05283d9801c5a3ec74b3e0f38004e.tar.gz | |
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
| -rw-r--r-- | .gitignore | 1 | ||||
| -rw-r--r-- | kafka/consumer/group.py | 6 | ||||
| -rw-r--r-- | kafka/producer/kafka.py | 6 |
3 files changed, 11 insertions, 2 deletions
@@ -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 106e96b..1e9b052 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 @@ class KafkaConsumer(six.Iterator): # 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 7c55b27..e1200f4 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 @@ class KafkaProducer(object): # 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 |