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
This commit is contained in:
eastlondoner
2016-06-29 18:49:28 +01:00
committed by Dana Powers
parent 229ac5d1c9
commit db47136671
3 changed files with 11 additions and 2 deletions

1
.gitignore vendored
View File

@@ -11,3 +11,4 @@ servers/*/resources/ssl*
.noseids
docs/_build
.cache*
.idea/

View File

@@ -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

View File

@@ -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