Use api_version() to select MetadataRequest from client

Author: dpkp

kafka/client_async.py

@@ -868,9 +868,9 @@ def _maybe_refresh_metadata(self, wakeup=False):
             if not topics and self.cluster.is_bootstrap(node_id):
                 topics = list(self.config['bootstrap_topics_filter'])
 
+            api_version = self.api_version(MetadataRequest, max_version=1)
             if self.cluster.need_all_topic_metadata or not topics:
-                topics = [] if self.config['api_version'] < (0, 10, 0) else None
-            api_version = 0 if self.config['api_version'] < (0, 10, 0) else 1
+                topics = MetadataRequest[api_version].ALL_TOPICS
             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)

kafka/protocol/metadata.py

@@ -135,34 +135,34 @@ class MetadataRequest_v0(Request):
     SCHEMA = Schema(
         ('topics', Array(String('utf-8')))
     )
-    ALL_TOPICS = None  # Empty Array (len 0) for topics returns all topics
+    ALL_TOPICS = [] # Empty Array (len 0) for topics returns all topics
 
 
 class MetadataRequest_v1(Request):
     API_KEY = 3
     API_VERSION = 1
     RESPONSE_TYPE = MetadataResponse_v1
     SCHEMA = MetadataRequest_v0.SCHEMA
-    ALL_TOPICS = -1  # Null Array (len -1) for topics returns all topics
-    NO_TOPICS = None  # Empty array (len 0) for topics returns no topics
+    ALL_TOPICS = None # Null Array (len -1) for topics returns all topics
+    NO_TOPICS = [] # Empty array (len 0) for topics returns no topics
 
 
 class MetadataRequest_v2(Request):
     API_KEY = 3
     API_VERSION = 2
     RESPONSE_TYPE = MetadataResponse_v2
     SCHEMA = MetadataRequest_v1.SCHEMA
-    ALL_TOPICS = -1  # Null Array (len -1) for topics returns all topics
-    NO_TOPICS = None  # Empty array (len 0) for topics returns no topics
+    ALL_TOPICS = None
+    NO_TOPICS = []
 
 
 class MetadataRequest_v3(Request):
     API_KEY = 3
     API_VERSION = 3
     RESPONSE_TYPE = MetadataResponse_v3
     SCHEMA = MetadataRequest_v1.SCHEMA
-    ALL_TOPICS = -1  # Null Array (len -1) for topics returns all topics
-    NO_TOPICS = None  # Empty array (len 0) for topics returns no topics
+    ALL_TOPICS = None
+    NO_TOPICS = []
 
 
 class MetadataRequest_v4(Request):
@@ -173,8 +173,8 @@ class MetadataRequest_v4(Request):
         ('topics', Array(String('utf-8'))),
         ('allow_auto_topic_creation', Boolean)
     )
-    ALL_TOPICS = -1  # Null Array (len -1) for topics returns all topics
-    NO_TOPICS = None  # Empty array (len 0) for topics returns no topics
+    ALL_TOPICS = None
+    NO_TOPICS = []
 
 
 class MetadataRequest_v5(Request):
@@ -186,8 +186,8 @@ class MetadataRequest_v5(Request):
     API_VERSION = 5
     RESPONSE_TYPE = MetadataResponse_v5
     SCHEMA = MetadataRequest_v4.SCHEMA
-    ALL_TOPICS = -1  # Null Array (len -1) for topics returns all topics
-    NO_TOPICS = None  # Empty array (len 0) for topics returns no topics
+    ALL_TOPICS = None
+    NO_TOPICS = []
 
 
 MetadataRequest = [