diff options
Diffstat (limited to 'kafka')
| -rw-r--r-- | kafka/client_async.py | 30 | ||||
| -rw-r--r-- | kafka/conn.py | 9 | ||||
| -rw-r--r-- | kafka/consumer/group.py | 44 | ||||
| -rw-r--r-- | kafka/producer/kafka.py | 30 |
4 files changed, 81 insertions, 32 deletions
diff --git a/kafka/client_async.py b/kafka/client_async.py index 8839dee..6fa9434 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -1,4 +1,4 @@ -from __future__ import absolute_import +from __future__ import absolute_import, division import copy import functools @@ -61,7 +61,16 @@ class KafkaClient(object): 'ssl_keyfile': None, 'ssl_password': None, 'ssl_crlfile': None, + 'api_version': None, + 'api_version_auto_timeout_ms': 2000, } + API_VERSIONS = [ + (0, 10), + (0, 9), + (0, 8, 2), + (0, 8, 1), + (0, 8, 0) + ] def __init__(self, **configs): """Initialize an asynchronous kafka client @@ -118,12 +127,24 @@ class KafkaClient(object): providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. default: none. + api_version (tuple): specify which kafka API version to use. Accepted + values are: (0, 8, 0), (0, 8, 1), (0, 8, 2), (0, 9), (0, 10) + If None, KafkaClient will attempt to infer the broker + version by probing various APIs. Default: None + 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 is None """ self.config = copy.copy(self.DEFAULT_CONFIG) for key in self.config: if key in configs: self.config[key] = configs[key] + if self.config['api_version'] is not None: + assert self.config['api_version'] in self.API_VERSIONS, ( + 'api_version [{}] must be one of: {}'.format( + self.config['api_version'], str(self.API_VERSIONS))) + self.cluster = ClusterMetadata(**self.config) self._topics = set() # empty set will fetch all topic metadata self._metadata_refresh_in_progress = False @@ -141,6 +162,11 @@ class KafkaClient(object): self._closed = False self._bootstrap(collect_hosts(self.config['bootstrap_servers'])) + # Check Broker Version if not set explicitly + if self.config['api_version'] is None: + check_timeout = self.config['api_version_auto_timeout_ms'] / 1000 + self.config['api_version'] = self.check_version(timeout=check_timeout) + def _bootstrap(self, hosts): # Exponential backoff if bootstrap fails backoff_ms = self.config['reconnect_backoff_ms'] * 2 ** self._bootstrap_fails @@ -683,7 +709,7 @@ class KafkaClient(object): is down and the client enters a bootstrap backoff sleep. This is only possible if node_id is None. - Returns: version str, i.e. '0.10', '0.9', '0.8.2', '0.8.1', '0.8.0' + Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), ... Raises: NodeNotReadyError (if node_id is provided) diff --git a/kafka/conn.py b/kafka/conn.py index 0418bc5..6028867 100644 --- a/kafka/conn.py +++ b/kafka/conn.py @@ -541,7 +541,12 @@ class BrokerConnection(object): return self._correlation_id def check_version(self, timeout=2, strict=False): - """Attempt to guess the broker version. This is a blocking call.""" + """Attempt to guess the broker version. + + Note: This is a blocking call. + + Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), ... + """ # Monkeypatch the connection request timeout # Generally this timeout should not get triggered @@ -643,7 +648,7 @@ class BrokerConnection(object): log.removeFilter(log_filter) self.config['request_timeout_ms'] = stashed_request_timeout_ms - return version + return tuple(map(int, version.split('.'))) def __repr__(self): return "<BrokerConnection host=%s/%s port=%d>" % (self.hostname, self.host, diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index 7fe509a..8fa43bc 100644 --- a/kafka/consumer/group.py +++ b/kafka/consumer/group.py @@ -150,12 +150,19 @@ class KafkaConsumer(six.Iterator): providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. default: none. - api_version (str): specify which kafka API version to use. - 0.9 enables full group coordination features; 0.8.2 enables - kafka-storage offset commits; 0.8.1 enables zookeeper-storage - 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 (tuple): specify which kafka API version to use. + If set to None, the client will attempt to infer the broker version + by probing various APIs. Default: None + Examples: + (0, 9) enables full group coordination features with automatic + partition assignment and rebalancing, + (0, 8, 2) enables kafka-storage offset commits with manual + partition assignment only, + (0, 8, 1) enables zookeeper-storage offset commits with manual + partition assignment only, + (0, 8, 0) enables basic functionality but requires manual + partition assignment and offset management. + For a full list of supported versions, see KafkaClient.API_VERSIONS 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' @@ -205,7 +212,7 @@ class KafkaConsumer(six.Iterator): 'ssl_keyfile': None, 'ssl_crlfile': None, 'ssl_password': None, - 'api_version': 'auto', + 'api_version': None, 'api_version_auto_timeout_ms': 2000, 'connections_max_idle_ms': 9 * 60 * 1000, # not implemented yet 'metric_reporters': [], @@ -222,7 +229,7 @@ class KafkaConsumer(six.Iterator): # Only check for extra config keys in top-level class assert not configs, 'Unrecognized configs: %s' % configs - deprecated = {'smallest': 'earliest', 'largest': 'latest' } + deprecated = {'smallest': 'earliest', 'largest': 'latest'} if self.config['auto_offset_reset'] in deprecated: new_config = deprecated[self.config['auto_offset_reset']] log.warning('use auto_offset_reset=%s (%s is deprecated)', @@ -239,16 +246,21 @@ class KafkaConsumer(six.Iterator): metric_group_prefix = 'consumer' # TODO _metrics likely needs to be passed to KafkaClient, etc. - self._client = KafkaClient(**self.config) + # api_version was previously a str. accept old format for now + if isinstance(self.config['api_version'], str): + str_version = self.config['api_version'] + if str_version == 'auto': + self.config['api_version'] = None + else: + self.config['api_version'] = tuple(map(int, str_version.split('.'))) + log.warning('use api_version=%s (%s is deprecated)', + str(self.config['api_version']), str_version) - # Check Broker Version if not set explicitly - if self.config['api_version'] == 'auto': - 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' + self._client = KafkaClient(**self.config) - # Convert api_version config to tuple for easy comparisons - self.config['api_version'] = tuple( - map(int, self.config['api_version'].split('.'))) + # Get auto-discovered version from client if necessary + if self.config['api_version'] is None: + self.config['api_version'] = self._client.config['api_version'] self._subscription = SubscriptionState(self.config['auto_offset_reset']) self._fetcher = Fetcher( diff --git a/kafka/producer/kafka.py b/kafka/producer/kafka.py index 63cff73..f5c5d19 100644 --- a/kafka/producer/kafka.py +++ b/kafka/producer/kafka.py @@ -213,9 +213,10 @@ class KafkaProducer(object): providing a file, only the leaf certificate will be checked against this CRL. The CRL can only be checked with Python 3.4+ or 2.7.9+. default: none. - 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 (tuple): specify which kafka API version to use. + For a full list of supported versions, see KafkaClient.API_VERSIONS + If set to None, the client will attempt to infer the broker version + by probing various APIs. Default: None 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' @@ -253,7 +254,7 @@ class KafkaProducer(object): 'ssl_certfile': None, 'ssl_keyfile': None, 'ssl_crlfile': None, - 'api_version': 'auto', + 'api_version': None, 'api_version_auto_timeout_ms': 2000 } @@ -274,16 +275,21 @@ class KafkaProducer(object): if self.config['acks'] == 'all': self.config['acks'] = -1 - client = KafkaClient(**self.config) + # api_version was previously a str. accept old format for now + if isinstance(self.config['api_version'], str): + deprecated = self.config['api_version'] + if deprecated == 'auto': + self.config['api_version'] = None + else: + self.config['api_version'] = tuple(map(int, deprecated.split('.'))) + log.warning('use api_version=%s (%s is deprecated)', + str(self.config['api_version']), deprecated) - # Check Broker Version if not set explicitly - if self.config['api_version'] == 'auto': - 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') + client = KafkaClient(**self.config) - # Convert api_version config to tuple for easy comparisons - self.config['api_version'] = tuple( - map(int, self.config['api_version'].split('.'))) + # Get auto-discovered version from client if necessary + if self.config['api_version'] is None: + self.config['api_version'] = client.config['api_version'] if self.config['compression_type'] == 'lz4': assert self.config['api_version'] >= (0, 8, 2), 'LZ4 Requires >= Kafka 0.8.2 Brokers' |