Do not use newly created ACL request / responses in allowed version lists, due to flexible versions enabling in kafka actually requiring a serialization protocol header update

Author: Gabriel Tincu

kafka/admin/client.py

@@ -570,7 +570,7 @@ def describe_acls(self, acl_filter):
                 operation=acl_filter.operation,
                 permission_type=acl_filter.permission_type
             )
-        elif version <= 1:
+        elif version <= 2:
             request = DescribeAclsRequest[version](
                 resource_type=acl_filter.resource_pattern.resource_type,
                 resource_name=acl_filter.resource_pattern.resource_name,
@@ -763,7 +763,7 @@ def delete_acls(self, acl_filters):
             request = DeleteAclsRequest[version](
                 filters=[self._convert_delete_acls_resource_request_v0(acl) for acl in acl_filters]
             )
-        elif version <= 1:
+        elif version <= 2:
             request = DeleteAclsRequest[version](
                 filters=[self._convert_delete_acls_resource_request_v1(acl) for acl in acl_filters]
             )

kafka/conn.py

@@ -24,7 +24,7 @@
 from kafka.future import Future
 from kafka.metrics.stats import Avg, Count, Max, Rate
 from kafka.oauth.abstract import AbstractTokenProvider
-from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest
+from kafka.protocol.admin import SaslHandShakeRequest, DescribeAclsRequest_v2
 from kafka.protocol.commit import OffsetFetchRequest
 from kafka.protocol.offset import OffsetRequest
 from kafka.protocol.produce import ProduceRequest
@@ -1169,7 +1169,7 @@ def _infer_broker_version_from_api_versions(self, api_versions):
         # in reverse order. As soon as we find one that works, return it
         test_cases = [
             # format (<broker version>, <needed struct>)
-            ((2, 5, 0), DescribeAclsRequest[2]),
+            ((2, 5, 0), DescribeAclsRequest_v2),
             ((2, 4, 0), ProduceRequest[8]),
             ((2, 3, 0), FetchRequest[11]),
             ((2, 2, 0), OffsetRequest[5]),

kafka/protocol/admin.py

@@ -523,8 +523,8 @@ class DescribeAclsRequest_v2(Request):
     SCHEMA = DescribeAclsRequest_v1.SCHEMA
 
 
-DescribeAclsRequest = [DescribeAclsRequest_v0, DescribeAclsRequest_v1, DescribeAclsRequest_v2]
-DescribeAclsResponse = [DescribeAclsResponse_v0, DescribeAclsResponse_v1, DescribeAclsResponse_v2]
+DescribeAclsRequest = [DescribeAclsRequest_v0, DescribeAclsRequest_v1]
+DescribeAclsResponse = [DescribeAclsResponse_v0, DescribeAclsResponse_v1]
 
 class CreateAclsResponse_v0(Response):
     API_KEY = 30