Add more admin apis to kafka.protocol.new

kafka/protocol/new/admin/__init__.py

@@ -1,4 +1,5 @@
 from .acl import *
+from .client_quotas import *
 from .cluster import *
 from .groups import *
 from .topics import *
@@ -10,6 +11,9 @@
     'ACLResourceType', 'ACLOperation',
     'ACLPermissionType', 'ACLResourcePatternType',
 
+    'AlterClientQuotasRequest', 'AlterClientQuotasResponse',
+    'DescribeClientQuotasRequest', 'DescribeClientQuotasResponse',
+
     'DescribeClusterRequest', 'DescribeClusterResponse',
     'DescribeConfigsRequest', 'DescribeConfigsResponse',
     'AlterConfigsRequest', 'AlterConfigsResponse',
@@ -23,5 +27,8 @@
     'CreateTopicsRequest', 'CreateTopicsResponse',
     'DeleteTopicsRequest', 'DeleteTopicsResponse',
     'CreatePartitionsRequest', 'CreatePartitionsResponse',
+    'AlterPartitionRequest', 'AlterPartitionResponse',
+    'AlterPartitionReassignmentsRequest', 'AlterPartitionReassignmentsResponse',
+    'ListPartitionReassignmentsRequest', 'ListPartitionReassignmentsResponse',
     'DeleteRecordsRequest', 'DeleteRecordsResponse',
 ]

kafka/protocol/new/admin/client_quotas.py

@@ -0,0 +1,14 @@
+from ..api_message import ApiMessage
+
+
+class AlterClientQuotasRequest(ApiMessage): pass
+class AlterClientQuotasResponse(ApiMessage): pass
+
+class DescribeClientQuotasRequest(ApiMessage): pass
+class DescribeClientQuotasResponse(ApiMessage): pass
+
+
+__all__ = [
+    'AlterClientQuotasRequest', 'AlterClientQuotasResponse',
+    'DescribeClientQuotasRequest', 'DescribeClientQuotasResponse',
+]

kafka/protocol/new/admin/cluster.py

@@ -19,9 +19,7 @@ class ElectLeadersRequest(ApiMessage): pass
 class ElectLeadersResponse(ApiMessage): pass
 
 class ElectionType(IntEnum):
-    """ Leader election type
-    """
-
+    """Leader election type"""
     PREFERRED = 0,
     UNCLEAN = 1
 

kafka/protocol/new/admin/topics.py

@@ -10,6 +10,15 @@ class DeleteTopicsResponse(ApiMessage): pass
 class CreatePartitionsRequest(ApiMessage): pass
 class CreatePartitionsResponse(ApiMessage): pass
 
+class AlterPartitionRequest(ApiMessage): pass
+class AlterPartitionResponse(ApiMessage): pass
+
+class AlterPartitionReassignmentsRequest(ApiMessage): pass
+class AlterPartitionReassignmentsResponse(ApiMessage): pass
+
+class ListPartitionReassignmentsRequest(ApiMessage): pass
+class ListPartitionReassignmentsResponse(ApiMessage): pass
+
 class DeleteRecordsRequest(ApiMessage): pass
 class DeleteRecordsResponse(ApiMessage): pass
 
@@ -18,5 +27,8 @@ class DeleteRecordsResponse(ApiMessage): pass
     'CreateTopicsRequest', 'CreateTopicsResponse',
     'DeleteTopicsRequest', 'DeleteTopicsResponse',
     'CreatePartitionsRequest', 'CreatePartitionsResponse',
+    'AlterPartitionRequest', 'AlterPartitionResponse',
+    'AlterPartitionReassignmentsRequest', 'AlterPartitionReassignmentsResponse',
+    'ListPartitionReassignmentsRequest', 'ListPartitionReassignmentsResponse',
     'DeleteRecordsRequest', 'DeleteRecordsResponse',
 ]

kafka/protocol/new/schemas/resources/AlterClientQuotasRequest.json

