feat(api): add automq node api

clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java

@@ -133,6 +133,9 @@ public enum ApiKeys {
     GET_KVS(ApiMessageType.GET_KVS, false, true),
     PUT_KVS(ApiMessageType.PUT_KVS, false, true),
     DELETE_KVS(ApiMessageType.DELETE_KVS, false, true),
+    AUTOMQ_REGISTER_NODE(ApiMessageType.AUTOMQ_REGISTER_NODE, false, false),
+    AUTOMQ_GET_NODES(ApiMessageType.AUTOMQ_GET_NODES, false, false),
+
     GET_NEXT_NODE_ID(ApiMessageType.GET_NEXT_NODE_ID, false, true),
     DESCRIBE_STREAMS(ApiMessageType.DESCRIBE_STREAMS, false, true);
     // AutoMQ for Kafka inject end

clients/src/main/java/org/apache/kafka/common/requests/s3/AutomqGetNodesRequest.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2024, AutoMQ HK Limited.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+package org.apache.kafka.common.requests.s3;
+
+import java.nio.ByteBuffer;
+import org.apache.kafka.common.message.AutomqGetNodesRequestData;
+import org.apache.kafka.common.message.AutomqGetNodesResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.AbstractResponse;
+import org.apache.kafka.common.requests.ApiError;
+
+public class AutomqGetNodesRequest extends AbstractRequest {
+    private final AutomqGetNodesRequestData data;
+
+    public AutomqGetNodesRequest(AutomqGetNodesRequestData data, short version) {
+        super(ApiKeys.AUTOMQ_GET_NODES, version);
+        this.data = data;
+    }
+
+    @Override
+    public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        ApiError apiError = ApiError.fromThrowable(e);
+        AutomqGetNodesResponseData response = new AutomqGetNodesResponseData()
+            .setErrorCode(apiError.error().code())
+            .setThrottleTimeMs(throttleTimeMs);
+        return new AutomqGetNodesResponse(response);
+    }
+
+    @Override
+    public AutomqGetNodesRequestData data() {
+        return data;
+    }
+
+    public static AutomqGetNodesRequest parse(ByteBuffer buffer, short version) {
+        return new AutomqGetNodesRequest(new AutomqGetNodesRequestData(new ByteBufferAccessor(buffer), version), version);
+    }
+
+    public static class Builder extends AbstractRequest.Builder<AutomqGetNodesRequest> {
+
+        private final AutomqGetNodesRequestData data;
+
+        public Builder(AutomqGetNodesRequestData data) {
+            super(ApiKeys.AUTOMQ_GET_NODES);
+            this.data = data;
+        }
+
+        @Override
+        public AutomqGetNodesRequest build(short version) {
+            return new AutomqGetNodesRequest(data, version);
+        }
+
+        @Override
+        public String toString() {
+            return data.toString();
+        }
+    }
+}

clients/src/main/java/org/apache/kafka/common/requests/s3/AutomqGetNodesResponse.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2024, AutoMQ HK Limited.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+package org.apache.kafka.common.requests.s3;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import org.apache.kafka.common.message.AutomqGetNodesResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+
+public class AutomqGetNodesResponse extends AbstractResponse {
+
+    private final AutomqGetNodesResponseData data;
+
+    public AutomqGetNodesResponse(AutomqGetNodesResponseData data) {
+        super(ApiKeys.AUTOMQ_GET_NODES);
+        this.data = data;
+    }
+
+    @Override
+    public AutomqGetNodesResponseData data() {
+        return data;
+    }
+
+    @Override
+    public Map<Errors, Integer> errorCounts() {
+        return errorCounts(Errors.forCode(data.errorCode()));
+    }
+
+    @Override
+    public int throttleTimeMs() {
+        return data.throttleTimeMs();
+    }
+
+    @Override
+    public void maybeSetThrottleTimeMs(int throttleTimeMs) {
+        data.setThrottleTimeMs(throttleTimeMs);
+    }
+
+    public static AutomqGetNodesResponse parse(ByteBuffer buffer, short version) {
+        return new AutomqGetNodesResponse(new AutomqGetNodesResponseData(
+            new ByteBufferAccessor(buffer), version));
+    }
+}

