align api_version checks in code w/ broker_api_versions

Author: dpkp

kafka/client_async.py

@@ -242,6 +242,11 @@ def __init__(self, **configs):
             self.config['api_version'] = self.check_version(timeout=check_timeout)
         elif self.config['api_version'] in BROKER_API_VERSIONS:
             self._api_versions = BROKER_API_VERSIONS[self.config['api_version']]
+        elif (self.config['api_version'] + (0,)) in BROKER_API_VERSIONS:
+            log.warning('Configured api_version %s is ambiguous; using %s',
+                        self.config['api_version'], self.config['api_version'] + (0,))
+            self.config['api_version'] = self.config['api_version'] + (0,)
+            self._api_versions = BROKER_API_VERSIONS[self.config['api_version']]
         else:
             compatible_version = None
             for v in sorted(BROKER_API_VERSIONS.keys(), reverse=True):
@@ -864,8 +869,8 @@ def _maybe_refresh_metadata(self, wakeup=False):
                 topics = list(self.config['bootstrap_topics_filter'])
 
             if self.cluster.need_all_topic_metadata or not topics:
-                topics = [] if self.config['api_version'] < (0, 10) else None
-            api_version = 0 if self.config['api_version'] < (0, 10) else 1
+                topics = [] if self.config['api_version'] < (0, 10, 0) else None
+            api_version = 0 if self.config['api_version'] < (0, 10, 0) else 1
             request = MetadataRequest[api_version](topics)
             log.debug("Sending metadata request %s to node %s", request, node_id)
             future = self.send(node_id, request, wakeup=wakeup)
@@ -913,7 +918,7 @@ def check_version(self, node_id=None, timeout=2, strict=False):
             is down and the client enters a bootstrap backoff sleep.
             This is only possible if node_id is None.
 
-        Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), ...
+        Returns: version tuple, i.e. (3, 9), (2, 0), (0, 10, 2) etc
 
         Raises:
             NodeNotReadyError (if node_id is provided)

kafka/conn.py

@@ -165,8 +165,8 @@ class BrokerConnection(object):
             or other configuration forbids use of all the specified ciphers),
             an ssl.SSLError will be raised. See ssl.SSLContext.set_ciphers
         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). Default: (0, 8, 2)
+            Must be None or >= (0, 10, 0) to enable SASL authentication.
+            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
@@ -214,7 +214,7 @@ class BrokerConnection(object):
         'ssl_crlfile': None,
         'ssl_password': None,
         'ssl_ciphers': None,
-        'api_version': (0, 8, 2),  # default to most restrictive
+        'api_version': None,
         'selector': selectors.DefaultSelector,
         'state_change_callback': lambda node_id, sock, conn: True,
         'metrics': None,
@@ -522,7 +522,7 @@ def _try_handshake(self):
         return False
 
     def _try_authenticate(self):
-        assert self.config['api_version'] is None or self.config['api_version'] >= (0, 10)
+        assert self.config['api_version'] is None or self.config['api_version'] >= (0, 10, 0)
 
         if self._sasl_auth_future is None:
             # Build a SaslHandShakeRequest message
@@ -1178,16 +1178,16 @@ def _infer_broker_version_from_api_versions(self, api_versions):
         test_cases = [
             # format (<broker version>, <needed struct>)
             # Make sure to update consumer_integration test check when adding newer versions.
-            ((2, 6, 0), DescribeClientQuotasRequest[0]),
-            ((2, 5, 0), DescribeAclsRequest_v2),
-            ((2, 4, 0), ProduceRequest[8]),
-            ((2, 3, 0), FetchRequest[11]),
-            ((2, 2, 0), OffsetRequest[5]),
-            ((2, 1, 0), FetchRequest[10]),
-            ((2, 0, 0), FetchRequest[8]),
-            ((1, 1, 0), FetchRequest[7]),
-            ((1, 0, 0), MetadataRequest[5]),
-            ((0, 11, 0), MetadataRequest[4]),
+            ((2, 6), DescribeClientQuotasRequest[0]),
+            ((2, 5), DescribeAclsRequest_v2),
+            ((2, 4), ProduceRequest[8]),
+            ((2, 3), FetchRequest[11]),
+            ((2, 2), OffsetRequest[5]),
+            ((2, 1), FetchRequest[10]),
+            ((2, 0), FetchRequest[8]),
+            ((1, 1), FetchRequest[7]),
+            ((1, 0), MetadataRequest[5]),
+            ((0, 11), MetadataRequest[4]),
             ((0, 10, 2), OffsetFetchRequest[2]),
             ((0, 10, 1), MetadataRequest[2]),
         ]
