KAFKA-15514: Metadata records Replicas->Assignment

build.gradle

@@ -1758,7 +1758,7 @@ project(':storage') {
   task processMessages(type:JavaExec) {
     mainClass = "org.apache.kafka.message.MessageGenerator"
     classpath = configurations.generator
-    args = [ "-p", " org.apache.kafka.server.log.remote.metadata.storage.generated",
+    args = [ "-p", "org.apache.kafka.server.log.remote.metadata.storage.generated",
              "-o", "src/generated/java/org/apache/kafka/server/log/remote/metadata/storage/generated",
              "-i", "src/main/resources/message",
              "-m", "MessageDataGenerator", "JsonConverterGenerator",

checkstyle/checkstyle.xml

@@ -112,7 +112,9 @@
     </module>
 
     <!-- code quality -->
-    <module name="MethodLength"/>
+    <module name="MethodLength">
+      <property name="max" value="170" />
+    </module>
     <module name="ParameterNumber">
       <!-- default is 8 -->
       <property name="max" value="13"/>

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

@@ -115,7 +115,8 @@ public enum ApiKeys {
     CONSUMER_GROUP_DESCRIBE(ApiMessageType.CONSUMER_GROUP_DESCRIBE),
     CONTROLLER_REGISTRATION(ApiMessageType.CONTROLLER_REGISTRATION),
     GET_TELEMETRY_SUBSCRIPTIONS(ApiMessageType.GET_TELEMETRY_SUBSCRIPTIONS),
-    PUSH_TELEMETRY(ApiMessageType.PUSH_TELEMETRY);
+    PUSH_TELEMETRY(ApiMessageType.PUSH_TELEMETRY),
+    ASSIGN_REPLICAS_TO_DIRS(ApiMessageType.ASSIGN_REPLICAS_TO_DIRS);
 
     private static final Map<ApiMessageType.ListenerType, EnumSet<ApiKeys>> APIS_BY_LISTENER =
         new EnumMap<>(ApiMessageType.ListenerType.class);

clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java

@@ -320,6 +320,8 @@ private static AbstractRequest doParseRequest(ApiKeys apiKey, short apiVersion,
                 return GetTelemetrySubscriptionsRequest.parse(buffer, apiVersion);
             case PUSH_TELEMETRY:
                 return PushTelemetryRequest.parse(buffer, apiVersion);
+            case ASSIGN_REPLICAS_TO_DIRS:
+                return AssignReplicasToDirsRequest.parse(buffer, apiVersion);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseRequest`, the " +
                         "code should be updated to do so.", apiKey));

clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java

@@ -257,6 +257,8 @@ public static AbstractResponse parseResponse(ApiKeys apiKey, ByteBuffer response
                 return GetTelemetrySubscriptionsResponse.parse(responseBuffer, version);
             case PUSH_TELEMETRY:
                 return PushTelemetryResponse.parse(responseBuffer, version);
+            case ASSIGN_REPLICAS_TO_DIRS:
+                return AssignReplicasToDirsResponse.parse(responseBuffer, version);
             default:
                 throw new AssertionError(String.format("ApiKey %s is not currently handled in `parseResponse`, the " +
                         "code should be updated to do so.", apiKey));

clients/src/main/java/org/apache/kafka/common/requests/AssignReplicasToDirsRequest.java

@@ -0,0 +1,75 @@
+/*
+ * 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.
+ */
+
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.message.AssignReplicasToDirsRequestData;
+import org.apache.kafka.common.message.AssignReplicasToDirsResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
+
+import java.nio.ByteBuffer;
+
+public class AssignReplicasToDirsRequest extends AbstractRequest {
+
+    public static class Builder extends AbstractRequest.Builder<AssignReplicasToDirsRequest> {
+
+        private final AssignReplicasToDirsRequestData data;
+
+        public Builder(AssignReplicasToDirsRequestData data) {
+            super(ApiKeys.ASSIGN_REPLICAS_TO_DIRS);
+            this.data = data;
+        }
+
+        @Override
+        public AssignReplicasToDirsRequest build(short version) {
+            return new AssignReplicasToDirsRequest(data, version);
+        }
+
+        @Override
+        public String toString() {
+            return data.toString();
+        }
+    }
+
+    private final AssignReplicasToDirsRequestData data;
+
+    public AssignReplicasToDirsRequest(AssignReplicasToDirsRequestData data, short version) {
+        super(ApiKeys.ASSIGN_REPLICAS_TO_DIRS, version);
+        this.data = data;
+    }
+
+    @Override
+    public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
+        return new AssignReplicasToDirsResponse(
+                new AssignReplicasToDirsResponseData()
+                        .setThrottleTimeMs(throttleTimeMs)
+                        .setErrorCode(Errors.forException(e).code())
+        );
+    }
+
+    @Override
+    public AssignReplicasToDirsRequestData data() {
+        return data;
+    }
+
+    public static AssignReplicasToDirsRequest parse(ByteBuffer buffer, short version) {
+        return new AssignReplicasToDirsRequest(new AssignReplicasToDirsRequestData(
+                new ByteBufferAccessor(buffer), version), version);
+    }
+}

clients/src/main/java/org/apache/kafka/common/requests/AssignReplicasToDirsResponse.java

@@ -0,0 +1,62 @@
+/*
+ * 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.
+ */
+package org.apache.kafka.common.requests;
+
+import org.apache.kafka.common.message.AssignReplicasToDirsResponseData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.protocol.Errors;
+
+import java.nio.ByteBuffer;
+
+import java.util.Collections;
+import java.util.Map;
+
+public class AssignReplicasToDirsResponse extends AbstractResponse {
+
+    private final AssignReplicasToDirsResponseData data;
+
+    public AssignReplicasToDirsResponse(AssignReplicasToDirsResponseData data) {
+        super(ApiKeys.ASSIGN_REPLICAS_TO_DIRS);
+        this.data = data;
+    }
+
+    @Override
+    public AssignReplicasToDirsResponseData data() {
+        return data;
+    }
+
+    @Override
+    public Map<Errors, Integer> errorCounts() {
+        return Collections.singletonMap(Errors.forCode(data.errorCode()), 1);
+    }
+
+    @Override
+    public int throttleTimeMs() {
+        return data.throttleTimeMs();
+    }
+
+    @Override
+    public void maybeSetThrottleTimeMs(int throttleTimeMs) {
+        data.setThrottleTimeMs(throttleTimeMs);
+    }
+
+    public static AssignReplicasToDirsResponse parse(ByteBuffer buffer, short version) {
+        return new AssignReplicasToDirsResponse(new AssignReplicasToDirsResponseData(
+                new ByteBufferAccessor(buffer), version));
+    }
+}

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

@@ -0,0 +1,40 @@
+// 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": 73,
+  "type": "request",
+  "listeners": ["controller"],
+  "name": "AssignReplicasToDirsRequest",
+  "validVersions": "0",
+  "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": "Directories", "type":  "[]DirectoryData", "versions": "0+", "fields":  [
+      { "name":  "Id", "type": "uuid", "versions": "0+", "about": "The ID of the directory" },
+      { "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
+        { "name":  "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+          "about": "The name of the assigned topic" },
+        { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
+          { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+            "about": "The partition index" }
+        ]}
+      ]}
+    ]}
+  ]
+}

clients/src/main/resources/common/message/AssignReplicasToDirsResponse.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": 73,
+  "type": "response",
+  "name": "AssignReplicasToDirsResponse",
+  "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 response error code" },
+    { "name": "Directories", "type":  "[]DirectoryData", "versions": "0+", "fields":  [
+      { "name":  "Id", "type": "uuid", "versions": "0+", "about": "The ID of the directory" },
+      { "name": "Topics", "type": "[]TopicData", "versions": "0+", "fields": [
+        { "name":  "Name", "type": "string", "versions": "0+", "entityType": "topicName",
+          "about": "The name of the assigned topic" },
+        { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [
+          { "name": "PartitionIndex", "type": "int32", "versions": "0+",
+            "about": "The partition index" },
+          { "name": "ErrorCode", "type": "int16", "versions": "0+",
+            "about": "The partition level error code" }
+        ]}
+      ]}
+    ]}
+  ]
+}

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

@@ -18,7 +18,7 @@
   "type": "request",
   "listeners": ["controller"],
   "name": "BrokerHeartbeatRequest",
-  "validVersions": "0",
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
@@ -30,6 +30,8 @@
     { "name": "WantFence", "type": "bool", "versions": "0+",
       "about": "True if the broker wants to be fenced, false otherwise." },
     { "name": "WantShutDown", "type": "bool", "versions": "0+",
-      "about": "True if the broker wants to be shut down, false otherwise." }
+      "about": "True if the broker wants to be shut down, false otherwise." },
+    { "name": "OfflineLogDirs", "type":  "[]uuid", "versions": "1+", "taggedVersions": "1+", "tag": "0",
+      "about": "Log directories that failed and went offline." }
   ]
 }

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

@@ -17,7 +17,7 @@
   "apiKey": 63,
   "type": "response",
   "name": "BrokerHeartbeatResponse",
-  "validVersions": "0",
+  "validVersions": "0-1",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",

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

@@ -14,14 +14,16 @@
 // limitations under the License.
 
 // Version 1 adds Zk broker epoch to the request if the broker is migrating from Zk mode to KRaft mode.
-//
+
 // Version 2 adds the PreviousBrokerEpoch for the KIP-966
+
+// Version 3 adds LogDirs for KIP-858
 {
   "apiKey":62,
   "type": "request",
   "listeners": ["controller"],
   "name": "BrokerRegistrationRequest",
-  "validVersions": "0-2",
+  "validVersions": "0-3",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
@@ -57,6 +59,8 @@
     { "name": "IsMigratingZkBroker", "type": "bool", "versions": "1+", "default": "false",
       "about": "If the required configurations for ZK migration are present, this value is set to true" },
     { "name": "PreviousBrokerEpoch", "type": "int64", "versions": "2+", "default": "-1",
-      "about": "The epoch before a clean shutdown." }
+      "about": "The epoch before a clean shutdown." },
+    { "name": "LogDirs", "type":  "[]uuid", "versions":  "3+",
+      "about": "Log directories configured in this broker which are available." }
   ]
 }

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

@@ -20,7 +20,7 @@
   "apiKey": 62,
   "type": "response",
   "name": "BrokerRegistrationResponse",
-  "validVersions": "0-2",
+  "validVersions": "0-3",
   "flexibleVersions": "0+",
   "fields": [
     { "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",

clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java

@@ -105,7 +105,8 @@ public void testAllVersionsHaveSchemas() {
         for (ApiMessageType type : ApiMessageType.values()) {
             assertEquals(0, type.lowestSupportedVersion());
 
-            assertEquals(type.requestSchemas().length, type.responseSchemas().length);
+            assertEquals(type.requestSchemas().length, type.responseSchemas().length,
+                    "request and response schemas must be the same length for " + type.name());
             for (Schema schema : type.requestSchemas())
                 assertNotNull(schema);
             for (Schema schema : type.responseSchemas())