clients/src/main/java/org/apache/kafka/common/requests/s3/AutomqRegisterNodeRequest.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2024, AutoMQ HK Limited.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+package org.apache.kafka.common.requests.s3;
+
+import java.nio.ByteBuffer;
+import org.apache.kafka.common.message.AutomqRegisterNodeRequestData;
+import org.apache.kafka.common.message.AutomqRegisterNodeResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.requests.AbstractRequest;
+import org.apache.kafka.common.requests.ApiError;
+
+public class AutomqRegisterNodeRequest extends AbstractRequest {
+
+    public static class Builder extends AbstractRequest.Builder<AutomqRegisterNodeRequest> {
+
+        private final AutomqRegisterNodeRequestData data;
+
+        public Builder(AutomqRegisterNodeRequestData data) {
+            super(ApiKeys.AUTOMQ_REGISTER_NODE);
+            this.data = data;
+        }
+
+        @Override
+        public AutomqRegisterNodeRequest build(short version) {
+            return new AutomqRegisterNodeRequest(data, version);
+        }
+
+        @Override
+        public String toString() {
+            return data.toString();
+        }
+    }
+
+    private final AutomqRegisterNodeRequestData data;
+
+    public AutomqRegisterNodeRequest(AutomqRegisterNodeRequestData data, short version) {
+        super(ApiKeys.AUTOMQ_REGISTER_NODE, version);
+        this.data = data;
+    }
+
+    @Override
+    public AutomqRegisterNodeResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        ApiError apiError = ApiError.fromThrowable(e);
+        AutomqRegisterNodeResponseData response = new AutomqRegisterNodeResponseData()
+            .setErrorCode(apiError.error().code())
+            .setThrottleTimeMs(throttleTimeMs);
+        return new AutomqRegisterNodeResponse(response);
+    }
+
+    @Override
+    public AutomqRegisterNodeRequestData data() {
+        return data;
+    }
+
+    public static AutomqRegisterNodeRequest parse(ByteBuffer buffer, short version) {
+        return new AutomqRegisterNodeRequest(new AutomqRegisterNodeRequestData(
+            new ByteBufferAccessor(buffer), version), version);
+    }
+}

clients/src/main/java/org/apache/kafka/common/requests/s3/AutomqRegisterNodeResponse.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2024, AutoMQ HK Limited.
+ *
+ * Use of this software is governed by the Business Source License
+ * included in the file BSL.md
+ *
+ * As of the Change Date specified in that file, in accordance with
+ * the Business Source License, use of this software will be governed
+ * by the Apache License, Version 2.0
+ */
+
+package org.apache.kafka.common.requests.s3;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+import org.apache.kafka.common.message.AutomqRegisterNodeResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.AbstractResponse;
+
+public class AutomqRegisterNodeResponse extends AbstractResponse {
+
+    private final AutomqRegisterNodeResponseData data;
+
+    public AutomqRegisterNodeResponse(AutomqRegisterNodeResponseData data) {
+        super(ApiKeys.AUTOMQ_REGISTER_NODE);
+        this.data = data;
+    }
+
+    @Override
+    public AutomqRegisterNodeResponseData data() {
+        return data;
+    }
+
+    @Override
+    public Map<Errors, Integer> errorCounts() {
+        return errorCounts(Errors.forCode(data.errorCode()));
+    }
+
+    @Override
+    public int throttleTimeMs() {
+        return data.throttleTimeMs();
+    }
+
+    @Override
+    public void maybeSetThrottleTimeMs(int throttleTimeMs) {
+        data.setThrottleTimeMs(throttleTimeMs);
+    }
+
+    public static AutomqRegisterNodeResponse parse(ByteBuffer buffer, short version) {
+        return new AutomqRegisterNodeResponse(new AutomqRegisterNodeResponseData(
+            new ByteBufferAccessor(buffer), version));
+    }
+}

clients/src/main/resources/common/message/AutomqGetNodesRequest.json

@@ -0,0 +1,34 @@
+// 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": 514,
+  "type": "request",
+  "listeners": [
+    "controller",
+    "broker"
+  ],
+  "name": "AutomqGetNodesRequest",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    {
+      "name": "NodeIds",
+      "type": "[]int32",
+      "versions": "0+",
+      "about": "List of node IDs for whose info is requested"
+    }
+  ]
+}

clients/src/main/resources/common/message/AutomqGetNodesResponse.json

@@ -0,0 +1,83 @@
+// 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": 514,
+  "type": "response",
+  "name": "AutomqGetNodesResponse",
+  "validVersions": "0",
+  "flexibleVersions": "0+",
+  "fields": [
+    {
+      "name": "ErrorCode",
+      "type": "int16",
+      "versions": "0+",
+      "about": "The top level response error code"
+    },
+    {
+      "name": "ThrottleTimeMs",
+      "type": "int32",
+      "versions": "0+",
+      "about": "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": "Nodes", "type": "[]NodeMetadata", "versions": "0+", "about": "Metadata of list of nodes", "fields": [
+      {
+        "name": "NodeId",
+        "type": "int32",
+        "versions": "0+",
+        "about": "The node ID"
+      },
+      {
+        "name": "NodeEpoch",
+        "type": "int64",
+        "versions": "0+",
+        "about": "The node epoch"
+      },
+      {
+        "name": "WALConfig",
+        "type": "string",
+        "versions": "0+",
+        "about": "The WAL configuration"
+      },
+      {
+        "name": "State",
+        "type": "string",
+        "versions": "0+",
+        "about": "The node state"
+      },
+      {
+        "name": "HasOpeningStreams",
+        "type": "bool",
+        "versions": "0+",
+        "about": "Whether the node has opening streams"
+      },
+      { "name": "Tags", "type": "[]Tag", "versions": "0+", "about": "The node tags.", "fields": [
+        {
+          "name": "Key",
+          "type": "string",
+          "versions": "0+",
+          "mapKey": true,
+          "about": "The tag key."
+        },
+        {
+          "name": "Value",
+          "type": "string",
+          "versions": "0+",
+          "about": "The tag value."
+        }
+      ]}
+    ]}
+  ]
+}