@@ -1209,7 +1209,7 @@ def check_version(self, timeout=2, strict=False, topics=[]):
 
         Note: This is a blocking call.
 
-        Returns: version tuple, i.e. (0, 10), (0, 9), (0, 8, 2), ...
+        Returns: version tuple, i.e. (3, 9), (2, 4), etc ...
         """
         timeout_at = time.time() + timeout
         log.info('Probing node %s broker version', self.node_id)
@@ -1240,7 +1240,7 @@ def reset_override_configs():
         test_cases = [
             # All cases starting from 0.10 will be based on ApiVersionsResponse
             ((0, 11), ApiVersionsRequest[1]()),
-            ((0, 10), ApiVersionsRequest[0]()),
+            ((0, 10, 0), ApiVersionsRequest[0]()),
             ((0, 9), ListGroupsRequest[0]()),
             ((0, 8, 2), GroupCoordinatorRequest[0]('kafka-python-default-group')),
             ((0, 8, 1), OffsetFetchRequest[0]('kafka-python-default-group', [])),
@@ -1273,7 +1273,7 @@ def reset_override_configs():
                 selector.close()
 
             if f.succeeded():
-                if version >= (0, 10):
+                if version >= (0, 10, 0):
                     # Starting from 0.10 kafka broker we determine version
                     # by looking at ApiVersionsResponse
                     api_versions = self._handle_api_versions_response(f.value)
@@ -1303,7 +1303,7 @@ def reset_override_configs():
                 # requests (bug...). In this case we expect to see a correlation
                 # id mismatch
                 elif (isinstance(f.exception, Errors.CorrelationIdError) and
-                      version == (0, 10)):
+                      version > (0, 9)):
                     pass
                 elif six.PY2:
                     assert isinstance(f.exception.args[0], socket.error)

kafka/consumer/fetcher.py

@@ -702,11 +702,11 @@ def _create_fetch_requests(self):
                 log.log(0, "Skipping fetch for partition %s because there is an inflight request to node %s",
                         partition, node_id)
 
-        if self.config['api_version'] >= (0, 11, 0):
+        if self.config['api_version'] >= (0, 11):
             version = 4
         elif self.config['api_version'] >= (0, 10, 1):
             version = 3
-        elif self.config['api_version'] >= (0, 10):
+        elif self.config['api_version'] >= (0, 10, 0):
             version = 2
         elif self.config['api_version'] == (0, 9):
             version = 1

kafka/coordinator/base.py

@@ -461,7 +461,7 @@ def _send_join_group_request(self):
                 self._generation.member_id,
                 self.protocol_type(),
                 member_metadata)
-        elif (0, 10, 1) <= self.config['api_version'] < (0, 11, 0):
+        elif (0, 10, 1) <= self.config['api_version'] < (0, 11):
             request = JoinGroupRequest[1](
                 self.group_id,
                 self.config['session_timeout_ms'],
@@ -562,7 +562,7 @@ def _handle_join_group_response(self, future, send_time, response):
 
     def _on_join_follower(self):
         # send follower's sync group with an empty assignment
-        version = 0 if self.config['api_version'] < (0, 11, 0) else 1
+        version = 0 if self.config['api_version'] < (0, 11) else 1
         request = SyncGroupRequest[version](
             self.group_id,
             self._generation.generation_id,
@@ -590,7 +590,7 @@ def _on_join_leader(self, response):
         except Exception as e:
             return Future().failure(e)
 
-        version = 0 if self.config['api_version'] < (0, 11, 0) else 1
+        version = 0 if self.config['api_version'] < (0, 11) else 1
         request = SyncGroupRequest[version](
             self.group_id,
             self._generation.generation_id,
@@ -771,7 +771,7 @@ def maybe_leave_group(self):
                 # this is a minimal effort attempt to leave the group. we do not
                 # attempt any resending if the request fails or times out.
                 log.info('Leaving consumer group (%s).', self.group_id)
-                version = 0 if self.config['api_version'] < (0, 11, 0) else 1
+                version = 0 if self.config['api_version'] < (0, 11) else 1
                 request = LeaveGroupRequest[version](self.group_id, self._generation.member_id)
                 future = self._client.send(self.coordinator_id, request)
                 future.add_callback(self._handle_leave_group_response)
@@ -799,7 +799,7 @@ def _send_heartbeat_request(self):
             e = Errors.NodeNotReadyError(self.coordinator_id)
             return Future().failure(e)
 
-        version = 0 if self.config['api_version'] < (0, 11, 0) else 1
+        version = 0 if self.config['api_version'] < (0, 11) else 1
         request = HeartbeatRequest[version](self.group_id,
                                             self._generation.generation_id,
                                             self._generation.member_id)

kafka/producer/kafka.py

@@ -393,7 +393,7 @@ def __init__(self, **configs):
             assert self.config['api_version'] >= (0, 8, 2), 'LZ4 Requires >= Kafka 0.8.2 Brokers'
 
         if self.config['compression_type'] == 'zstd':
-            assert self.config['api_version'] >= (2, 1, 0), 'Zstd Requires >= Kafka 2.1.0 Brokers'
+            assert self.config['api_version'] >= (2, 1), 'Zstd Requires >= Kafka 2.1 Brokers'
 
         # Check compression_type for library support
         ct = self.config['compression_type']
@@ -524,7 +524,7 @@ def partitions_for(self, topic):
     def _max_usable_produce_magic(self):
         if self.config['api_version'] >= (0, 11):
             return 2
-        elif self.config['api_version'] >= (0, 10):
+        elif self.config['api_version'] >= (0, 10, 0):
             return 1
         else:
             return 0

kafka/producer/sender.py

@@ -313,7 +313,7 @@ def _produce_request(self, node_id, acks, timeout, batches):
         elif self.config['api_version'] >= (0, 11):
             version = 3
             kwargs = dict(transactional_id=None)
-        elif self.config['api_version'] >= (0, 10):
+        elif self.config['api_version'] >= (0, 10, 0):
             version = 2
         elif self.config['api_version'] == (0, 9):
             version = 1

test/test_client_async.py

@@ -266,7 +266,7 @@ def test_least_loaded_node():
 def test_set_topics(mocker):
     request_update = mocker.patch.object(ClusterMetadata, 'request_update')
     request_update.side_effect = lambda: Future()
-    cli = KafkaClient(api_version=(0, 10))
+    cli = KafkaClient(api_version=(0, 10, 0))
 
     # replace 'empty' with 'non empty'
     request_update.reset_mock()

test/test_consumer.py

@@ -18,7 +18,7 @@ def test_request_timeout_larger_than_connections_max_idle_ms_raises(self):
             KafkaConsumer(bootstrap_servers='localhost:9092', api_version=(0, 9), request_timeout_ms=50000, connections_max_idle_ms=40000)
 
     def test_subscription_copy(self):
-        consumer = KafkaConsumer('foo', api_version=(0, 10))
+        consumer = KafkaConsumer('foo', api_version=(0, 10, 0))
         sub = consumer.subscription()
         assert sub is not consumer.subscription()
         assert sub == set(['foo'])