@@ -0,0 +1,47 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 49,
+  "type": "request",
+  "listeners": ["broker", "controller"],
+  "name": "AlterClientQuotasRequest",
+  "validVersions": "0-1",
+  // Version 1 enables flexible versions.
+  "flexibleVersions": "1+",
+  "fields": [
+    { "name": "Entries", "type": "[]EntryData", "versions": "0+",
+      "about": "The quota configuration entries to alter.", "fields": [
+      { "name": "Entity", "type": "[]EntityData", "versions": "0+",
+        "about": "The quota entity to alter.", "fields": [
+        { "name": "EntityType", "type": "string", "versions": "0+",
+          "about": "The entity type." },
+        { "name": "EntityName", "type": "string", "versions": "0+", "nullableVersions": "0+",
+          "about": "The name of the entity, or null if the default." }
+      ]},
+      { "name": "Ops", "type": "[]OpData", "versions": "0+",
+        "about": "An individual quota configuration entry to alter.", "fields": [
+        { "name": "Key", "type": "string", "versions": "0+",
+          "about": "The quota configuration key." },
+        { "name": "Value", "type": "float64", "versions": "0+",
+          "about": "The value to set, otherwise ignored if the value is to be removed." },
+        { "name": "Remove", "type": "bool", "versions": "0+",
+          "about": "Whether the quota configuration value should be removed, otherwise set." }
+      ]}
+    ]},
+    { "name": "ValidateOnly", "type": "bool", "versions": "0+",
+      "about": "Whether the alteration should be validated, but not performed." }
+  ]
+}

kafka/protocol/new/schemas/resources/AlterClientQuotasResponse.json

@@ -0,0 +1,41 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 49,
+  "type": "response",
+  "name": "AlterClientQuotasResponse",
+  // Version 1 enables flexible versions.
+  "validVersions": "0-1",
+  "flexibleVersions": "1+",
+  "fields": [
+    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
+      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+    { "name": "Entries", "type": "[]EntryData", "versions": "0+",
+      "about": "The quota configuration entries to alter.", "fields": [
+      { "name": "ErrorCode", "type": "int16", "versions": "0+",
+        "about": "The error code, or `0` if the quota alteration succeeded." },
+      { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
+        "about": "The error message, or `null` if the quota alteration succeeded." },
+      { "name": "Entity", "type": "[]EntityData", "versions": "0+",
+        "about": "The quota entity to alter.", "fields": [
+        { "name": "EntityType", "type": "string", "versions": "0+",
+          "about": "The entity type." },
+        { "name": "EntityName", "type": "string", "versions": "0+", "nullableVersions": "0+",
+          "about": "The name of the entity, or null if the default." }
+      ]}
+    ]}
+  ]
+}

kafka/protocol/new/schemas/resources/AlterPartitionReassignmentsRequest.json

@@ -0,0 +1,39 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 45,
+  "type": "request",
+  "listeners": ["broker", "controller"],
+  "name": "AlterPartitionReassignmentsRequest",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "TimeoutMs", "type": "int32", "versions": "0+", "default": "60000",
+      "about": "The time in ms to wait for the request to complete." },
+    { "name": "Topics", "type": "[]ReassignableTopic", "versions": "0+",
+      "about": "The topics to reassign.", "fields": [
+      { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+        "about": "The topic name." },
+      { "name": "Partitions", "type": "[]ReassignablePartition", "versions": "0+",
+        "about": "The partitions to reassign.", "fields": [
+        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+          "about": "The partition index." },
+        { "name": "Replicas", "type": "[]int32", "versions": "0+", "nullableVersions": "0+", "default": "null", "entityType": "brokerId",
+          "about": "The replicas to place the partitions on, or null to cancel a pending reassignment for this partition." }
+      ]}
+    ]}
+  ]
+}

kafka/protocol/new/schemas/resources/AlterPartitionReassignmentsResponse.json

@@ -0,0 +1,44 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 45,
+  "type": "response",
+  "name": "AlterPartitionReassignmentsResponse",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
+      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+    { "name": "ErrorCode", "type": "int16", "versions": "0+",
+      "about": "The top-level error code, or 0 if there was no error." },
+    { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
+      "about": "The top-level error message, or null if there was no error." },
+    { "name": "Responses", "type": "[]ReassignableTopicResponse", "versions": "0+",
+      "about": "The responses to topics to reassign.", "fields": [
+      { "name": "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+        "about": "The topic name." },
+      { "name": "Partitions", "type": "[]ReassignablePartitionResponse", "versions": "0+",
+        "about": "The responses to partitions to reassign.", "fields": [
+        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+          "about": "The partition index." },
+        { "name": "ErrorCode", "type": "int16", "versions": "0+",
+          "about": "The error code for this partition, or 0 if there was no error." },
+        { "name": "ErrorMessage", "type": "string", "versions": "0+", "nullableVersions": "0+",
+          "about": "The error message for this partition, or null if there was no error." }
+      ]}
+    ]}
+  ]
+}

kafka/protocol/new/schemas/resources/AlterPartitionRequest.json

