kafka.admin: use new protocol attrs

Author: dpkp

kafka/admin/client.py

@@ -471,13 +471,13 @@ def create_topics(self, new_topics, timeout_ms=None, validate_only=False, raise_
                     "validate_only requires CreateTopicsRequest >= v1, which is not supported by Kafka {}."
                     .format(self.config['api_version']))
             request = CreateTopicsRequest[version](
-                create_topic_requests=[self._convert_new_topic_request(new_topic) for new_topic in new_topics],
-                timeout=timeout_ms
+                topics=[self._convert_new_topic_request(new_topic) for new_topic in new_topics],
+                timeout_ms=timeout_ms
             )
         elif version <= 3:
             request = CreateTopicsRequest[version](
-                create_topic_requests=[self._convert_new_topic_request(new_topic) for new_topic in new_topics],
-                timeout=timeout_ms,
+                topics=[self._convert_new_topic_request(new_topic) for new_topic in new_topics],
+                timeout_ms=timeout_ms,
                 validate_only=validate_only
             )
         else:
@@ -504,7 +504,7 @@ def delete_topics(self, topics, timeout_ms=None, raise_errors=True):
         """
         version = self._client.api_version(DeleteTopicsRequest, max_version=3)
         timeout_ms = self._validate_timeout(timeout_ms)
-        request = DeleteTopicsRequest[version](topics=topics, timeout=timeout_ms)
+        request = DeleteTopicsRequest[version](topics=topics, timeout_ms=timeout_ms)
         def get_response_errors(r):
             for response in r.responses:
                 yield Errors.for_code(response[1])
@@ -640,19 +640,19 @@ def describe_acls(self, acl_filter):
         version = self._client.api_version(DescribeAclsRequest, max_version=1)
         if version == 0:
             request = DescribeAclsRequest[version](
-                resource_type=acl_filter.resource_pattern.resource_type,
-                resource_name=acl_filter.resource_pattern.resource_name,
-                principal=acl_filter.principal,
-                host=acl_filter.host,
+                resource_type_filter=acl_filter.resource_pattern.resource_type,
+                resource_name_filter=acl_filter.resource_pattern.resource_name,
+                principal_filter=acl_filter.principal,
+                host_filter=acl_filter.host,
                 operation=acl_filter.operation,
                 permission_type=acl_filter.permission_type
             )
         elif version <= 1:
             request = DescribeAclsRequest[version](
-                resource_type=acl_filter.resource_pattern.resource_type,
-                resource_name=acl_filter.resource_pattern.resource_name,
+                resource_type_filter=acl_filter.resource_pattern.resource_type,
+                resource_name_filter=acl_filter.resource_pattern.resource_name,
                 resource_pattern_type_filter=acl_filter.resource_pattern.pattern_type,
-                principal=acl_filter.principal,
+                principal_filter=acl_filter.principal,
                 host=acl_filter.host,
                 operation=acl_filter.operation,
                 permission_type=acl_filter.permission_type
@@ -727,7 +727,7 @@ def _convert_create_acls_response_to_acls(acls, create_response):
 
         creations_error = []
         creations_success = []
-        for i, creations in enumerate(create_response.creation_responses):
+        for i, creations in enumerate(create_response.results):
             if version <= 1:
                 error_code, error_message = creations
                 acl = acls[i]
@@ -827,7 +827,7 @@ def _convert_delete_acls_response_to_matching_acls(acl_filters, delete_response)
         """
         version = delete_response.API_VERSION
         filter_result_list = []
-        for i, filter_responses in enumerate(delete_response.filter_responses):
+        for i, filter_responses in enumerate(delete_response.filter_results):
             filter_error_code, filter_error_message, matching_acls = filter_responses
             filter_error = Errors.for_code(filter_error_code)
             acl_result_list = []