@@ -0,0 +1,60 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implie
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 56,
+  "type": "request",
+  "listeners": ["controller"],
+  "name": "AlterPartitionRequest",
+  // Versions 0-1 were removed in Apache Kafka 4.0, version 2 is the new baseline.
+
+  // Version 1 adds LeaderRecoveryState field (KIP-704).
+  // Version 2 adds TopicId field to replace TopicName field (KIP-841).
+  //
+  // Version 3 adds the NewIsrEpochs field and deprecates the NewIsr field (KIP-903).
+  "validVersions": "2-3",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
+      "about": "The ID of the requesting broker." },
+    { "name": "BrokerEpoch", "type": "int64", "versions": "0+", "default": "-1",
+      "about": "The epoch of the requesting broker." },
+    { "name": "Topics", "type": "[]TopicData", "versions": "0+",
+      "about": "The topics to alter ISRs for.", "fields": [
+      { "name":  "TopicId", "type": "uuid", "versions": "2+", "ignorable": true,
+        "about": "The ID of the topic to alter ISRs for." },
+      { "name": "Partitions", "type": "[]PartitionData", "versions": "0+",
+        "about": "The partitions to alter ISRs for.", "fields": [
+        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+          "about": "The partition index." },
+        { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
+          "about": "The leader epoch of this partition." },
+        { "name": "NewIsr", "type": "[]int32", "versions": "0-2", "entityType": "brokerId",
+          "about": "The ISR for this partition. Deprecated since version 3." },
+        { "name": "NewIsrWithEpochs", "type": "[]BrokerState", "versions": "3+",
+          "about":  "The ISR for this partition.", "fields": [
+          { "name": "BrokerId", "type": "int32", "versions": "3+", "entityType": "brokerId",
+            "about": "The ID of the broker." },
+          { "name": "BrokerEpoch", "type": "int64", "versions": "3+", "default": "-1",
+            "about": "The epoch of the broker. It will be -1 if the epoch check is not supported." }
+        ]},
+        { "name": "LeaderRecoveryState", "type": "int8", "versions": "1+", "default": "0",
+          "about": "1 if the partition is recovering from an unclean leader election; 0 otherwise." },
+        { "name": "PartitionEpoch", "type": "int32", "versions": "0+",
+          "about": "The expected epoch of the partition which is being updated." }
+      ]}
+    ]}
+  ]
+}

kafka/protocol/new/schemas/resources/AlterPartitionResponse.json

@@ -0,0 +1,57 @@
+// Licensed to the Apache Software Foundation (ASF) under one or more
+// contributor license agreements.  See the NOTICE file distributed with
+// this work for additional information regarding copyright ownership.
+// The ASF licenses this file to You under the Apache License, Version 2.0
+// (the "License"); you may not use this file except in compliance with
+// the License.  You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+{
+  "apiKey": 56,
+  "type": "response",
+  "name": "AlterPartitionResponse",
+  // Versions 0-1 were removed in Apache Kafka 4.0, version 2 is the new baseline.
+
+  // Version 1 adds LeaderRecoveryState field (KIP-704).
+  // Version 2 adds TopicId field to replace TopicName field, can return the following new errors:
+  // INELIGIBLE_REPLICA, NEW_LEADER_ELECTED and UNKNOWN_TOPIC_ID (KIP-841).
+  //
+  // Version 3 is the same as version 2 (KIP-903).
+  "validVersions": "2-3",
+  "flexibleVersions": "0+",
+  "fields": [
+    { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
+      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota." },
+    { "name": "ErrorCode", "type": "int16", "versions": "0+",
+      "about": "The top level response error code." },
+    { "name": "Topics", "type": "[]TopicData", "versions": "0+",
+      "about": "The responses for each topic.", "fields": [
+      { "name":  "TopicId", "type": "uuid", "versions": "2+", "ignorable": true,
+        "about": "The ID of the topic." },
+      { "name": "Partitions", "type": "[]PartitionData", "versions": "0+",
+        "about": "The responses for each partition.", "fields": [
+        { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+          "about": "The partition index." },
+        { "name": "ErrorCode", "type": "int16", "versions": "0+",
+          "about": "The partition level error code." },
+        { "name": "LeaderId", "type": "int32", "versions": "0+", "entityType": "brokerId",
+          "about": "The broker ID of the leader." },
+        { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
+          "about": "The leader epoch." },
+        { "name": "Isr", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
+          "about": "The in-sync replica IDs." },
+        { "name": "LeaderRecoveryState", "type": "int8", "versions": "1+", "default": "0", "ignorable": true,
+          "about": "1 if the partition is recovering from an unclean leader election; 0 otherwise." },
+        { "name": "PartitionEpoch", "type": "int32", "versions": "0+",
+          "about": "The current epoch for the partition for KRaft controllers." }
+      ]}
+    ]}
+  ]
+}