@@ -1054,8 +1054,8 @@ def create_partitions(self, topic_partitions, timeout_ms=None, validate_only=Fal
         version = self._client.api_version(CreatePartitionsRequest, max_version=1)
         timeout_ms = self._validate_timeout(timeout_ms)
         request = CreatePartitionsRequest[version](
-            topic_partitions=[self._convert_create_partitions_request(topic_name, new_partitions) for topic_name, new_partitions in topic_partitions.items()],
-            timeout=timeout_ms,
+            topics=[self._convert_create_partitions_request(topic_name, new_partitions) for topic_name, new_partitions in topic_partitions.items()],
+            timeout_ms=timeout_ms,
             validate_only=validate_only
         )
         def get_response_errors(r):
@@ -1594,7 +1594,7 @@ def perform_leader_election(self, election_type, topic_partitions=None, timeout_
         request = ElectLeadersRequest[version](
             election_type=ElectionType(election_type),
             topic_partitions=self._get_topic_partitions(topic_partitions),
-            timeout=timeout_ms,
+            timeout_ms=timeout_ms,
         )
         # TODO convert structs to a more pythonic interface
         def get_response_errors(r):

test/integration/fixtures.py

@@ -614,7 +614,7 @@ def _create_topic_via_admin_api(self, topic_name, num_partitions, replication_fa
         request = CreateTopicsRequest[0]([(topic_name, num_partitions,
                                            replication_factor, [], [])], timeout_ms)
         response = self._send_request(request, timeout=timeout_ms)
-        for topic_result in response.topic_errors:
+        for topic_result in response.topics:
             error_code = topic_result[1]
             if error_code != 0:
                 raise errors.for_code(error_code)

test/integration/test_admin_integration.py

@@ -103,9 +103,9 @@ def test_describe_configs_broker_resource_returns_configs(kafka_admin_client):
     configs = kafka_admin_client.describe_configs([ConfigResource(ConfigResourceType.BROKER, broker_id)])
 
     assert len(configs) == 1
-    assert configs[0].resources[0][2] == ConfigResourceType.BROKER
-    assert configs[0].resources[0][3] == str(broker_id)
-    assert len(configs[0].resources[0][4]) > 1
+    assert configs[0].results[0][2] == ConfigResourceType.BROKER
+    assert configs[0].results[0][3] == str(broker_id)
+    assert len(configs[0].results[0][4]) > 1
 
 
 @pytest.mark.xfail(condition=True,
@@ -118,9 +118,9 @@ def test_describe_configs_topic_resource_returns_configs(topic, kafka_admin_clie
     configs = kafka_admin_client.describe_configs([ConfigResource(ConfigResourceType.TOPIC, topic)])
 
     assert len(configs) == 1
-    assert configs[0].resources[0][2] == ConfigResourceType.TOPIC
-    assert configs[0].resources[0][3] == topic
-    assert len(configs[0].resources[0][4]) > 1
+    assert configs[0].results[0][2] == ConfigResourceType.TOPIC
+    assert configs[0].results[0][3] == topic
+    assert len(configs[0].results[0][4]) > 1
 
 
 @pytest.mark.skipif(env_kafka_version() < (0, 11), reason="Describe config features require broker >=0.11")
@@ -135,11 +135,11 @@ def test_describe_configs_mixed_resources_returns_configs(topic, kafka_admin_cli
     assert len(configs) == 2
 
     for config in configs:
-        assert (config.resources[0][2] == ConfigResourceType.TOPIC
-                and config.resources[0][3] == topic) or \
-               (config.resources[0][2] == ConfigResourceType.BROKER
-                and config.resources[0][3] == str(broker_id))
-        assert len(config.resources[0][4]) > 1
+        assert (config.results[0][2] == ConfigResourceType.TOPIC
+                and config.results[0][3] == topic) or \
+               (config.results[0][2] == ConfigResourceType.BROKER
+                and config.results[0][3] == str(broker_id))
+        assert len(config.results[0][4]) > 1
 
 
 @pytest.mark.skipif(env_kafka_version() < (0, 11), reason="Describe config features require broker >=0.11")