From 2ca98af0f00c3cd7b842d8fa00eab85d72cee45a Mon Sep 17 00:00:00 2001 From: Igor Soarez Date: Sun, 29 Oct 2023 18:07:10 +0000 Subject: [PATCH 1/6] KAFKA-15355: Message schema changes https://issues.apache.org/jira/browse/KAFKA-15355 --- build.gradle | 2 +- checkstyle/checkstyle.xml | 4 +- .../errors/InvalidMetadataException.java | 2 +- .../apache/kafka/common/protocol/ApiKeys.java | 3 +- .../common/requests/AbstractRequest.java | 2 + .../common/requests/AbstractResponse.java | 2 + .../requests/AssignReplicasToDirsRequest.java | 75 ++++++++++ .../AssignReplicasToDirsResponse.java | 62 ++++++++ .../message/AssignReplicasToDirsRequest.json | 40 +++++ .../message/AssignReplicasToDirsResponse.json | 41 ++++++ .../message/BrokerHeartbeatRequest.json | 6 +- .../message/BrokerHeartbeatResponse.json | 2 +- .../message/BrokerRegistrationRequest.json | 10 +- .../message/BrokerRegistrationResponse.json | 2 +- .../common/message/ApiMessageTypeTest.java | 3 +- .../common/requests/RequestResponseTest.java | 69 +++++++++ .../kafka/network/RequestConvertToJson.scala | 2 + .../scala/kafka/server/ControllerApis.scala | 10 ++ ...AlterUserScramCredentialsRequestTest.scala | 2 +- .../kafka/server/ControllerApisTest.scala | 28 ++++ .../unit/kafka/server/RequestQuotaTest.scala | 3 + .../controller/PartitionChangeBuilder.java | 5 + .../controller/ReplicationControlManager.java | 9 +- .../org/apache/kafka/image/TopicImage.java | 2 +- .../kafka/metadata/PartitionRegistration.java | 46 +++++- .../BrokerRegistrationChangeRecord.json | 8 +- .../metadata/PartitionChangeRecord.json | 11 +- .../common/metadata/PartitionRecord.json | 10 +- .../common/metadata/RegisterBrokerRecord.json | 9 +- .../PartitionChangeBuilderTest.java | 123 +++++++++++++++- .../metrics/ControllerMetricsChangesTest.java | 16 +- .../kafka/image/ImageDowngradeTest.java | 30 ++++ .../metadata/PartitionRegistrationTest.java | 47 +++++- .../org/apache/kafka/common/DirectoryId.java | 137 ++++++++++++++++-- .../kafka/server/common/MetadataVersion.java | 17 ++- .../apache/kafka/common/DirectoryIdTest.java | 95 ++++++++++++ 36 files changed, 876 insertions(+), 59 deletions(-) create mode 100644 clients/src/main/java/org/apache/kafka/common/requests/AssignReplicasToDirsRequest.java create mode 100644 clients/src/main/java/org/apache/kafka/common/requests/AssignReplicasToDirsResponse.java create mode 100644 clients/src/main/resources/common/message/AssignReplicasToDirsRequest.json create mode 100644 clients/src/main/resources/common/message/AssignReplicasToDirsResponse.json create mode 100644 server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java diff --git a/build.gradle b/build.gradle index fbda13d5a7983..96ca37de40cee 100644 --- a/build.gradle +++ b/build.gradle @@ -1751,7 +1751,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", diff --git a/checkstyle/checkstyle.xml b/checkstyle/checkstyle.xml index 8cce1fd387276..c42bad21b1b9a 100644 --- a/checkstyle/checkstyle.xml +++ b/checkstyle/checkstyle.xml @@ -112,7 +112,9 @@ - + + + diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java index e6663dbf185d0..f8ae485c99282 100644 --- a/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java +++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java @@ -19,7 +19,7 @@ /** * An exception that may indicate the client's metadata is out of date */ -public abstract class InvalidMetadataException extends RetriableException { +public class InvalidMetadataException extends RetriableException { private static final long serialVersionUID = 1L; diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java b/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java index 1da9eeaccd454..1635da2e200da 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java +++ b/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> APIS_BY_LISTENER = new EnumMap<>(ApiMessageType.ListenerType.class); diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java index 2d5b0c66329a1..e71c5debf62c7 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java +++ b/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)); diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java index 1747ed1dbd297..d849a0740b3a3 100644 --- a/clients/src/main/java/org/apache/kafka/common/requests/AbstractResponse.java +++ b/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)); diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AssignReplicasToDirsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/AssignReplicasToDirsRequest.java new file mode 100644 index 0000000000000..4edd726850a50 --- /dev/null +++ b/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 { + + 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); + } +} diff --git a/clients/src/main/java/org/apache/kafka/common/requests/AssignReplicasToDirsResponse.java b/clients/src/main/java/org/apache/kafka/common/requests/AssignReplicasToDirsResponse.java new file mode 100644 index 0000000000000..e92ebcb103bac --- /dev/null +++ b/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 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)); + } +} diff --git a/clients/src/main/resources/common/message/AssignReplicasToDirsRequest.json b/clients/src/main/resources/common/message/AssignReplicasToDirsRequest.json new file mode 100644 index 0000000000000..b349520745b34 --- /dev/null +++ b/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": "TopicId", "type": "uuid", "versions": "0+", + "about": "The ID of the assigned topic" }, + { "name": "Partitions", "type": "[]PartitionData", "versions": "0+", "fields": [ + { "name": "PartitionIndex", "type": "int32", "versions": "0+", + "about": "The partition index" } + ]} + ]} + ]} + ] +} diff --git a/clients/src/main/resources/common/message/AssignReplicasToDirsResponse.json b/clients/src/main/resources/common/message/AssignReplicasToDirsResponse.json new file mode 100644 index 0000000000000..185d379584cfe --- /dev/null +++ b/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": "TopicId", "type": "uuid", "versions": "0+", + "about": "The ID 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" } + ]} + ]} + ]} + ] +} diff --git a/clients/src/main/resources/common/message/BrokerHeartbeatRequest.json b/clients/src/main/resources/common/message/BrokerHeartbeatRequest.json index 2cf2577cad2c1..9ebdc3707d140 100644 --- a/clients/src/main/resources/common/message/BrokerHeartbeatRequest.json +++ b/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." } ] } diff --git a/clients/src/main/resources/common/message/BrokerHeartbeatResponse.json b/clients/src/main/resources/common/message/BrokerHeartbeatResponse.json index ce9aaba4e26ab..9956daf1ff0e6 100644 --- a/clients/src/main/resources/common/message/BrokerHeartbeatResponse.json +++ b/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+", diff --git a/clients/src/main/resources/common/message/BrokerRegistrationRequest.json b/clients/src/main/resources/common/message/BrokerRegistrationRequest.json index 6c51eac301c9c..24e4f07489b50 100644 --- a/clients/src/main/resources/common/message/BrokerRegistrationRequest.json +++ b/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." } ] } diff --git a/clients/src/main/resources/common/message/BrokerRegistrationResponse.json b/clients/src/main/resources/common/message/BrokerRegistrationResponse.json index 3b483abfc05a5..1f2ff94162f6c 100644 --- a/clients/src/main/resources/common/message/BrokerRegistrationResponse.json +++ b/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+", diff --git a/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java b/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java index 40da2701eac1c..fb37cc80a1a9a 100644 --- a/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java +++ b/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()) diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java index 9ad4569225b2d..a802ee3e58035 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java @@ -61,6 +61,8 @@ import org.apache.kafka.common.message.ApiVersionsResponseData; import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion; import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersionCollection; +import org.apache.kafka.common.message.AssignReplicasToDirsRequestData; +import org.apache.kafka.common.message.AssignReplicasToDirsResponseData; import org.apache.kafka.common.message.BeginQuorumEpochRequestData; import org.apache.kafka.common.message.BeginQuorumEpochResponseData; import org.apache.kafka.common.message.BrokerHeartbeatRequestData; @@ -1071,6 +1073,7 @@ private AbstractRequest getRequest(ApiKeys apikey, short version) { case CONTROLLER_REGISTRATION: return createControllerRegistrationRequest(version); case GET_TELEMETRY_SUBSCRIPTIONS: return createGetTelemetrySubscriptionsRequest(version); case PUSH_TELEMETRY: return createPushTelemetryRequest(version); + case ASSIGN_REPLICAS_TO_DIRS: return createAssignReplicasToDirsRequest(version); default: throw new IllegalArgumentException("Unknown API key " + apikey); } } @@ -1150,6 +1153,7 @@ private AbstractResponse getResponse(ApiKeys apikey, short version) { case CONTROLLER_REGISTRATION: return createControllerRegistrationResponse(); case GET_TELEMETRY_SUBSCRIPTIONS: return createGetTelemetrySubscriptionsResponse(); case PUSH_TELEMETRY: return createPushTelemetryResponse(); + case ASSIGN_REPLICAS_TO_DIRS: return createAssignReplicasToDirsResponse(); default: throw new IllegalArgumentException("Unknown API key " + apikey); } } @@ -1178,6 +1182,71 @@ private ConsumerGroupDescribeResponse createConsumerGroupDescribeResponse() { return new ConsumerGroupDescribeResponse(data); } + private AssignReplicasToDirsRequest createAssignReplicasToDirsRequest(short version) { + AssignReplicasToDirsRequestData data = new AssignReplicasToDirsRequestData() + .setBrokerId(1) + .setBrokerEpoch(123L) + .setDirectories(Arrays.asList( + new AssignReplicasToDirsRequestData.DirectoryData() + .setId(Uuid.randomUuid()) + .setTopics(Arrays.asList( + new AssignReplicasToDirsRequestData.TopicData() + .setTopicId(Uuid.fromString("qo0Pcp70TdGnAa7YKMKCqw")) + .setPartitions(Arrays.asList( + new AssignReplicasToDirsRequestData.PartitionData() + .setPartitionIndex(8) + )) + )), + new AssignReplicasToDirsRequestData.DirectoryData() + .setId(Uuid.randomUuid()) + .setTopics(Arrays.asList( + new AssignReplicasToDirsRequestData.TopicData() + .setTopicId(Uuid.fromString("yEu11V7HTRGIwm6FDWFhzg")) + .setPartitions(Arrays.asList( + new AssignReplicasToDirsRequestData.PartitionData() + .setPartitionIndex(2), + new AssignReplicasToDirsRequestData.PartitionData() + .setPartitionIndex(80) + )) + )) + )); + return new AssignReplicasToDirsRequest.Builder(data).build(version); + } + + private AssignReplicasToDirsResponse createAssignReplicasToDirsResponse() { + AssignReplicasToDirsResponseData data = new AssignReplicasToDirsResponseData() + .setErrorCode(Errors.NONE.code()) + .setThrottleTimeMs(123) + .setDirectories(Arrays.asList( + new AssignReplicasToDirsResponseData.DirectoryData() + .setId(Uuid.randomUuid()) + .setTopics(Arrays.asList( + new AssignReplicasToDirsResponseData.TopicData() + .setTopicId(Uuid.fromString("sKhZV8LnTA275KvByB9bVg")) + .setPartitions(Arrays.asList( + new AssignReplicasToDirsResponseData.PartitionData() + .setPartitionIndex(8) + .setErrorCode(Errors.NONE.code()) + )) + )), + new AssignReplicasToDirsResponseData.DirectoryData() + .setId(Uuid.randomUuid()) + .setTopics(Arrays.asList( + new AssignReplicasToDirsResponseData.TopicData() + .setTopicId(Uuid.fromString("ORLP5NEzRo64SvKq1hIVQg")) + .setPartitions(Arrays.asList( + new AssignReplicasToDirsResponseData.PartitionData() + .setPartitionIndex(2) + .setErrorCode(Errors.UNKNOWN_TOPIC_OR_PARTITION.code()), + new AssignReplicasToDirsResponseData.PartitionData() + .setPartitionIndex(8) + .setErrorCode(Errors.NONE.code()) + )) + )) + )); + return new AssignReplicasToDirsResponse(data); + } + private ConsumerGroupHeartbeatRequest createConsumerGroupHeartbeatRequest(short version) { ConsumerGroupHeartbeatRequestData data = new ConsumerGroupHeartbeatRequestData() .setGroupId("group") diff --git a/core/src/main/scala/kafka/network/RequestConvertToJson.scala b/core/src/main/scala/kafka/network/RequestConvertToJson.scala index e635a5820d547..6844acf66cba7 100644 --- a/core/src/main/scala/kafka/network/RequestConvertToJson.scala +++ b/core/src/main/scala/kafka/network/RequestConvertToJson.scala @@ -100,6 +100,7 @@ object RequestConvertToJson { case req: ConsumerGroupHeartbeatRequest => ConsumerGroupHeartbeatRequestDataJsonConverter.write(req.data, request.version) case req: ConsumerGroupDescribeRequest => ConsumerGroupDescribeRequestDataJsonConverter.write(req.data, request.version) case req: ControllerRegistrationRequest => ControllerRegistrationRequestDataJsonConverter.write(req.data, request.version) + case req: AssignReplicasToDirsRequest => AssignReplicasToDirsRequestDataJsonConverter.write(req.data, request.version) case _ => throw new IllegalStateException(s"ApiKey ${request.apiKey} is not currently handled in `request`, the " + "code should be updated to do so."); } @@ -180,6 +181,7 @@ object RequestConvertToJson { case res: ConsumerGroupHeartbeatResponse => ConsumerGroupHeartbeatResponseDataJsonConverter.write(res.data, version) case res: ConsumerGroupDescribeResponse => ConsumerGroupDescribeResponseDataJsonConverter.write(res.data, version) case req: ControllerRegistrationResponse => ControllerRegistrationResponseDataJsonConverter.write(req.data, version) + case res: AssignReplicasToDirsResponse => AssignReplicasToDirsResponseDataJsonConverter.write(res.data, version) case _ => throw new IllegalStateException(s"ApiKey ${response.apiKey} is not currently handled in `response`, the " + "code should be updated to do so."); } diff --git a/core/src/main/scala/kafka/server/ControllerApis.scala b/core/src/main/scala/kafka/server/ControllerApis.scala index 856d9365ae883..9015679893b62 100644 --- a/core/src/main/scala/kafka/server/ControllerApis.scala +++ b/core/src/main/scala/kafka/server/ControllerApis.scala @@ -128,6 +128,7 @@ class ControllerApis( case ApiKeys.UPDATE_FEATURES => handleUpdateFeatures(request) case ApiKeys.DESCRIBE_CLUSTER => handleDescribeCluster(request) case ApiKeys.CONTROLLER_REGISTRATION => handleControllerRegistration(request) + case ApiKeys.ASSIGN_REPLICAS_TO_DIRS => handleAssignReplicasToDirs(request) case _ => throw new ApiException(s"Unsupported ApiKey ${request.context.header.apiKey}") } @@ -1074,4 +1075,13 @@ class ControllerApis( new DescribeClusterResponse(response.setThrottleTimeMs(requestThrottleMs))) CompletableFuture.completedFuture[Unit](()) } + + def handleAssignReplicasToDirs(request: RequestChannel.Request): CompletableFuture[Unit] = { + val assignReplicasToDirsRequest = request.body[AssignReplicasToDirsRequest] + + // TODO KAFKA-15426 + requestHelper.sendMaybeThrottle(request, + assignReplicasToDirsRequest.getErrorResponse(Errors.UNSUPPORTED_VERSION.exception)) + CompletableFuture.completedFuture[Unit](()) + } } diff --git a/core/src/test/scala/unit/kafka/server/AlterUserScramCredentialsRequestTest.scala b/core/src/test/scala/unit/kafka/server/AlterUserScramCredentialsRequestTest.scala index 159524c92d1ae..3e4ac4b6771db 100644 --- a/core/src/test/scala/unit/kafka/server/AlterUserScramCredentialsRequestTest.scala +++ b/core/src/test/scala/unit/kafka/server/AlterUserScramCredentialsRequestTest.scala @@ -404,7 +404,7 @@ class AlterUserScramCredentialsRequestTest extends BaseRequestTest { val results = response.data.results assertEquals(1, results.size) checkAllErrorsAlteringCredentials(results, Errors.UNSUPPORTED_VERSION, - "when altering the credentials on unsupported IPB version") + "when altering the credentials on unsupported IBP version") assertEquals("The current metadata version does not support SCRAM", results.get(0).errorMessage) } diff --git a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala index 200a898c60596..5afc22f782459 100644 --- a/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala +++ b/core/src/test/scala/unit/kafka/server/ControllerApisTest.scala @@ -1095,6 +1095,34 @@ class ControllerApisTest { assertEquals(Errors.NOT_CONTROLLER, response.error) } + @Test + def testAssignReplicasToDirsReturnsUnsupportedVersion(): Unit = { + val controller = mock(classOf[Controller]) + val controllerApis = createControllerApis(None, controller) + + val request = + new AssignReplicasToDirsRequest.Builder( + new AssignReplicasToDirsRequestData() + .setBrokerId(1) + .setBrokerEpoch(123L) + .setDirectories(util.Arrays.asList( + new AssignReplicasToDirsRequestData.DirectoryData() + .setId(Uuid.randomUuid()) + .setTopics(util.Arrays.asList( + new AssignReplicasToDirsRequestData.TopicData() + .setTopicId(Uuid.fromString("pcPTaiQfRXyZG88kO9k2aA")) + .setPartitions(util.Arrays.asList( + new AssignReplicasToDirsRequestData.PartitionData() + .setPartitionIndex(8) + )) + )) + ))).build() + + val expectedResponse = new AssignReplicasToDirsResponseData().setErrorCode(Errors.UNSUPPORTED_VERSION.code) + val response = handleRequest[AssignReplicasToDirsResponse](request, controllerApis) + assertEquals(expectedResponse, response.data) + } + private def handleRequest[T <: AbstractResponse]( request: AbstractRequest, controllerApis: ControllerApis diff --git a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala index 2f7692d49c17e..a817881b17a7c 100644 --- a/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala +++ b/core/src/test/scala/unit/kafka/server/RequestQuotaTest.scala @@ -719,6 +719,9 @@ class RequestQuotaTest extends BaseRequestTest { case ApiKeys.PUSH_TELEMETRY => new PushTelemetryRequest.Builder(new PushTelemetryRequestData(), true) + case ApiKeys.ASSIGN_REPLICAS_TO_DIRS => + new AssignReplicasToDirsRequest.Builder(new AssignReplicasToDirsRequestData()) + case _ => throw new IllegalArgumentException("Unsupported API key " + apiKey) } diff --git a/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java b/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java index d89996380315b..d0e2b91cb7ba5 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java +++ b/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java @@ -26,6 +26,7 @@ import java.util.function.IntPredicate; import java.util.stream.Collectors; +import org.apache.kafka.common.DirectoryId; import org.apache.kafka.common.Uuid; import org.apache.kafka.common.message.AlterPartitionRequestData.BrokerState; import org.apache.kafka.common.metadata.PartitionChangeRecord; @@ -54,6 +55,7 @@ public static boolean changeRecordIsNoOp(PartitionChangeRecord record) { if (record.removingReplicas() != null) return false; if (record.addingReplicas() != null) return false; if (record.leaderRecoveryState() != LeaderRecoveryState.NO_CHANGE) return false; + if (record.directories() != null) return false; return true; } @@ -382,6 +384,9 @@ public Optional build() { private void setAssignmentChanges(PartitionChangeRecord record) { if (!targetReplicas.isEmpty() && !targetReplicas.equals(Replicas.toList(partition.replicas))) { + if (metadataVersion.isDirectoryAssignmentSupported()) { + record.setDirectories(DirectoryId.update(partition.replicas, partition.directories, targetReplicas)); + } record.setReplicas(targetReplicas); } if (!targetRemoving.equals(Replicas.toList(partition.removingReplicas))) { diff --git a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java index e923e671dbbeb..0262c0d435840 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java +++ b/metadata/src/main/java/org/apache/kafka/controller/ReplicationControlManager.java @@ -77,6 +77,7 @@ import org.apache.kafka.common.requests.AlterPartitionRequest; import org.apache.kafka.common.requests.ApiError; import org.apache.kafka.common.utils.LogContext; +import org.apache.kafka.image.writer.ImageWriterOptions; import org.apache.kafka.metadata.BrokerHeartbeatReply; import org.apache.kafka.metadata.BrokerRegistration; import org.apache.kafka.metadata.BrokerRegistrationFencingChange; @@ -791,7 +792,9 @@ private ApiError createTopic(ControllerRequestContext context, for (Entry partEntry : newParts.entrySet()) { int partitionIndex = partEntry.getKey(); PartitionRegistration info = partEntry.getValue(); - records.add(info.toRecord(topicId, partitionIndex, featureControl.metadataVersion().partitionRecordVersion())); + records.add(info.toRecord(topicId, partitionIndex, new ImageWriterOptions.Builder(). + setMetadataVersion(featureControl.metadataVersion()). + build())); } return ApiError.NONE; } @@ -1697,7 +1700,9 @@ void createPartitions(ControllerRequestContext context, " time(s): All brokers are currently fenced or in controlled shutdown."); } records.add(buildPartitionRegistration(replicas, isr) - .toRecord(topicId, partitionId, featureControl.metadataVersion().partitionRecordVersion())); + .toRecord(topicId, partitionId, new ImageWriterOptions.Builder(). + setMetadataVersion(featureControl.metadataVersion()). + build())); partitionId++; } } diff --git a/metadata/src/main/java/org/apache/kafka/image/TopicImage.java b/metadata/src/main/java/org/apache/kafka/image/TopicImage.java index f617b8fb08209..66c22cb7e65bd 100644 --- a/metadata/src/main/java/org/apache/kafka/image/TopicImage.java +++ b/metadata/src/main/java/org/apache/kafka/image/TopicImage.java @@ -68,7 +68,7 @@ public void write(ImageWriter writer, ImageWriterOptions options) { for (Entry entry : partitions.entrySet()) { int partitionId = entry.getKey(); PartitionRegistration partition = entry.getValue(); - writer.write(partition.toRecord(id, partitionId, options.metadataVersion().partitionRecordVersion())); + writer.write(partition.toRecord(id, partitionId, options)); } } diff --git a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java index 65d0652cd75aa..17b0c57543c4f 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java @@ -17,11 +17,13 @@ package org.apache.kafka.metadata; +import org.apache.kafka.common.DirectoryId; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.Uuid; import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState; import org.apache.kafka.common.metadata.PartitionChangeRecord; import org.apache.kafka.common.metadata.PartitionRecord; +import org.apache.kafka.image.writer.ImageWriterOptions; import org.apache.kafka.server.common.ApiMessageAndVersion; import org.slf4j.Logger; @@ -39,6 +41,7 @@ public class PartitionRegistration { */ static public class Builder { private int[] replicas; + private DirectoryId[] directories; private int[] isr; private int[] removingReplicas = Replicas.NONE; private int[] addingReplicas = Replicas.NONE; @@ -54,6 +57,11 @@ public Builder setReplicas(int[] replicas) { return this; } + public Builder setDirectories(DirectoryId[] directories) { + this.directories = directories; + return this; + } + public Builder setIsr(int[] isr) { this.isr = isr; return this; @@ -102,6 +110,8 @@ public Builder setPartitionEpoch(Integer partitionEpoch) { public PartitionRegistration build() { if (replicas == null) { throw new IllegalStateException("You must set replicas."); + } else if (directories != null && directories.length != replicas.length) { + throw new IllegalStateException("The lengths for replicas and directories do not match."); } else if (isr == null) { throw new IllegalStateException("You must set isr."); } else if (removingReplicas == null) { @@ -122,8 +132,13 @@ public PartitionRegistration build() { throw new IllegalStateException("You must set last known elr."); } + if (directories == null) { + directories = DirectoryId.unassignedArray(replicas.length); + } + return new PartitionRegistration( replicas, + directories, isr, removingReplicas, addingReplicas, @@ -138,6 +153,7 @@ public PartitionRegistration build() { } public final int[] replicas; + public final DirectoryId[] directories; public final int[] isr; public final int[] removingReplicas; public final int[] addingReplicas; @@ -154,6 +170,7 @@ public static boolean electionWasClean(int newLeader, int[] isr) { public PartitionRegistration(PartitionRecord record) { this(Replicas.toArray(record.replicas()), + DirectoryId.toArray(record.directories()), Replicas.toArray(record.isr()), Replicas.toArray(record.removingReplicas()), Replicas.toArray(record.addingReplicas()), @@ -165,10 +182,11 @@ public PartitionRegistration(PartitionRecord record) { Replicas.toArray(record.lastKnownELR())); } - private PartitionRegistration(int[] replicas, int[] isr, int[] removingReplicas, + private PartitionRegistration(int[] replicas, DirectoryId[] directories, int[] isr, int[] removingReplicas, int[] addingReplicas, int leader, LeaderRecoveryState leaderRecoveryState, int leaderEpoch, int partitionEpoch, int[] elr, int[] lastKnownElr) { this.replicas = replicas; + this.directories = directories; this.isr = isr; this.removingReplicas = removingReplicas; this.addingReplicas = addingReplicas; @@ -185,6 +203,8 @@ private PartitionRegistration(int[] replicas, int[] isr, int[] removingReplicas, public PartitionRegistration merge(PartitionChangeRecord record) { int[] newReplicas = (record.replicas() == null) ? replicas : Replicas.toArray(record.replicas()); + DirectoryId[] newDirectories = (record.directories() == null) ? + directories : DirectoryId.toArray(record.directories()); int[] newIsr = (record.isr() == null) ? isr : Replicas.toArray(record.isr()); int[] newRemovingReplicas = (record.removingReplicas() == null) ? removingReplicas : Replicas.toArray(record.removingReplicas()); @@ -206,6 +226,7 @@ public PartitionRegistration merge(PartitionChangeRecord record) { int[] newElr = (record.eligibleLeaderReplicas() == null) ? elr : Replicas.toArray(record.eligibleLeaderReplicas()); int[] newLastKnownElr = (record.lastKnownELR() == null) ? lastKnownElr : Replicas.toArray(record.lastKnownELR()); return new PartitionRegistration(newReplicas, + newDirectories, newIsr, newRemovingReplicas, newAddingReplicas, @@ -226,6 +247,12 @@ public String diff(PartitionRegistration prev) { append(" -> ").append(Arrays.toString(replicas)); prefix = ", "; } + if (!Arrays.equals(directories, prev.directories)) { + builder.append(prefix).append("directories: "). + append(Arrays.toString(prev.directories)). + append(" -> ").append(Arrays.toString(directories)); + prefix = ", "; + } if (!Arrays.equals(isr, prev.isr)) { builder.append(prefix).append("isr: "). append(Arrays.toString(prev.isr)). @@ -298,7 +325,7 @@ public int preferredReplica() { return replicas.length == 0 ? LeaderConstants.NO_LEADER : replicas[0]; } - public ApiMessageAndVersion toRecord(Uuid topicId, int partitionId, short version) { + public ApiMessageAndVersion toRecord(Uuid topicId, int partitionId, ImageWriterOptions options) { PartitionRecord record = new PartitionRecord(). setPartitionId(partitionId). setTopicId(topicId). @@ -310,11 +337,22 @@ public ApiMessageAndVersion toRecord(Uuid topicId, int partitionId, short versio setLeaderRecoveryState(leaderRecoveryState.value()). setLeaderEpoch(leaderEpoch). setPartitionEpoch(partitionEpoch); - if (version > 0) { + if (options.metadataVersion().isElrSupported()) { record.setEligibleLeaderReplicas(Replicas.toList(elr)). setLastKnownELR(Replicas.toList(lastKnownElr)); } - return new ApiMessageAndVersion(record, version); + if (options.metadataVersion().isDirectoryAssignmentSupported()) { + record.setDirectories(DirectoryId.toList(directories)); + } else { + for (int i = 0; i < directories.length; i++) { + Uuid directory = directories[i]; + if (!DirectoryId.UNASSIGNED.equals(directory)) { + options.handleLoss("the directory assignment state of one or more replicas"); + break; + } + } + } + return new ApiMessageAndVersion(record, options.metadataVersion().partitionRecordVersion()); } public LeaderAndIsrPartitionState toLeaderAndIsrPartitionState(TopicPartition tp, diff --git a/metadata/src/main/resources/common/metadata/BrokerRegistrationChangeRecord.json b/metadata/src/main/resources/common/metadata/BrokerRegistrationChangeRecord.json index 81bebaaff276c..2824c51fcf1c8 100644 --- a/metadata/src/main/resources/common/metadata/BrokerRegistrationChangeRecord.json +++ b/metadata/src/main/resources/common/metadata/BrokerRegistrationChangeRecord.json @@ -13,11 +13,13 @@ // See the License for the specific language governing permissions and // limitations under the License. +// Version 1 adds InControlledShutdown +// Version 2 adds LogDirs { "apiKey": 17, "type": "metadata", "name": "BrokerRegistrationChangeRecord", - "validVersions": "0-1", + "validVersions": "0-2", "flexibleVersions": "0+", "fields": [ { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId", @@ -27,6 +29,8 @@ { "name": "Fenced", "type": "int8", "versions": "0+", "taggedVersions": "0+", "tag": 0, "about": "-1 if the broker has been unfenced, 0 if no change, 1 if the broker has been fenced." }, { "name": "InControlledShutdown", "type": "int8", "versions": "1+", "taggedVersions": "1+", "tag": 1, - "about": "0 if no change, 1 if the broker is in controlled shutdown." } + "about": "0 if no change, 1 if the broker is in controlled shutdown." }, + { "name": "LogDirs", "type": "[]uuid", "versions": "2+", "taggedVersions": "2+", "tag": "2", + "about": "Log directories configured in this broker which are available." } ] } diff --git a/metadata/src/main/resources/common/metadata/PartitionChangeRecord.json b/metadata/src/main/resources/common/metadata/PartitionChangeRecord.json index 9b58269b35b88..0471b9875f6f6 100644 --- a/metadata/src/main/resources/common/metadata/PartitionChangeRecord.json +++ b/metadata/src/main/resources/common/metadata/PartitionChangeRecord.json @@ -17,9 +17,9 @@ "apiKey": 5, "type": "metadata", "name": "PartitionChangeRecord", - "validVersions": "0-1", - // Version 1 implements Eligiable Leader Replicas and LastKnownELR as described in KIP-966. - // + // Version 1 implements Eligible Leader Replicas and LastKnownELR as described in KIP-966. + // Version 2 adds Directories for KIP-858 + "validVersions": "0-2", "flexibleVersions": "0+", "fields": [ { "name": "PartitionId", "type": "int32", "versions": "0+", "default": "-1", @@ -48,6 +48,9 @@ "about": "null if the ELR didn't change; the new eligible leader replicas otherwise." }, { "name": "LastKnownELR", "type": "[]int32", "default": "null", "entityType": "brokerId", "versions": "1+", "nullableVersions": "1+", "taggedVersions": "1+", "tag": 7, - "about": "null if the LastKnownELR didn't change; the last known eligible leader replicas otherwise." } + "about": "null if the LastKnownELR didn't change; the last known eligible leader replicas otherwise." }, + { "name": "Directories", "type": "[]uuid", "default": "null", + "versions": "2+", "nullableVersions": "2+", "taggedVersions": "2+", "tag": 8, + "about": "null if the log dirs didn't change; the new log directory for each replica otherwise."} ] } diff --git a/metadata/src/main/resources/common/metadata/PartitionRecord.json b/metadata/src/main/resources/common/metadata/PartitionRecord.json index e65070968ed68..f8b6c1b7a8bcf 100644 --- a/metadata/src/main/resources/common/metadata/PartitionRecord.json +++ b/metadata/src/main/resources/common/metadata/PartitionRecord.json @@ -17,9 +17,9 @@ "apiKey": 3, "type": "metadata", "name": "PartitionRecord", - "validVersions": "0-1", - // Version 1 implements Eligiable Leader Replicas and LastKnownELR as described in KIP-966. - // + // Version 1 implements Eligible Leader Replicas and LastKnownELR as described in KIP-966. + // Version 2 adds Directories for KIP-858 + "validVersions": "0-2", "flexibleVersions": "0+", "fields": [ { "name": "PartitionId", "type": "int32", "versions": "0+", "default": "-1", @@ -47,6 +47,8 @@ "about": "The eligible leader replicas of this partition." }, { "name": "LastKnownELR", "type": "[]int32", "default": "null", "entityType": "brokerId", "versions": "1+", "nullableVersions": "1+", "taggedVersions": "1+", "tag": 2, - "about": "The last known eligible leader replicas of this partition." } + "about": "The last known eligible leader replicas of this partition." }, + { "name": "Directories", "type": "[]uuid", "versions": "2+", + "about": "The log directory hosting each replica, sorted in the same exact order as the Replicas field."} ] } diff --git a/metadata/src/main/resources/common/metadata/RegisterBrokerRecord.json b/metadata/src/main/resources/common/metadata/RegisterBrokerRecord.json index 66b740573d5aa..42c1fa2fb9a27 100644 --- a/metadata/src/main/resources/common/metadata/RegisterBrokerRecord.json +++ b/metadata/src/main/resources/common/metadata/RegisterBrokerRecord.json @@ -13,11 +13,14 @@ // See the License for the specific language governing permissions and // limitations under the License. +// Version 1 adds InControlledShutdown +// Version 2 adds IsMigratingZkBroker +// Version 3 adds LogDirs { "apiKey": 0, "type": "metadata", "name": "RegisterBrokerRecord", - "validVersions": "0-2", + "validVersions": "0-3", "flexibleVersions": "0+", "fields": [ { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId", @@ -53,6 +56,8 @@ { "name": "Fenced", "type": "bool", "versions": "0+", "default": "true", "about": "True if the broker is fenced." }, { "name": "InControlledShutdown", "type": "bool", "versions": "1+", "default": "false", - "about": "True if the broker is in controlled shutdown." } + "about": "True if the broker is in controlled shutdown." }, + { "name": "LogDirs", "type": "[]uuid", "versions": "3+", "taggedVersions": "3+", "tag": "0", + "about": "Log directories configured in this broker which are available." } ] } diff --git a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java index 019275adcc2e7..ebe996a04e4d3 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java @@ -17,6 +17,7 @@ package org.apache.kafka.controller; +import org.apache.kafka.common.DirectoryId; import org.apache.kafka.common.Uuid; import org.apache.kafka.common.metadata.PartitionChangeRecord; import org.apache.kafka.common.protocol.types.TaggedFields; @@ -33,6 +34,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.mockito.Mockito; import java.util.Arrays; import java.util.Collections; @@ -49,12 +51,15 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.params.provider.Arguments.arguments; +import static org.mockito.Mockito.when; @Timeout(value = 40) public class PartitionChangeBuilderTest { private static Stream partitionChangeRecordVersions() { - return IntStream.range(PartitionChangeRecord.LOWEST_SUPPORTED_VERSION, PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION + 1).mapToObj(version -> Arguments.of((short) version)); + return IntStream.range(PartitionChangeRecord.LOWEST_SUPPORTED_VERSION, PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION + 1) + .filter(v -> v != PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION) // TODO test latest record version in KAFKA-15514 + .mapToObj(version -> Arguments.of((short) version)); } @Test @@ -63,7 +68,7 @@ public void testChangeRecordIsNoOp() { * to update changeRecordIsNoOp to take into account the new schema or tagged fields. */ // Check that the supported versions haven't changed - assertEquals(1, PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION); + assertEquals(2, PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION); assertEquals(0, PartitionChangeRecord.LOWEST_SUPPORTED_VERSION); // For the latest version check that the number of tagged fields hasn't changed TaggedFields taggedFields = (TaggedFields) PartitionChangeRecord.SCHEMA_0.get(2).def.type; @@ -87,10 +92,20 @@ public void testChangeRecordIsNoOp() { .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value()) ) ); + assertFalse(changeRecordIsNoOp(new PartitionChangeRecord().setDirectories(Arrays.asList( + Uuid.fromString("5JwD0VNXRV2Wr9CCON38Tw"), + Uuid.fromString("zpL1bRzTQXmmgdxlLHOWuw"), + Uuid.fromString("6iGUpAkHQXC6bY0FTcPRDw") + )))); } private static final PartitionRegistration FOO = new PartitionRegistration.Builder(). setReplicas(new int[] {2, 1, 3}). + setDirectories(new DirectoryId[]{ + DirectoryId.fromString("dpdvA5AZSWySmnPFTnu5Kw"), + DirectoryId.fromString("V60B3cglScq3Xk8BX1NxAQ"), + DirectoryId.UNASSIGNED, + }). setIsr(new int[] {2, 1, 3}). setLeader(1). setLeaderRecoveryState(LeaderRecoveryState.RECOVERED). @@ -100,8 +115,15 @@ public void testChangeRecordIsNoOp() { private final static Uuid FOO_ID = Uuid.fromString("FbrrdcfiR-KC2CPSTHaJrg"); + private static MetadataVersion metadataVersionForDirAssignmentInfo() { + MetadataVersion metadataVersion = Mockito.spy(MetadataVersion.latest()); + when(metadataVersion.isDirectoryAssignmentSupported()).thenReturn(true); + return metadataVersion; + } + private static MetadataVersion metadataVersionForPartitionChangeRecordVersion(short version) { - return isElrEnabled(version) ? MetadataVersion.IBP_3_7_IV1 : MetadataVersion.IBP_3_7_IV0; + return isDirAssignmentEnabled(version) ? metadataVersionForDirAssignmentInfo() : + isElrEnabled(version) ? MetadataVersion.IBP_3_7_IV1 : MetadataVersion.IBP_3_7_IV0; } private static PartitionChangeBuilder createFooBuilder(MetadataVersion metadataVersion) { @@ -114,6 +136,12 @@ private static PartitionChangeBuilder createFooBuilder(short version) { private static final PartitionRegistration BAR = new PartitionRegistration.Builder(). setReplicas(new int[] {1, 2, 3, 4}). + setDirectories(new DirectoryId[] { + DirectoryId.UNASSIGNED, + DirectoryId.fromString("X5FnAcIgTheWgTMzeO5WHw"), + DirectoryId.fromString("GtrcdoSOTm2vFMGFeZq0eg"), + DirectoryId.fromString("YcOqPw5ARmeKr1y9W3AkFw"), + }). setIsr(new int[] {1, 2, 3}). setRemovingReplicas(new int[] {1}). setAddingReplicas(new int[] {4}). @@ -129,12 +157,21 @@ private static boolean isElrEnabled(short partitionChangeRecordVersion) { return partitionChangeRecordVersion > 0; } + private static boolean isDirAssignmentEnabled(short partitionChangeRecordVersion) { + return partitionChangeRecordVersion > 1; + } + private static PartitionChangeBuilder createBarBuilder(short version) { return new PartitionChangeBuilder(BAR, BAR_ID, 0, r -> r != 3, metadataVersionForPartitionChangeRecordVersion(version), 2).setEligibleLeaderReplicasEnabled(isElrEnabled(version)); } private static final PartitionRegistration BAZ = new PartitionRegistration.Builder(). setReplicas(new int[] {2, 1, 3}). + setDirectories(new DirectoryId[] { + DirectoryId.fromString("ywnfFpTBTbOsFdZ6uAdOmw"), + DirectoryId.fromString("Th0x70ecRbWvZNNV33jyRA"), + DirectoryId.fromString("j216tuSoQsC9JFd1Z5ZP6w"), + }). setIsr(new int[] {1, 3}). setLeader(3). setLeaderRecoveryState(LeaderRecoveryState.RECOVERED). @@ -150,6 +187,11 @@ private static PartitionChangeBuilder createBazBuilder(short version) { private static final PartitionRegistration OFFLINE = new PartitionRegistration.Builder(). setReplicas(new int[] {2, 1, 3}). + setDirectories(new DirectoryId[]{ + DirectoryId.fromString("iYGgiDV5Sb2EtH6hbgYnCA"), + DirectoryId.fromString("XI2t4qAUSkGlLZSKeEVf8g"), + DirectoryId.fromString("eqRW24kIRlitzQFzmovE0Q") + }). setIsr(new int[] {3}). setLeader(-1). setLeaderRecoveryState(LeaderRecoveryState.RECOVERED). @@ -485,6 +527,11 @@ public void testChangeInLeadershipDoesNotChangeRecoveryState(boolean isLeaderRec LeaderRecoveryState recoveryState = LeaderRecoveryState.RECOVERING; PartitionRegistration registration = new PartitionRegistration.Builder(). setReplicas(new int[] {leaderId, leaderId + 1, leaderId + 2}). + setDirectories(new DirectoryId[] { + DirectoryId.fromString("1sF6XXLkSN2LtDums7CJ8Q"), + DirectoryId.fromString("iaBBVsoHQR6NDKXwliKMqw"), + DirectoryId.fromString("sHaBwjdrR2S3bL4E1RKC8Q") + }). setIsr(new int[] {leaderId}). setLeader(leaderId). setLeaderRecoveryState(recoveryState). @@ -550,6 +597,11 @@ void testUncleanSetsLeaderRecoveringState(boolean isLeaderRecoverySupported, boo int leaderId = 1; PartitionRegistration registration = new PartitionRegistration.Builder(). setReplicas(new int[] {leaderId, leaderId + 1, leaderId + 2}). + setDirectories(new DirectoryId[] { + DirectoryId.fromString("uYpxts0pS4K4bk5XOoXB4g"), + DirectoryId.fromString("kS6fHEqwRYucduWkmvsevw"), + DirectoryId.fromString("De9RqRThQRGjKg3i3yzUxA") + }). setIsr(new int[] {leaderId + 1, leaderId + 2}). setLeader(NO_LEADER). setLeaderRecoveryState(LeaderRecoveryState.RECOVERED). @@ -601,6 +653,12 @@ void testUncleanSetsLeaderRecoveringState(boolean isLeaderRecoverySupported, boo public void testStoppedLeaderIsDemotedAfterReassignmentCompletesEvenIfNoNewEligibleLeaders() { // Set up PartitionRegistration as if there's an ongoing reassignment from [0, 1] to [2, 3] int[] replicas = new int[] {2, 3, 0, 1}; + DirectoryId[] directories = { + DirectoryId.fromString("XCBQClkBSZyphD87QUXzDA"), + DirectoryId.fromString("Or2Rp9tTQOSVuy12hsfmTA"), + DirectoryId.fromString("pThsodMNSwGvljTfc1RNVQ"), + DirectoryId.fromString("d8CGoNJmS5mJdF20tc8P7g") + }; // The ISR starts off with the old replicas int[] isr = new int[] {0, 1}; // We're removing [0, 1] @@ -614,6 +672,7 @@ public void testStoppedLeaderIsDemotedAfterReassignmentCompletesEvenIfNoNewEligi int partitionEpoch = 0; PartitionRegistration part = new PartitionRegistration.Builder(). setReplicas(replicas). + setDirectories(directories). setIsr(isr). setRemovingReplicas(removingReplicas). setAddingReplicas(addingReplicas). @@ -668,6 +727,12 @@ private MetadataVersion leaderRecoveryMetadataVersion(boolean isSupported) { public void testEligibleLeaderReplicas_IsrShrinkBelowMinISR(short version) { PartitionRegistration partition = new PartitionRegistration.Builder() .setReplicas(new int[] {1, 2, 3, 4}) + .setDirectories(new DirectoryId[] { + DirectoryId.fromString("NeQeLdHhSXi4tQGaFcszKA"), + DirectoryId.fromString("LsVrQZ73RSSuEWA8hhqQhg"), + DirectoryId.fromString("0IaY4zXKRR6sROgE8yHfnw"), + DirectoryId.fromString("1WxphfLCSZqMHKK4JMppuw") + }) .setIsr(new int[] {1, 2, 3, 4}) .setLeader(1) .setLeaderRecoveryState(LeaderRecoveryState.RECOVERED) @@ -711,6 +776,12 @@ public void testEligibleLeaderReplicas_IsrShrinkBelowMinISR(short version) { public void testEligibleLeaderReplicas_IsrExpandAboveMinISR(short version) { PartitionRegistration partition = new PartitionRegistration.Builder() .setReplicas(new int[] {1, 2, 3, 4}) + .setDirectories(new DirectoryId[]{ + DirectoryId.fromString("CWgRKBKkToGn1HKzNb2qqQ"), + DirectoryId.fromString("SCnk7zfSQSmlKqvV702d3A"), + DirectoryId.fromString("9tO0QHlJRhimjKfH8m9d8A"), + DirectoryId.fromString("JaaqVOxNT2OGVNCCIFA2JQ") + }) .setIsr(new int[] {1, 2}) .setElr(new int[] {3}) .setLastKnownElr(new int[] {4}) @@ -751,6 +822,12 @@ public void testEligibleLeaderReplicas_IsrExpandAboveMinISR(short version) { public void testEligibleLeaderReplicas_IsrAddNewMemberNotInELR(short version) { PartitionRegistration partition = new PartitionRegistration.Builder() .setReplicas(new int[] {1, 2, 3, 4}) + .setDirectories(new DirectoryId[]{ + DirectoryId.fromString("gPcIwlldQXikdUB3F4GB6w"), + DirectoryId.fromString("gFs7V8mKR66z8e5qwtjIMA"), + DirectoryId.fromString("zKHU2fwrRkuypqTgITl46g"), + DirectoryId.fromString("zEgmBBh8QJGqbBIvzvH7JA") + }) .setIsr(new int[] {1}) .setElr(new int[] {3}) .setLastKnownElr(new int[] {2}) @@ -797,6 +874,12 @@ public void testEligibleLeaderReplicas_IsrAddNewMemberNotInELR(short version) { public void testEligibleLeaderReplicas_RemoveUncleanShutdownReplicasFromElr(short version) { PartitionRegistration partition = new PartitionRegistration.Builder() .setReplicas(new int[] {1, 2, 3, 4}) + .setDirectories(new DirectoryId[] { + DirectoryId.fromString("keB9ssIPRlibyVJT5FcBVA"), + DirectoryId.fromString("FhezfoReTSmHoKxi8wOIOg"), + DirectoryId.fromString("QHtFxu8LShm6RiyAP6PxYg"), + DirectoryId.fromString("tUJOMtvMQkGga30ydluvbQ") + }) .setIsr(new int[] {1}) .setElr(new int[] {2, 3}) .setLastKnownElr(new int[] {}) @@ -838,4 +921,38 @@ public void testEligibleLeaderReplicas_RemoveUncleanShutdownReplicasFromElr(shor assertTrue(Arrays.equals(new int[]{}, partition.lastKnownElr), partition.toString()); } } + + @Test + void testKeepsDirectoriesAfterReassignment() { + PartitionRegistration registration = new PartitionRegistration.Builder(). + setReplicas(new int[] {2, 1, 3}). + setDirectories(new DirectoryId[] { + DirectoryId.fromString("v1PVrX6uS5m8CByXlLfmWg"), + DirectoryId.fromString("iU2znv45Q9yQkOpkTSy3jA"), + DirectoryId.fromString("fM5NKyWTQHqEihjIkUl99Q") + }). + setIsr(new int[] {2, 1, 3}). + setLeader(1). + setLeaderRecoveryState(LeaderRecoveryState.RECOVERED). + setLeaderEpoch(100). + setPartitionEpoch(200). + build(); + Optional built = new PartitionChangeBuilder(registration, FOO_ID, + 0, r -> true, metadataVersionForDirAssignmentInfo(), 2). + setTargetReplicas(Arrays.asList(3, 1, 4)).build(); + Optional expected = Optional.of(new ApiMessageAndVersion( + new PartitionChangeRecord(). + setTopicId(FOO_ID). + setPartitionId(0). + setLeader(1). + setReplicas(Arrays.asList(3, 1, 4)). + setDirectories(Arrays.asList( + DirectoryId.fromString("fM5NKyWTQHqEihjIkUl99Q"), + DirectoryId.fromString("iU2znv45Q9yQkOpkTSy3jA"), + DirectoryId.UNASSIGNED + )), + (short) 2 + )); + assertEquals(expected, built); + } } diff --git a/metadata/src/test/java/org/apache/kafka/controller/metrics/ControllerMetricsChangesTest.java b/metadata/src/test/java/org/apache/kafka/controller/metrics/ControllerMetricsChangesTest.java index d68460febd0af..80867fa896e34 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/metrics/ControllerMetricsChangesTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/metrics/ControllerMetricsChangesTest.java @@ -27,8 +27,10 @@ import org.apache.kafka.common.metadata.PartitionRecord; import org.apache.kafka.image.TopicDelta; import org.apache.kafka.image.TopicImage; +import org.apache.kafka.image.writer.ImageWriterOptions; import org.apache.kafka.metadata.BrokerRegistration; import org.apache.kafka.metadata.PartitionRegistration; +import org.apache.kafka.server.common.MetadataVersion; import org.junit.jupiter.api.Test; import static org.apache.kafka.controller.metrics.ControllerMetricsTestUtils.FakePartitionRegistrationType.NORMAL; @@ -158,17 +160,19 @@ public void testHandleDeletedTopic() { static final TopicDelta TOPIC_DELTA2; static { + ImageWriterOptions options = new ImageWriterOptions.Builder(). + setMetadataVersion(MetadataVersion.IBP_3_7_IV0).build(); // highest MV for PartitionRecord v0 TOPIC_DELTA1 = new TopicDelta(new TopicImage("foo", FOO_ID, Collections.emptyMap())); TOPIC_DELTA1.replay((PartitionRecord) fakePartitionRegistration(NORMAL). - toRecord(FOO_ID, 0, (short) 0).message()); + toRecord(FOO_ID, 0, options).message()); TOPIC_DELTA1.replay((PartitionRecord) fakePartitionRegistration(NORMAL). - toRecord(FOO_ID, 1, (short) 0).message()); + toRecord(FOO_ID, 1, options).message()); TOPIC_DELTA1.replay((PartitionRecord) fakePartitionRegistration(NORMAL). - toRecord(FOO_ID, 2, (short) 0).message()); + toRecord(FOO_ID, 2, options).message()); TOPIC_DELTA1.replay((PartitionRecord) fakePartitionRegistration(NON_PREFERRED_LEADER). - toRecord(FOO_ID, 3, (short) 0).message()); + toRecord(FOO_ID, 3, options).message()); TOPIC_DELTA1.replay((PartitionRecord) fakePartitionRegistration(NON_PREFERRED_LEADER). - toRecord(FOO_ID, 4, (short) 0).message()); + toRecord(FOO_ID, 4, options).message()); TOPIC_DELTA2 = new TopicDelta(TOPIC_DELTA1.apply()); TOPIC_DELTA2.replay(new PartitionChangeRecord(). @@ -176,7 +180,7 @@ public void testHandleDeletedTopic() { setTopicId(FOO_ID). setLeader(1)); TOPIC_DELTA2.replay((PartitionRecord) fakePartitionRegistration(NORMAL). - toRecord(FOO_ID, 5, (short) 0).message()); + toRecord(FOO_ID, 5, options).message()); } @Test diff --git a/metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java b/metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java index b417bebf8acbc..e802a2e82471e 100644 --- a/metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java +++ b/metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java @@ -17,11 +17,13 @@ package org.apache.kafka.image; +import org.apache.kafka.common.DirectoryId; import org.apache.kafka.common.Uuid; import org.apache.kafka.common.metadata.FeatureLevelRecord; import org.apache.kafka.common.metadata.PartitionRecord; import org.apache.kafka.common.metadata.RegisterBrokerRecord; import org.apache.kafka.common.metadata.TopicRecord; +import org.apache.kafka.common.metadata.ZkMigrationStateRecord; import org.apache.kafka.image.writer.ImageWriterOptions; import org.apache.kafka.image.writer.RecordListWriter; import org.apache.kafka.image.writer.UnwritableMetadataException; @@ -33,10 +35,13 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import java.util.function.Consumer; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; @Timeout(value = 40) @@ -126,6 +131,31 @@ public void testPreControlledShutdownStateVersion() throws Throwable { TEST_RECORDS.get(1))); } + @Test + void testDirectoryAssignmentState() { + MetadataVersion outputMetadataVersion = MetadataVersion.IBP_3_7_IV0; + MetadataVersion inputMetadataVersion = spy(outputMetadataVersion); // TODO replace with actual MV after bump for KIP-858 + when(inputMetadataVersion.isDirectoryAssignmentSupported()).thenReturn(true); + PartitionRecord testPartitionRecord = (PartitionRecord) TEST_RECORDS.get(1).message(); + writeWithExpectedLosses(outputMetadataVersion, + Collections.singletonList("the directory assignment state of one or more replicas"), + Arrays.asList( + metadataVersionRecord(inputMetadataVersion), + TEST_RECORDS.get(0), + new ApiMessageAndVersion( + testPartitionRecord.duplicate().setDirectories(Arrays.asList( + DirectoryId.fromString("c7QfSi6xSIGQVh3Qd5RJxA"), + DirectoryId.fromString("rWaCHejCRRiptDMvW5Xw0g"))), + (short) 2)), + Arrays.asList( + metadataVersionRecord(outputMetadataVersion), + new ApiMessageAndVersion(new ZkMigrationStateRecord(), (short) 0), + TEST_RECORDS.get(0), + new ApiMessageAndVersion( + testPartitionRecord.duplicate().setDirectories(Collections.emptyList()), + (short) 0))); + } + private static void writeWithExpectedLosses( MetadataVersion metadataVersion, List expectedLosses, diff --git a/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java b/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java index 5c73bf0109f05..36f2e4ac9f752 100644 --- a/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java +++ b/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java @@ -22,20 +22,26 @@ import net.jqwik.api.ForAll; import net.jqwik.api.Property; import net.jqwik.api.Provide; +import org.apache.kafka.common.DirectoryId; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.Uuid; import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState; import org.apache.kafka.common.metadata.PartitionChangeRecord; import org.apache.kafka.common.metadata.PartitionRecord; +import org.apache.kafka.image.writer.ImageWriterOptions; +import org.apache.kafka.image.writer.UnwritableMetadataException; import org.apache.kafka.server.common.ApiMessageAndVersion; +import org.apache.kafka.server.common.MetadataVersion; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.Timeout; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.stream.IntStream; import java.util.stream.Stream; @@ -44,6 +50,8 @@ import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.when; @Timeout(40) @@ -81,7 +89,8 @@ public void testRecordRoundTrip() { setReplicas(new int[]{1, 2, 3}).setIsr(new int[]{1, 2}).setRemovingReplicas(new int[]{1}).setLeader(1).setLeaderRecoveryState(LeaderRecoveryState.RECOVERED).setLeaderEpoch(0).setPartitionEpoch(0).build(); Uuid topicId = Uuid.fromString("OGdAI5nxT_m-ds3rJMqPLA"); int partitionId = 4; - ApiMessageAndVersion record = registrationA.toRecord(topicId, partitionId, (short) 0); + ApiMessageAndVersion record = registrationA.toRecord(topicId, partitionId, new ImageWriterOptions.Builder(). + setMetadataVersion(MetadataVersion.IBP_3_7_IV0).build()); // highest MV for PartitionRecord v0 PartitionRegistration registrationB = new PartitionRegistration((PartitionRecord) record.message()); assertEquals(registrationA, registrationB); @@ -124,6 +133,7 @@ public void testToLeaderAndIsrPartitionState() { @Test public void testMergePartitionChangeRecordWithReassignmentData() { PartitionRegistration partition0 = new PartitionRegistration.Builder().setReplicas(new int[] {1, 2, 3}). + setDirectories(new DirectoryId[]{DirectoryId.fromString("FbRuu7CeQtq5YFreEzg16g"), DirectoryId.fromString("4rtHTelWSSStAFMODOg3cQ"), DirectoryId.fromString("Id1WXzHURROilVxZWJNZlw")}). setIsr(new int[] {1, 2, 3}).setLeader(1).setLeaderRecoveryState(LeaderRecoveryState.RECOVERED).setLeaderEpoch(100).setPartitionEpoch(200).build(); PartitionRegistration partition1 = partition0.merge(new PartitionChangeRecord(). setRemovingReplicas(Collections.singletonList(3)). @@ -162,6 +172,7 @@ public void testBuilderThrowsIllegalStateExceptionWhenMissingLeader() { setIsr(new int[]{0}). setRemovingReplicas(new int[]{0}). setAddingReplicas(new int[]{0}); + IllegalStateException exception = assertThrows(IllegalStateException.class, () -> builder.build()); assertEquals("You must set leader.", exception.getMessage()); } @@ -250,6 +261,13 @@ public void testBuilderSetsDefaultAddingAndRemovingReplicas() { public void testPartitionRegistrationToRecord(short version) { PartitionRegistration.Builder builder = new PartitionRegistration.Builder(). setReplicas(new int[]{0, 1, 2, 3, 4}). + setDirectories(new DirectoryId[]{ + DirectoryId.UNASSIGNED, + DirectoryId.fromString("KBJBm9GVRAG9Ffe25odmmg"), + DirectoryId.LOST, + DirectoryId.fromString("7DZNT5qBS7yFF7VMMHS7kw"), + DirectoryId.fromString("cJGPUZsMSEqbidOLYLOIXg") + }). setIsr(new int[]{0, 1}). setLeader(0). setLeaderRecoveryState(LeaderRecoveryState.RECOVERED). @@ -268,12 +286,35 @@ public void testPartitionRegistrationToRecord(short version) { setLeaderRecoveryState(LeaderRecoveryState.RECOVERED.value()). setLeaderEpoch(0). setPartitionEpoch(0); + MetadataVersion metadataVersion = spy(MetadataVersion.latest()); + when(metadataVersion.partitionRecordVersion()).thenReturn(version); if (version > 0) { expectRecord. setEligibleLeaderReplicas(Arrays.asList(new Integer[]{2, 3})). setLastKnownELR(Arrays.asList(new Integer[]{4})); + } else { + when(metadataVersion.isElrSupported()).thenReturn(false); + } + if (version > 1) { + expectRecord.setDirectories(Arrays.asList( + DirectoryId.UNASSIGNED, + DirectoryId.fromString("KBJBm9GVRAG9Ffe25odmmg"), + DirectoryId.LOST, + DirectoryId.fromString("7DZNT5qBS7yFF7VMMHS7kw"), + DirectoryId.fromString("cJGPUZsMSEqbidOLYLOIXg") + )); + when(metadataVersion.isDirectoryAssignmentSupported()).thenReturn(true); + } + List exceptions = new ArrayList<>(); + ImageWriterOptions options = new ImageWriterOptions.Builder(). + setMetadataVersion(metadataVersion). + setLossHandler(exceptions::add). + build(); + assertEquals(new ApiMessageAndVersion(expectRecord, version), partitionRegistration.toRecord(topicID, 0, options)); + if (version < 2) { + assertTrue(exceptions.stream(). + anyMatch(e -> e.getMessage().contains("the directory assignment state of one or more replicas"))); } - assertEquals(new ApiMessageAndVersion(expectRecord, version), partitionRegistration.toRecord(topicID, 0, version)); assertEquals(Replicas.toList(Replicas.NONE), Replicas.toList(partitionRegistration.addingReplicas)); } @@ -313,5 +354,7 @@ Arbitrary uniqueSamples() { new PartitionRegistration.Builder().setReplicas(new int[] {1, 2, 3, 4, 5, 6}).setIsr(new int[] {1, 2, 3}).setAddingReplicas(new int[] {4, 5, 6}). setLeader(1).setLeaderRecoveryState(LeaderRecoveryState.RECOVERED).setLeaderEpoch(100).setPartitionEpoch(200).setElr(new int[] {2, 3}).setLastKnownElr(new int[] {1, 2}).build() ); + } + } diff --git a/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java b/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java index b9917730f91f4..b6db2dd3ca8c1 100644 --- a/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java +++ b/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java @@ -16,55 +16,166 @@ */ package org.apache.kafka.common; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; +import java.util.List; +import java.util.Map; import java.util.Set; -public class DirectoryId { +public class DirectoryId extends Uuid { /** - * A UUID that is used to identify new or unknown dir assignments. + * A DirectoryId that is used to identify new or unknown dir assignments. */ - public static final Uuid UNASSIGNED = new Uuid(0L, 0L); + public static final DirectoryId UNASSIGNED = new DirectoryId(0L, 0L); /** - * A UUID that is used to represent unspecified offline dirs. + * A DirectoryId that is used to represent unspecified offline dirs. */ - public static final Uuid LOST = new Uuid(0L, 1L); + public static final DirectoryId LOST = new DirectoryId(0L, 1L); /** - * A UUID that is used to represent and unspecified log directory, + * A DirectoryId that is used to represent and unspecified log directory, * that is expected to have been previously selected to host an * associated replica. This contrasts with {@code UNASSIGNED_DIR}, * which is associated with (typically new) replicas that may not * yet have been placed in any log directory. */ - public static final Uuid MIGRATING = new Uuid(0L, 2L); + public static final DirectoryId MIGRATING = new DirectoryId(0L, 2L); /** * The set of reserved UUIDs that will never be returned by the random method. */ - public static final Set RESERVED; + public static final Set RESERVED; static { - HashSet reserved = new HashSet<>(Uuid.RESERVED); - // The first 100 UUIDs are reserved for future use. + HashSet reserved = new HashSet<>(); + // The first 100 DirectoryIds are reserved for future use. for (long i = 0L; i < 100L; i++) { - reserved.add(new Uuid(0L, i)); + reserved.add(new DirectoryId(0L, i)); } RESERVED = Collections.unmodifiableSet(reserved); } + /** + * Constructs a Directory ID from the underlying 128 bits, + * exactly as a {@link Uuid} is constructed. + */ + private DirectoryId(long mostSigBits, long leastSigBits) { + super(mostSigBits, leastSigBits); + } + + /** + * Creates a DirectoryId based on a base64 string encoding used in the toString() method. + */ + public static DirectoryId fromString(String str) { + return DirectoryId.fromUuid(Uuid.fromString(str)); + } + + /** + * Creates a DirectoryId based on a {@link Uuid}. + */ + public static DirectoryId fromUuid(Uuid uuid) { + return new DirectoryId(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits()); + } + /** * Static factory to generate a directory ID. * * This will not generate a reserved UUID (first 100), or one whose string representation starts with a dash ("-") */ - public static Uuid random() { + public static DirectoryId random() { Uuid uuid = Uuid.randomUuid(); while (RESERVED.contains(uuid) || uuid.toString().startsWith("-")) { uuid = Uuid.randomUuid(); } - return uuid; + return DirectoryId.fromUuid(uuid); + } + + /** + * Convert a list of Uuid to an array of DirectoryId. + * + * @param list The input list + * @return The output array + */ + public static DirectoryId[] toArray(List list) { + if (list == null) return null; + DirectoryId[] array = new DirectoryId[list.size()]; + for (int i = 0; i < list.size(); i++) { + array[i] = DirectoryId.fromUuid(list.get(i)); + } + return array; + } + + /** + * Convert an array of DirectoryIds to a list of Uuid. + * + * @param array The input array + * @return The output list + */ + public static List toList(DirectoryId[] array) { + if (array == null) return null; + List list = new ArrayList<>(array.length); + list.addAll(Arrays.asList(array)); + return list; + } + + /** + * Calculate the new directory information based on an existing replica assignment. + * Replicas for which there already is a directory ID keep the same directory. + * All other replicas get {@link #UNASSIGNED}. + * @param currentReplicas The current replicas, represented by the broker IDs + * @param currentDirectories The current directory information + * @param newReplicas The new replica list + * @return The new directory list + * @throws IllegalArgumentException If currentReplicas and currentDirectories have different lengths, + * or if there are duplicate broker IDs in the replica lists + */ + public static List update(int[] currentReplicas, DirectoryId[] currentDirectories, List newReplicas) { + if (currentReplicas == null) currentReplicas = new int[0]; + if (currentDirectories == null) currentDirectories = new DirectoryId[0]; + Map assignments = buildAssignmentMap(currentReplicas, currentDirectories); + List consolidated = new ArrayList<>(newReplicas.size()); + for (int i = 0; i < newReplicas.size(); i++) { + int newReplica = newReplicas.get(i); + Uuid newDirectory = assignments.getOrDefault(newReplica, UNASSIGNED); + consolidated.add(newDirectory); + } + return consolidated; + } + + /** + * Build a mapping from replica to directory based on two lists of the same size and order. + * @param replicas The replicas, represented by the broker IDs + * @param directories The directory information + * @return A map, linking each replica to its assigned directory + * @throws IllegalArgumentException If replicas and directories have different lengths, + * or if there are duplicate broker IDs in the replica list + */ + private static Map buildAssignmentMap(int[] replicas, DirectoryId[] directories) { + if (replicas.length != directories.length) { + throw new IllegalArgumentException("The lengths for replicas and directories do not match."); + } + Map assignments = new HashMap<>(); + for (int i = 0; i < replicas.length; i++) { + int brokerId = replicas[i]; + Uuid directory = directories[i]; + if (assignments.put(brokerId, directory) != null) { + throw new IllegalArgumentException("Duplicate broker ID in assignment"); + } + } + return assignments; + } + + /** + * Create an array with the specified number of entries set to {@link #UNASSIGNED}. + */ + public static DirectoryId[] unassignedArray(int length) { + DirectoryId[] array = new DirectoryId[length]; + Arrays.fill(array, UNASSIGNED); + return array; } } diff --git a/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java b/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java index 8f22ed582a6c1..b77e0dba0bac3 100644 --- a/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java +++ b/server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java @@ -293,6 +293,10 @@ public boolean isElrSupported() { return this.isAtLeast(IBP_3_7_IV1); } + public boolean isDirectoryAssignmentSupported() { + return false; // TODO: Bump IBP for JBOD support in KRaft + } + public boolean isKRaftSupported() { return this.featureLevel > 0; } @@ -320,7 +324,10 @@ public boolean isMigrationSupported() { } public short registerBrokerRecordVersion() { - if (isMigrationSupported()) { + if (isDirectoryAssignmentSupported()) { + // new logDirs field + return (short) 3; + } else if (isMigrationSupported()) { // new isMigrationZkBroker field return (short) 2; } else if (isInControlledShutdownStateSupported()) { @@ -344,7 +351,9 @@ public boolean isControllerRegistrationSupported() { } public short partitionChangeRecordVersion() { - if (isElrSupported()) { + if (isDirectoryAssignmentSupported()) { + return (short) 2; + } else if (isElrSupported()) { return (short) 1; } else { return (short) 0; @@ -352,7 +361,9 @@ public short partitionChangeRecordVersion() { } public short partitionRecordVersion() { - if (isElrSupported()) { + if (isDirectoryAssignmentSupported()) { + return (short) 2; + } else if (isElrSupported()) { return (short) 1; } else { return (short) 0; diff --git a/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java b/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java new file mode 100644 index 0000000000000..c5a16de923ba4 --- /dev/null +++ b/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java @@ -0,0 +1,95 @@ +/* + * 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; + +import org.junit.jupiter.api.Test; + +import java.util.Arrays; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class DirectoryIdTest { + + @Test + void testToArray() { + assertNull(DirectoryId.toArray(null)); + assertArrayEquals( + new DirectoryId[]{ + DirectoryId.MIGRATING, DirectoryId.fromString("UXyU9i5ARn6W00ON2taeWA") + }, + DirectoryId.toArray(Arrays.asList( + DirectoryId.MIGRATING, DirectoryId.fromString("UXyU9i5ARn6W00ON2taeWA") + )) + ); + } + + @Test + void testToList() { + assertNull(DirectoryId.toList(null)); + assertEquals( + Arrays.asList( + DirectoryId.MIGRATING, DirectoryId.fromString("UXyU9i5ARn6W00ON2taeWA") + ), + DirectoryId.toList(new DirectoryId[]{ + DirectoryId.MIGRATING, DirectoryId.fromString("UXyU9i5ARn6W00ON2taeWA") + }) + ); + } + + @Test + void testUpdate() { + assertThrows(IllegalArgumentException.class, () -> DirectoryId.update( + new int[] {1}, + new DirectoryId[] {DirectoryId.UNASSIGNED, DirectoryId.LOST}, + Arrays.asList(2, 3) + )); + assertEquals( + Arrays.asList( + DirectoryId.fromString("YXY0bQYEQmmyOQ6ZDfGgSQ"), + DirectoryId.fromString("5SZij3DRQgaFbvzR9KooLg"), + DirectoryId.UNASSIGNED + ), + DirectoryId.update( + new int[] {1, 2, 3}, + new DirectoryId[] { + DirectoryId.fromString("MgVK5KSwTxe65eYATaoQrg"), + DirectoryId.fromString("YXY0bQYEQmmyOQ6ZDfGgSQ"), + DirectoryId.fromString("5SZij3DRQgaFbvzR9KooLg") + }, + Arrays.asList(2, 3, 4) + ) + ); + assertEquals( + Arrays.asList( + DirectoryId.UNASSIGNED, + DirectoryId.UNASSIGNED, + DirectoryId.UNASSIGNED + ), + DirectoryId.update( + new int[] {1, 2}, + new DirectoryId[] { + DirectoryId.UNASSIGNED, + DirectoryId.UNASSIGNED + }, + Arrays.asList(1, 2, 3) + ) + ); + } +} From cea4f124982d82de9f1b9a8d21893924cb57c85e Mon Sep 17 00:00:00 2001 From: Igor Soarez Date: Mon, 30 Oct 2023 20:41:20 +0000 Subject: [PATCH 2/6] Changes from PR review --- .../controller/PartitionChangeBuilder.java | 2 +- .../kafka/metadata/PartitionRegistration.java | 9 +-- .../PartitionChangeBuilderTest.java | 56 +++++++++++++------ .../org/apache/kafka/common/DirectoryId.java | 13 ++--- .../apache/kafka/common/DirectoryIdTest.java | 46 +++++++++++++-- 5 files changed, 89 insertions(+), 37 deletions(-) diff --git a/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java b/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java index d0e2b91cb7ba5..c269dd4739482 100644 --- a/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java +++ b/metadata/src/main/java/org/apache/kafka/controller/PartitionChangeBuilder.java @@ -385,7 +385,7 @@ public Optional build() { private void setAssignmentChanges(PartitionChangeRecord record) { if (!targetReplicas.isEmpty() && !targetReplicas.equals(Replicas.toList(partition.replicas))) { if (metadataVersion.isDirectoryAssignmentSupported()) { - record.setDirectories(DirectoryId.update(partition.replicas, partition.directories, targetReplicas)); + record.setDirectories(DirectoryId.createDirectoriesFrom(partition.replicas, partition.directories, targetReplicas)); } record.setReplicas(targetReplicas); } diff --git a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java index 17b0c57543c4f..bd038c98557d6 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java @@ -132,10 +132,6 @@ public PartitionRegistration build() { throw new IllegalStateException("You must set last known elr."); } - if (directories == null) { - directories = DirectoryId.unassignedArray(replicas.length); - } - return new PartitionRegistration( replicas, directories, @@ -186,7 +182,7 @@ private PartitionRegistration(int[] replicas, DirectoryId[] directories, int[] i int[] addingReplicas, int leader, LeaderRecoveryState leaderRecoveryState, int leaderEpoch, int partitionEpoch, int[] elr, int[] lastKnownElr) { this.replicas = replicas; - this.directories = directories; + this.directories = directories != null ? directories : DirectoryId.unassignedArray(replicas.length); this.isr = isr; this.removingReplicas = removingReplicas; this.addingReplicas = addingReplicas; @@ -345,8 +341,7 @@ public ApiMessageAndVersion toRecord(Uuid topicId, int partitionId, ImageWriterO record.setDirectories(DirectoryId.toList(directories)); } else { for (int i = 0; i < directories.length; i++) { - Uuid directory = directories[i]; - if (!DirectoryId.UNASSIGNED.equals(directory)) { + if (!DirectoryId.UNASSIGNED.equals(directories[i])) { options.handleLoss("the directory assignment state of one or more replicas"); break; } diff --git a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java index ebe996a04e4d3..e1de7e7beee6b 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java @@ -38,6 +38,7 @@ import java.util.Arrays; import java.util.Collections; +import java.util.Map; import java.util.Optional; import java.util.function.IntPredicate; import java.util.stream.IntStream; @@ -57,9 +58,7 @@ @Timeout(value = 40) public class PartitionChangeBuilderTest { private static Stream partitionChangeRecordVersions() { - return IntStream.range(PartitionChangeRecord.LOWEST_SUPPORTED_VERSION, PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION + 1) - .filter(v -> v != PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION) // TODO test latest record version in KAFKA-15514 - .mapToObj(version -> Arguments.of((short) version)); + return IntStream.range(PartitionChangeRecord.LOWEST_SUPPORTED_VERSION, PartitionChangeRecord.HIGHEST_SUPPORTED_VERSION + 1).mapToObj(version -> Arguments.of((short) version)); } @Test @@ -115,6 +114,7 @@ public void testChangeRecordIsNoOp() { private final static Uuid FOO_ID = Uuid.fromString("FbrrdcfiR-KC2CPSTHaJrg"); + // TODO remove this after after MetadataVersion bump for KIP-858 private static MetadataVersion metadataVersionForDirAssignmentInfo() { MetadataVersion metadataVersion = Mockito.spy(MetadataVersion.latest()); when(metadataVersion.isDirectoryAssignmentSupported()).thenReturn(true); @@ -385,26 +385,34 @@ public void testIsrChangeAndLeaderChange(short version) { @ParameterizedTest @MethodSource("partitionChangeRecordVersions") public void testReassignmentRearrangesReplicas(short version) { - assertEquals(Optional.of(new ApiMessageAndVersion(new PartitionChangeRecord(). + PartitionChangeRecord expectedRecord = new PartitionChangeRecord(). setTopicId(FOO_ID). setPartitionId(0). - setReplicas(Arrays.asList(3, 2, 1)), - version)), + setReplicas(Arrays.asList(3, 2, 1)); + if (version > 1) { + Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); + expectedRecord.setDirectories(Arrays.asList(dirs.get(3), dirs.get(2), dirs.get(1))); + } + assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), createFooBuilder(version).setTargetReplicas(Arrays.asList(3, 2, 1)).build()); } @ParameterizedTest @MethodSource("partitionChangeRecordVersions") public void testIsrEnlargementCompletesReassignment(short version) { - assertEquals(Optional.of(new ApiMessageAndVersion(new PartitionChangeRecord(). + PartitionChangeRecord expectedRecord = new PartitionChangeRecord(). setTopicId(BAR_ID). setPartitionId(0). setReplicas(Arrays.asList(2, 3, 4)). setIsr(Arrays.asList(2, 3, 4)). setLeader(2). setRemovingReplicas(Collections.emptyList()). - setAddingReplicas(Collections.emptyList()), - version)), + setAddingReplicas(Collections.emptyList()); + if (version > 1) { + Map dirs = DirectoryId.createAssignmentMap(BAR.replicas, BAR.directories); + expectedRecord.setDirectories(Arrays.asList(dirs.get(2), dirs.get(3), dirs.get(4))); + } + assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), createBarBuilder(version).setTargetIsrWithBrokerStates(AlterPartitionRequest.newIsrToSimpleNewIsrWithBrokerEpochs(Arrays.asList(1, 2, 3, 4))).build()); } @@ -414,14 +422,18 @@ public void testRevertReassignment(short version) { PartitionReassignmentRevert revert = new PartitionReassignmentRevert(BAR); assertEquals(Arrays.asList(1, 2, 3), revert.replicas()); assertEquals(Arrays.asList(1, 2, 3), revert.isr()); - assertEquals(Optional.of(new ApiMessageAndVersion(new PartitionChangeRecord(). + PartitionChangeRecord expectedRecord = new PartitionChangeRecord(). setTopicId(BAR_ID). setPartitionId(0). setReplicas(Arrays.asList(1, 2, 3)). setLeader(1). setRemovingReplicas(Collections.emptyList()). - setAddingReplicas(Collections.emptyList()), - version)), + setAddingReplicas(Collections.emptyList()); + if (version > 1) { + Map dirs = DirectoryId.createAssignmentMap(BAR.replicas, BAR.directories); + expectedRecord.setDirectories(Arrays.asList(dirs.get(1), dirs.get(2), dirs.get(3))); + } + assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), createBarBuilder(version). setTargetReplicas(revert.replicas()). setTargetIsrWithBrokerStates(AlterPartitionRequest.newIsrToSimpleNewIsrWithBrokerEpochs(revert.isr())). @@ -438,13 +450,17 @@ public void testRemovingReplicaReassignment(short version) { assertEquals(Collections.singletonList(3), replicas.removing()); assertEquals(Collections.emptyList(), replicas.adding()); assertEquals(Arrays.asList(1, 2, 3), replicas.replicas()); - assertEquals(Optional.of(new ApiMessageAndVersion(new PartitionChangeRecord(). + PartitionChangeRecord expectedRecord = new PartitionChangeRecord(). setTopicId(FOO_ID). setPartitionId(0). setReplicas(Arrays.asList(1, 2)). setIsr(Arrays.asList(2, 1)). - setLeader(1), - version)), + setLeader(1); + if (version > 1) { + Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); + expectedRecord.setDirectories(Arrays.asList(dirs.get(1), dirs.get(2))); + } + assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), createFooBuilder(version). setTargetReplicas(replicas.replicas()). setTargetRemoving(replicas.removing()). @@ -459,12 +475,16 @@ public void testAddingReplicaReassignment(short version) { assertEquals(Collections.emptyList(), replicas.removing()); assertEquals(Collections.singletonList(4), replicas.adding()); assertEquals(Arrays.asList(1, 2, 3, 4), replicas.replicas()); - assertEquals(Optional.of(new ApiMessageAndVersion(new PartitionChangeRecord(). + PartitionChangeRecord expectedRecord = new PartitionChangeRecord(). setTopicId(FOO_ID). setPartitionId(0). setReplicas(Arrays.asList(1, 2, 3, 4)). - setAddingReplicas(Collections.singletonList(4)), - version)), + setAddingReplicas(Collections.singletonList(4)); + if (version > 1) { + Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); + expectedRecord.setDirectories(Arrays.asList(dirs.get(1), dirs.get(2), dirs.get(3), DirectoryId.UNASSIGNED)); + } + assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), createFooBuilder(version). setTargetReplicas(replicas.replicas()). setTargetAdding(replicas.adding()). diff --git a/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java b/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java index b6db2dd3ca8c1..6c22d70f283ee 100644 --- a/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java +++ b/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java @@ -134,13 +134,12 @@ public static List toList(DirectoryId[] array) { * @throws IllegalArgumentException If currentReplicas and currentDirectories have different lengths, * or if there are duplicate broker IDs in the replica lists */ - public static List update(int[] currentReplicas, DirectoryId[] currentDirectories, List newReplicas) { + public static List createDirectoriesFrom(int[] currentReplicas, DirectoryId[] currentDirectories, List newReplicas) { if (currentReplicas == null) currentReplicas = new int[0]; if (currentDirectories == null) currentDirectories = new DirectoryId[0]; - Map assignments = buildAssignmentMap(currentReplicas, currentDirectories); + Map assignments = createAssignmentMap(currentReplicas, currentDirectories); List consolidated = new ArrayList<>(newReplicas.size()); - for (int i = 0; i < newReplicas.size(); i++) { - int newReplica = newReplicas.get(i); + for (int newReplica : newReplicas) { Uuid newDirectory = assignments.getOrDefault(newReplica, UNASSIGNED); consolidated.add(newDirectory); } @@ -155,14 +154,14 @@ public static List update(int[] currentReplicas, DirectoryId[] currentDire * @throws IllegalArgumentException If replicas and directories have different lengths, * or if there are duplicate broker IDs in the replica list */ - private static Map buildAssignmentMap(int[] replicas, DirectoryId[] directories) { + public static Map createAssignmentMap(int[] replicas, DirectoryId[] directories) { if (replicas.length != directories.length) { throw new IllegalArgumentException("The lengths for replicas and directories do not match."); } - Map assignments = new HashMap<>(); + Map assignments = new HashMap<>(); for (int i = 0; i < replicas.length; i++) { int brokerId = replicas[i]; - Uuid directory = directories[i]; + DirectoryId directory = directories[i]; if (assignments.put(brokerId, directory) != null) { throw new IllegalArgumentException("Duplicate broker ID in assignment"); } diff --git a/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java b/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java index c5a16de923ba4..8fe7ec2ea19d4 100644 --- a/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java +++ b/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java @@ -19,14 +19,30 @@ import org.junit.jupiter.api.Test; import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Set; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; public class DirectoryIdTest { + @Test + void testReserved() { + Set seen = new HashSet<>(100); + for (DirectoryId reservedId : DirectoryId.RESERVED) { + assertEquals(0L, reservedId.getMostSignificantBits(), "Unexpected reserved msb value"); + long lsb = reservedId.getLeastSignificantBits(); + assertTrue(lsb >= 0 && lsb < 100L, "Unexpected reserved lsb value"); + assertTrue(seen.add(lsb), "Duplicate reserved value"); + } + assertEquals(100, DirectoryId.RESERVED.size()); + } + @Test void testToArray() { assertNull(DirectoryId.toArray(null)); @@ -54,8 +70,8 @@ void testToList() { } @Test - void testUpdate() { - assertThrows(IllegalArgumentException.class, () -> DirectoryId.update( + void testCreateDirectoriesFrom() { + assertThrows(IllegalArgumentException.class, () -> DirectoryId.createDirectoriesFrom( new int[] {1}, new DirectoryId[] {DirectoryId.UNASSIGNED, DirectoryId.LOST}, Arrays.asList(2, 3) @@ -66,7 +82,7 @@ void testUpdate() { DirectoryId.fromString("5SZij3DRQgaFbvzR9KooLg"), DirectoryId.UNASSIGNED ), - DirectoryId.update( + DirectoryId.createDirectoriesFrom( new int[] {1, 2, 3}, new DirectoryId[] { DirectoryId.fromString("MgVK5KSwTxe65eYATaoQrg"), @@ -82,7 +98,7 @@ void testUpdate() { DirectoryId.UNASSIGNED, DirectoryId.UNASSIGNED ), - DirectoryId.update( + DirectoryId.createDirectoriesFrom( new int[] {1, 2}, new DirectoryId[] { DirectoryId.UNASSIGNED, @@ -92,4 +108,26 @@ void testUpdate() { ) ); } + + @Test + void testCreateAssignmentMap() { + assertThrows(IllegalArgumentException.class, + () -> DirectoryId.createAssignmentMap(new int[]{1, 2}, DirectoryId.unassignedArray(3))); + assertEquals( + new HashMap() {{ + put(1, DirectoryId.fromString("upjfkCrUR9GNn1i94ip1wg")); + put(2, DirectoryId.fromString("bCF3l0RIQjOKhUqgbivHZA")); + put(3, DirectoryId.fromString("Fg3mFhcVQlqCWRk4dZazxw")); + put(4, DirectoryId.fromString("bv9TEYi4TqOm52hLmrxT5w")); + }}, + DirectoryId.createAssignmentMap( + new int[] {1, 2, 3, 4}, + new DirectoryId[] { + DirectoryId.fromString("upjfkCrUR9GNn1i94ip1wg"), + DirectoryId.fromString("bCF3l0RIQjOKhUqgbivHZA"), + DirectoryId.fromString("Fg3mFhcVQlqCWRk4dZazxw"), + DirectoryId.fromString("bv9TEYi4TqOm52hLmrxT5w") + }) + ); + } } From e280bbba492a40e391b55129bc92ee809ee459a4 Mon Sep 17 00:00:00 2001 From: Igor Soarez Date: Mon, 30 Oct 2023 20:51:50 +0000 Subject: [PATCH 3/6] Check for invalid record --- .../apache/kafka/metadata/PartitionRegistration.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java index bd038c98557d6..977a310caac42 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java @@ -20,6 +20,7 @@ import org.apache.kafka.common.DirectoryId; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.Uuid; +import org.apache.kafka.common.errors.InvalidMetadataException; import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState; import org.apache.kafka.common.metadata.PartitionChangeRecord; import org.apache.kafka.common.metadata.PartitionRecord; @@ -28,6 +29,7 @@ import org.slf4j.Logger; import java.util.Arrays; +import java.util.List; import java.util.Objects; import static org.apache.kafka.metadata.LeaderConstants.NO_LEADER; @@ -164,9 +166,16 @@ public static boolean electionWasClean(int newLeader, int[] isr) { return newLeader == NO_LEADER || Replicas.contains(isr, newLeader); } + private static List checkDirectories(PartitionRecord record) { + if (record.replicas().size() != record.directories().size()) { + throw new InvalidMetadataException("The lengths for replicas and directories do not match: " + record); + } + return record.directories(); + } + public PartitionRegistration(PartitionRecord record) { this(Replicas.toArray(record.replicas()), - DirectoryId.toArray(record.directories()), + DirectoryId.toArray(checkDirectories(record)), Replicas.toArray(record.isr()), Replicas.toArray(record.removingReplicas()), Replicas.toArray(record.addingReplicas()), From d05e23ee311eb201c2f74cc3145578c6e12c23ce Mon Sep 17 00:00:00 2001 From: Igor Soarez Date: Tue, 31 Oct 2023 11:22:25 +0000 Subject: [PATCH 4/6] Use specific exception invalid metadata type --- checkstyle/import-control-metadata.xml | 3 ++ .../errors/InvalidMetadataException.java | 2 +- .../InvalidReplicaDirectoriesException.java | 36 +++++++++++++++++++ .../kafka/metadata/PartitionRegistration.java | 17 ++++++--- 4 files changed, 52 insertions(+), 6 deletions(-) create mode 100644 metadata/src/main/java/org/apache/kafka/common/errors/InvalidReplicaDirectoriesException.java diff --git a/checkstyle/import-control-metadata.xml b/checkstyle/import-control-metadata.xml index 3fe650c20f83d..d4643c19979ef 100644 --- a/checkstyle/import-control-metadata.xml +++ b/checkstyle/import-control-metadata.xml @@ -61,6 +61,9 @@ + + + diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java index f8ae485c99282..e6663dbf185d0 100644 --- a/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java +++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java @@ -19,7 +19,7 @@ /** * An exception that may indicate the client's metadata is out of date */ -public class InvalidMetadataException extends RetriableException { +public abstract class InvalidMetadataException extends RetriableException { private static final long serialVersionUID = 1L; diff --git a/metadata/src/main/java/org/apache/kafka/common/errors/InvalidReplicaDirectoriesException.java b/metadata/src/main/java/org/apache/kafka/common/errors/InvalidReplicaDirectoriesException.java new file mode 100644 index 0000000000000..c83a8b5bd4a79 --- /dev/null +++ b/metadata/src/main/java/org/apache/kafka/common/errors/InvalidReplicaDirectoriesException.java @@ -0,0 +1,36 @@ +/* + * 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.errors; + +import org.apache.kafka.common.metadata.PartitionChangeRecord; +import org.apache.kafka.common.metadata.PartitionRecord; + +/** + * A record was encountered where the number of directories does not match the number of replicas. + */ +public class InvalidReplicaDirectoriesException extends InvalidMetadataException { + + private static final long serialVersionUID = 1L; + + public InvalidReplicaDirectoriesException(PartitionRecord record) { + super("The lengths for replicas and directories do not match: " + record); + } + + public InvalidReplicaDirectoriesException(PartitionChangeRecord record) { + super("The lengths for replicas and directories do not match: " + record); + } +} diff --git a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java index 977a310caac42..31167b4c57850 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java @@ -20,7 +20,7 @@ import org.apache.kafka.common.DirectoryId; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.Uuid; -import org.apache.kafka.common.errors.InvalidMetadataException; +import org.apache.kafka.common.errors.InvalidReplicaDirectoriesException; import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState; import org.apache.kafka.common.metadata.PartitionChangeRecord; import org.apache.kafka.common.metadata.PartitionRecord; @@ -167,8 +167,15 @@ public static boolean electionWasClean(int newLeader, int[] isr) { } private static List checkDirectories(PartitionRecord record) { - if (record.replicas().size() != record.directories().size()) { - throw new InvalidMetadataException("The lengths for replicas and directories do not match: " + record); + if (record.directories() != null && !record.directories().isEmpty() && record.replicas().size() != record.directories().size()) { + throw new InvalidReplicaDirectoriesException(record); + } + return record.directories(); + } + + private static List checkDirectories(PartitionChangeRecord record) { + if (record.replicas() != null && record.directories() != null && !record.directories().isEmpty() && record.replicas().size() != record.directories().size()) { + throw new InvalidReplicaDirectoriesException(record); } return record.directories(); } @@ -191,7 +198,7 @@ private PartitionRegistration(int[] replicas, DirectoryId[] directories, int[] i int[] addingReplicas, int leader, LeaderRecoveryState leaderRecoveryState, int leaderEpoch, int partitionEpoch, int[] elr, int[] lastKnownElr) { this.replicas = replicas; - this.directories = directories != null ? directories : DirectoryId.unassignedArray(replicas.length); + this.directories = directories != null && directories.length > 0 ? directories : DirectoryId.unassignedArray(replicas.length); this.isr = isr; this.removingReplicas = removingReplicas; this.addingReplicas = addingReplicas; @@ -209,7 +216,7 @@ public PartitionRegistration merge(PartitionChangeRecord record) { int[] newReplicas = (record.replicas() == null) ? replicas : Replicas.toArray(record.replicas()); DirectoryId[] newDirectories = (record.directories() == null) ? - directories : DirectoryId.toArray(record.directories()); + directories : DirectoryId.toArray(checkDirectories(record)); int[] newIsr = (record.isr() == null) ? isr : Replicas.toArray(record.isr()); int[] newRemovingReplicas = (record.removingReplicas() == null) ? removingReplicas : Replicas.toArray(record.removingReplicas()); From 04a79cd4cec478e4cfece50c50fe4abd7f8c4e50 Mon Sep 17 00:00:00 2001 From: Igor Soarez Date: Tue, 31 Oct 2023 18:15:08 +0000 Subject: [PATCH 5/6] Revert use of DirectoryId as a separate type --- .../java/org/apache/kafka/common/Uuid.java | 30 +++++ .../org/apache/kafka/common/UuidTest.java | 28 +++++ .../kafka/metadata/PartitionRegistration.java | 16 +-- .../PartitionChangeBuilderTest.java | 118 +++++++++--------- .../kafka/image/ImageDowngradeTest.java | 5 +- .../metadata/PartitionRegistrationTest.java | 16 +-- .../org/apache/kafka/common/DirectoryId.java | 92 ++++---------- .../apache/kafka/common/DirectoryIdTest.java | 66 +++------- 8 files changed, 175 insertions(+), 196 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/Uuid.java b/clients/src/main/java/org/apache/kafka/common/Uuid.java index b39d52eb96f14..45e2b9f1d8fb2 100644 --- a/clients/src/main/java/org/apache/kafka/common/Uuid.java +++ b/clients/src/main/java/org/apache/kafka/common/Uuid.java @@ -17,10 +17,12 @@ package org.apache.kafka.common; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.Arrays; import java.util.Base64; import java.util.Collections; import java.util.HashSet; +import java.util.List; import java.util.Set; /** @@ -169,4 +171,32 @@ public int compareTo(Uuid other) { return 0; } } + + /** + * Convert a list of Uuid to an array of Uuid. + * + * @param list The input list + * @return The output array + */ + public static Uuid[] toArray(List list) { + if (list == null) return null; + Uuid[] array = new Uuid[list.size()]; + for (int i = 0; i < list.size(); i++) { + array[i] = list.get(i); + } + return array; + } + + /** + * Convert an array of Uuids to a list of Uuid. + * + * @param array The input array + * @return The output list + */ + public static List toList(Uuid[] array) { + if (array == null) return null; + List list = new ArrayList<>(array.length); + list.addAll(Arrays.asList(array)); + return list; + } } diff --git a/clients/src/test/java/org/apache/kafka/common/UuidTest.java b/clients/src/test/java/org/apache/kafka/common/UuidTest.java index 67d841925cef9..f5067a953cd0d 100644 --- a/clients/src/test/java/org/apache/kafka/common/UuidTest.java +++ b/clients/src/test/java/org/apache/kafka/common/UuidTest.java @@ -19,11 +19,14 @@ import org.junit.jupiter.api.RepeatedTest; import org.junit.jupiter.api.Test; +import java.util.Arrays; import java.util.Base64; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; public class UuidTest { @@ -108,4 +111,29 @@ public void testFromStringWithInvalidInput() { assertThrows(IllegalArgumentException.class, () -> Uuid.fromString(undersizeString)); } + @Test + void testToArray() { + assertNull(Uuid.toArray(null)); + assertArrayEquals( + new Uuid[]{ + Uuid.ZERO_UUID, Uuid.fromString("UXyU9i5ARn6W00ON2taeWA") + }, + Uuid.toArray(Arrays.asList( + Uuid.ZERO_UUID, Uuid.fromString("UXyU9i5ARn6W00ON2taeWA") + )) + ); + } + + @Test + void testToList() { + assertNull(Uuid.toList(null)); + assertEquals( + Arrays.asList( + Uuid.ZERO_UUID, Uuid.fromString("UXyU9i5ARn6W00ON2taeWA") + ), + Uuid.toList(new Uuid[]{ + Uuid.ZERO_UUID, Uuid.fromString("UXyU9i5ARn6W00ON2taeWA") + }) + ); + } } diff --git a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java index 31167b4c57850..034c0b8bdb00b 100644 --- a/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java +++ b/metadata/src/main/java/org/apache/kafka/metadata/PartitionRegistration.java @@ -43,7 +43,7 @@ public class PartitionRegistration { */ static public class Builder { private int[] replicas; - private DirectoryId[] directories; + private Uuid[] directories; private int[] isr; private int[] removingReplicas = Replicas.NONE; private int[] addingReplicas = Replicas.NONE; @@ -59,7 +59,7 @@ public Builder setReplicas(int[] replicas) { return this; } - public Builder setDirectories(DirectoryId[] directories) { + public Builder setDirectories(Uuid[] directories) { this.directories = directories; return this; } @@ -151,7 +151,7 @@ public PartitionRegistration build() { } public final int[] replicas; - public final DirectoryId[] directories; + public final Uuid[] directories; public final int[] isr; public final int[] removingReplicas; public final int[] addingReplicas; @@ -182,7 +182,7 @@ private static List checkDirectories(PartitionChangeRecord record) { public PartitionRegistration(PartitionRecord record) { this(Replicas.toArray(record.replicas()), - DirectoryId.toArray(checkDirectories(record)), + Uuid.toArray(checkDirectories(record)), Replicas.toArray(record.isr()), Replicas.toArray(record.removingReplicas()), Replicas.toArray(record.addingReplicas()), @@ -194,7 +194,7 @@ public PartitionRegistration(PartitionRecord record) { Replicas.toArray(record.lastKnownELR())); } - private PartitionRegistration(int[] replicas, DirectoryId[] directories, int[] isr, int[] removingReplicas, + private PartitionRegistration(int[] replicas, Uuid[] directories, int[] isr, int[] removingReplicas, int[] addingReplicas, int leader, LeaderRecoveryState leaderRecoveryState, int leaderEpoch, int partitionEpoch, int[] elr, int[] lastKnownElr) { this.replicas = replicas; @@ -215,8 +215,8 @@ private PartitionRegistration(int[] replicas, DirectoryId[] directories, int[] i public PartitionRegistration merge(PartitionChangeRecord record) { int[] newReplicas = (record.replicas() == null) ? replicas : Replicas.toArray(record.replicas()); - DirectoryId[] newDirectories = (record.directories() == null) ? - directories : DirectoryId.toArray(checkDirectories(record)); + Uuid[] newDirectories = (record.directories() == null) ? + directories : Uuid.toArray(checkDirectories(record)); int[] newIsr = (record.isr() == null) ? isr : Replicas.toArray(record.isr()); int[] newRemovingReplicas = (record.removingReplicas() == null) ? removingReplicas : Replicas.toArray(record.removingReplicas()); @@ -354,7 +354,7 @@ public ApiMessageAndVersion toRecord(Uuid topicId, int partitionId, ImageWriterO setLastKnownELR(Replicas.toList(lastKnownElr)); } if (options.metadataVersion().isDirectoryAssignmentSupported()) { - record.setDirectories(DirectoryId.toList(directories)); + record.setDirectories(Uuid.toList(directories)); } else { for (int i = 0; i < directories.length; i++) { if (!DirectoryId.UNASSIGNED.equals(directories[i])) { diff --git a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java index e1de7e7beee6b..817d96f1c1496 100644 --- a/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java +++ b/metadata/src/test/java/org/apache/kafka/controller/PartitionChangeBuilderTest.java @@ -100,9 +100,9 @@ public void testChangeRecordIsNoOp() { private static final PartitionRegistration FOO = new PartitionRegistration.Builder(). setReplicas(new int[] {2, 1, 3}). - setDirectories(new DirectoryId[]{ - DirectoryId.fromString("dpdvA5AZSWySmnPFTnu5Kw"), - DirectoryId.fromString("V60B3cglScq3Xk8BX1NxAQ"), + setDirectories(new Uuid[]{ + Uuid.fromString("dpdvA5AZSWySmnPFTnu5Kw"), + Uuid.fromString("V60B3cglScq3Xk8BX1NxAQ"), DirectoryId.UNASSIGNED, }). setIsr(new int[] {2, 1, 3}). @@ -136,11 +136,11 @@ private static PartitionChangeBuilder createFooBuilder(short version) { private static final PartitionRegistration BAR = new PartitionRegistration.Builder(). setReplicas(new int[] {1, 2, 3, 4}). - setDirectories(new DirectoryId[] { + setDirectories(new Uuid[] { DirectoryId.UNASSIGNED, - DirectoryId.fromString("X5FnAcIgTheWgTMzeO5WHw"), - DirectoryId.fromString("GtrcdoSOTm2vFMGFeZq0eg"), - DirectoryId.fromString("YcOqPw5ARmeKr1y9W3AkFw"), + Uuid.fromString("X5FnAcIgTheWgTMzeO5WHw"), + Uuid.fromString("GtrcdoSOTm2vFMGFeZq0eg"), + Uuid.fromString("YcOqPw5ARmeKr1y9W3AkFw"), }). setIsr(new int[] {1, 2, 3}). setRemovingReplicas(new int[] {1}). @@ -167,10 +167,10 @@ private static PartitionChangeBuilder createBarBuilder(short version) { private static final PartitionRegistration BAZ = new PartitionRegistration.Builder(). setReplicas(new int[] {2, 1, 3}). - setDirectories(new DirectoryId[] { - DirectoryId.fromString("ywnfFpTBTbOsFdZ6uAdOmw"), - DirectoryId.fromString("Th0x70ecRbWvZNNV33jyRA"), - DirectoryId.fromString("j216tuSoQsC9JFd1Z5ZP6w"), + setDirectories(new Uuid[] { + Uuid.fromString("ywnfFpTBTbOsFdZ6uAdOmw"), + Uuid.fromString("Th0x70ecRbWvZNNV33jyRA"), + Uuid.fromString("j216tuSoQsC9JFd1Z5ZP6w"), }). setIsr(new int[] {1, 3}). setLeader(3). @@ -187,10 +187,10 @@ private static PartitionChangeBuilder createBazBuilder(short version) { private static final PartitionRegistration OFFLINE = new PartitionRegistration.Builder(). setReplicas(new int[] {2, 1, 3}). - setDirectories(new DirectoryId[]{ - DirectoryId.fromString("iYGgiDV5Sb2EtH6hbgYnCA"), - DirectoryId.fromString("XI2t4qAUSkGlLZSKeEVf8g"), - DirectoryId.fromString("eqRW24kIRlitzQFzmovE0Q") + setDirectories(new Uuid[]{ + Uuid.fromString("iYGgiDV5Sb2EtH6hbgYnCA"), + Uuid.fromString("XI2t4qAUSkGlLZSKeEVf8g"), + Uuid.fromString("eqRW24kIRlitzQFzmovE0Q") }). setIsr(new int[] {3}). setLeader(-1). @@ -390,7 +390,7 @@ public void testReassignmentRearrangesReplicas(short version) { setPartitionId(0). setReplicas(Arrays.asList(3, 2, 1)); if (version > 1) { - Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); + Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); expectedRecord.setDirectories(Arrays.asList(dirs.get(3), dirs.get(2), dirs.get(1))); } assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), @@ -409,7 +409,7 @@ public void testIsrEnlargementCompletesReassignment(short version) { setRemovingReplicas(Collections.emptyList()). setAddingReplicas(Collections.emptyList()); if (version > 1) { - Map dirs = DirectoryId.createAssignmentMap(BAR.replicas, BAR.directories); + Map dirs = DirectoryId.createAssignmentMap(BAR.replicas, BAR.directories); expectedRecord.setDirectories(Arrays.asList(dirs.get(2), dirs.get(3), dirs.get(4))); } assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), @@ -430,7 +430,7 @@ public void testRevertReassignment(short version) { setRemovingReplicas(Collections.emptyList()). setAddingReplicas(Collections.emptyList()); if (version > 1) { - Map dirs = DirectoryId.createAssignmentMap(BAR.replicas, BAR.directories); + Map dirs = DirectoryId.createAssignmentMap(BAR.replicas, BAR.directories); expectedRecord.setDirectories(Arrays.asList(dirs.get(1), dirs.get(2), dirs.get(3))); } assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), @@ -457,7 +457,7 @@ public void testRemovingReplicaReassignment(short version) { setIsr(Arrays.asList(2, 1)). setLeader(1); if (version > 1) { - Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); + Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); expectedRecord.setDirectories(Arrays.asList(dirs.get(1), dirs.get(2))); } assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), @@ -481,7 +481,7 @@ public void testAddingReplicaReassignment(short version) { setReplicas(Arrays.asList(1, 2, 3, 4)). setAddingReplicas(Collections.singletonList(4)); if (version > 1) { - Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); + Map dirs = DirectoryId.createAssignmentMap(FOO.replicas, FOO.directories); expectedRecord.setDirectories(Arrays.asList(dirs.get(1), dirs.get(2), dirs.get(3), DirectoryId.UNASSIGNED)); } assertEquals(Optional.of(new ApiMessageAndVersion(expectedRecord, version)), @@ -547,10 +547,10 @@ public void testChangeInLeadershipDoesNotChangeRecoveryState(boolean isLeaderRec LeaderRecoveryState recoveryState = LeaderRecoveryState.RECOVERING; PartitionRegistration registration = new PartitionRegistration.Builder(). setReplicas(new int[] {leaderId, leaderId + 1, leaderId + 2}). - setDirectories(new DirectoryId[] { - DirectoryId.fromString("1sF6XXLkSN2LtDums7CJ8Q"), - DirectoryId.fromString("iaBBVsoHQR6NDKXwliKMqw"), - DirectoryId.fromString("sHaBwjdrR2S3bL4E1RKC8Q") + setDirectories(new Uuid[] { + Uuid.fromString("1sF6XXLkSN2LtDums7CJ8Q"), + Uuid.fromString("iaBBVsoHQR6NDKXwliKMqw"), + Uuid.fromString("sHaBwjdrR2S3bL4E1RKC8Q") }). setIsr(new int[] {leaderId}). setLeader(leaderId). @@ -617,10 +617,10 @@ void testUncleanSetsLeaderRecoveringState(boolean isLeaderRecoverySupported, boo int leaderId = 1; PartitionRegistration registration = new PartitionRegistration.Builder(). setReplicas(new int[] {leaderId, leaderId + 1, leaderId + 2}). - setDirectories(new DirectoryId[] { - DirectoryId.fromString("uYpxts0pS4K4bk5XOoXB4g"), - DirectoryId.fromString("kS6fHEqwRYucduWkmvsevw"), - DirectoryId.fromString("De9RqRThQRGjKg3i3yzUxA") + setDirectories(new Uuid[] { + Uuid.fromString("uYpxts0pS4K4bk5XOoXB4g"), + Uuid.fromString("kS6fHEqwRYucduWkmvsevw"), + Uuid.fromString("De9RqRThQRGjKg3i3yzUxA") }). setIsr(new int[] {leaderId + 1, leaderId + 2}). setLeader(NO_LEADER). @@ -673,11 +673,11 @@ void testUncleanSetsLeaderRecoveringState(boolean isLeaderRecoverySupported, boo public void testStoppedLeaderIsDemotedAfterReassignmentCompletesEvenIfNoNewEligibleLeaders() { // Set up PartitionRegistration as if there's an ongoing reassignment from [0, 1] to [2, 3] int[] replicas = new int[] {2, 3, 0, 1}; - DirectoryId[] directories = { - DirectoryId.fromString("XCBQClkBSZyphD87QUXzDA"), - DirectoryId.fromString("Or2Rp9tTQOSVuy12hsfmTA"), - DirectoryId.fromString("pThsodMNSwGvljTfc1RNVQ"), - DirectoryId.fromString("d8CGoNJmS5mJdF20tc8P7g") + Uuid[] directories = { + Uuid.fromString("XCBQClkBSZyphD87QUXzDA"), + Uuid.fromString("Or2Rp9tTQOSVuy12hsfmTA"), + Uuid.fromString("pThsodMNSwGvljTfc1RNVQ"), + Uuid.fromString("d8CGoNJmS5mJdF20tc8P7g") }; // The ISR starts off with the old replicas int[] isr = new int[] {0, 1}; @@ -747,11 +747,11 @@ private MetadataVersion leaderRecoveryMetadataVersion(boolean isSupported) { public void testEligibleLeaderReplicas_IsrShrinkBelowMinISR(short version) { PartitionRegistration partition = new PartitionRegistration.Builder() .setReplicas(new int[] {1, 2, 3, 4}) - .setDirectories(new DirectoryId[] { - DirectoryId.fromString("NeQeLdHhSXi4tQGaFcszKA"), - DirectoryId.fromString("LsVrQZ73RSSuEWA8hhqQhg"), - DirectoryId.fromString("0IaY4zXKRR6sROgE8yHfnw"), - DirectoryId.fromString("1WxphfLCSZqMHKK4JMppuw") + .setDirectories(new Uuid[] { + Uuid.fromString("NeQeLdHhSXi4tQGaFcszKA"), + Uuid.fromString("LsVrQZ73RSSuEWA8hhqQhg"), + Uuid.fromString("0IaY4zXKRR6sROgE8yHfnw"), + Uuid.fromString("1WxphfLCSZqMHKK4JMppuw") }) .setIsr(new int[] {1, 2, 3, 4}) .setLeader(1) @@ -796,11 +796,11 @@ public void testEligibleLeaderReplicas_IsrShrinkBelowMinISR(short version) { public void testEligibleLeaderReplicas_IsrExpandAboveMinISR(short version) { PartitionRegistration partition = new PartitionRegistration.Builder() .setReplicas(new int[] {1, 2, 3, 4}) - .setDirectories(new DirectoryId[]{ - DirectoryId.fromString("CWgRKBKkToGn1HKzNb2qqQ"), - DirectoryId.fromString("SCnk7zfSQSmlKqvV702d3A"), - DirectoryId.fromString("9tO0QHlJRhimjKfH8m9d8A"), - DirectoryId.fromString("JaaqVOxNT2OGVNCCIFA2JQ") + .setDirectories(new Uuid[]{ + Uuid.fromString("CWgRKBKkToGn1HKzNb2qqQ"), + Uuid.fromString("SCnk7zfSQSmlKqvV702d3A"), + Uuid.fromString("9tO0QHlJRhimjKfH8m9d8A"), + Uuid.fromString("JaaqVOxNT2OGVNCCIFA2JQ") }) .setIsr(new int[] {1, 2}) .setElr(new int[] {3}) @@ -842,11 +842,11 @@ public void testEligibleLeaderReplicas_IsrExpandAboveMinISR(short version) { public void testEligibleLeaderReplicas_IsrAddNewMemberNotInELR(short version) { PartitionRegistration partition = new PartitionRegistration.Builder() .setReplicas(new int[] {1, 2, 3, 4}) - .setDirectories(new DirectoryId[]{ - DirectoryId.fromString("gPcIwlldQXikdUB3F4GB6w"), - DirectoryId.fromString("gFs7V8mKR66z8e5qwtjIMA"), - DirectoryId.fromString("zKHU2fwrRkuypqTgITl46g"), - DirectoryId.fromString("zEgmBBh8QJGqbBIvzvH7JA") + .setDirectories(new Uuid[]{ + Uuid.fromString("gPcIwlldQXikdUB3F4GB6w"), + Uuid.fromString("gFs7V8mKR66z8e5qwtjIMA"), + Uuid.fromString("zKHU2fwrRkuypqTgITl46g"), + Uuid.fromString("zEgmBBh8QJGqbBIvzvH7JA") }) .setIsr(new int[] {1}) .setElr(new int[] {3}) @@ -894,11 +894,11 @@ public void testEligibleLeaderReplicas_IsrAddNewMemberNotInELR(short version) { public void testEligibleLeaderReplicas_RemoveUncleanShutdownReplicasFromElr(short version) { PartitionRegistration partition = new PartitionRegistration.Builder() .setReplicas(new int[] {1, 2, 3, 4}) - .setDirectories(new DirectoryId[] { - DirectoryId.fromString("keB9ssIPRlibyVJT5FcBVA"), - DirectoryId.fromString("FhezfoReTSmHoKxi8wOIOg"), - DirectoryId.fromString("QHtFxu8LShm6RiyAP6PxYg"), - DirectoryId.fromString("tUJOMtvMQkGga30ydluvbQ") + .setDirectories(new Uuid[] { + Uuid.fromString("keB9ssIPRlibyVJT5FcBVA"), + Uuid.fromString("FhezfoReTSmHoKxi8wOIOg"), + Uuid.fromString("QHtFxu8LShm6RiyAP6PxYg"), + Uuid.fromString("tUJOMtvMQkGga30ydluvbQ") }) .setIsr(new int[] {1}) .setElr(new int[] {2, 3}) @@ -946,10 +946,10 @@ public void testEligibleLeaderReplicas_RemoveUncleanShutdownReplicasFromElr(shor void testKeepsDirectoriesAfterReassignment() { PartitionRegistration registration = new PartitionRegistration.Builder(). setReplicas(new int[] {2, 1, 3}). - setDirectories(new DirectoryId[] { - DirectoryId.fromString("v1PVrX6uS5m8CByXlLfmWg"), - DirectoryId.fromString("iU2znv45Q9yQkOpkTSy3jA"), - DirectoryId.fromString("fM5NKyWTQHqEihjIkUl99Q") + setDirectories(new Uuid[] { + Uuid.fromString("v1PVrX6uS5m8CByXlLfmWg"), + Uuid.fromString("iU2znv45Q9yQkOpkTSy3jA"), + Uuid.fromString("fM5NKyWTQHqEihjIkUl99Q") }). setIsr(new int[] {2, 1, 3}). setLeader(1). @@ -967,8 +967,8 @@ void testKeepsDirectoriesAfterReassignment() { setLeader(1). setReplicas(Arrays.asList(3, 1, 4)). setDirectories(Arrays.asList( - DirectoryId.fromString("fM5NKyWTQHqEihjIkUl99Q"), - DirectoryId.fromString("iU2znv45Q9yQkOpkTSy3jA"), + Uuid.fromString("fM5NKyWTQHqEihjIkUl99Q"), + Uuid.fromString("iU2znv45Q9yQkOpkTSy3jA"), DirectoryId.UNASSIGNED )), (short) 2 diff --git a/metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java b/metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java index e802a2e82471e..d781280e1518f 100644 --- a/metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java +++ b/metadata/src/test/java/org/apache/kafka/image/ImageDowngradeTest.java @@ -17,7 +17,6 @@ package org.apache.kafka.image; -import org.apache.kafka.common.DirectoryId; import org.apache.kafka.common.Uuid; import org.apache.kafka.common.metadata.FeatureLevelRecord; import org.apache.kafka.common.metadata.PartitionRecord; @@ -144,8 +143,8 @@ void testDirectoryAssignmentState() { TEST_RECORDS.get(0), new ApiMessageAndVersion( testPartitionRecord.duplicate().setDirectories(Arrays.asList( - DirectoryId.fromString("c7QfSi6xSIGQVh3Qd5RJxA"), - DirectoryId.fromString("rWaCHejCRRiptDMvW5Xw0g"))), + Uuid.fromString("c7QfSi6xSIGQVh3Qd5RJxA"), + Uuid.fromString("rWaCHejCRRiptDMvW5Xw0g"))), (short) 2)), Arrays.asList( metadataVersionRecord(outputMetadataVersion), diff --git a/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java b/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java index 36f2e4ac9f752..757cb76f9eb32 100644 --- a/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java +++ b/metadata/src/test/java/org/apache/kafka/metadata/PartitionRegistrationTest.java @@ -133,7 +133,7 @@ public void testToLeaderAndIsrPartitionState() { @Test public void testMergePartitionChangeRecordWithReassignmentData() { PartitionRegistration partition0 = new PartitionRegistration.Builder().setReplicas(new int[] {1, 2, 3}). - setDirectories(new DirectoryId[]{DirectoryId.fromString("FbRuu7CeQtq5YFreEzg16g"), DirectoryId.fromString("4rtHTelWSSStAFMODOg3cQ"), DirectoryId.fromString("Id1WXzHURROilVxZWJNZlw")}). + setDirectories(new Uuid[]{Uuid.fromString("FbRuu7CeQtq5YFreEzg16g"), Uuid.fromString("4rtHTelWSSStAFMODOg3cQ"), Uuid.fromString("Id1WXzHURROilVxZWJNZlw")}). setIsr(new int[] {1, 2, 3}).setLeader(1).setLeaderRecoveryState(LeaderRecoveryState.RECOVERED).setLeaderEpoch(100).setPartitionEpoch(200).build(); PartitionRegistration partition1 = partition0.merge(new PartitionChangeRecord(). setRemovingReplicas(Collections.singletonList(3)). @@ -261,12 +261,12 @@ public void testBuilderSetsDefaultAddingAndRemovingReplicas() { public void testPartitionRegistrationToRecord(short version) { PartitionRegistration.Builder builder = new PartitionRegistration.Builder(). setReplicas(new int[]{0, 1, 2, 3, 4}). - setDirectories(new DirectoryId[]{ + setDirectories(new Uuid[]{ DirectoryId.UNASSIGNED, - DirectoryId.fromString("KBJBm9GVRAG9Ffe25odmmg"), + Uuid.fromString("KBJBm9GVRAG9Ffe25odmmg"), DirectoryId.LOST, - DirectoryId.fromString("7DZNT5qBS7yFF7VMMHS7kw"), - DirectoryId.fromString("cJGPUZsMSEqbidOLYLOIXg") + Uuid.fromString("7DZNT5qBS7yFF7VMMHS7kw"), + Uuid.fromString("cJGPUZsMSEqbidOLYLOIXg") }). setIsr(new int[]{0, 1}). setLeader(0). @@ -298,10 +298,10 @@ public void testPartitionRegistrationToRecord(short version) { if (version > 1) { expectRecord.setDirectories(Arrays.asList( DirectoryId.UNASSIGNED, - DirectoryId.fromString("KBJBm9GVRAG9Ffe25odmmg"), + Uuid.fromString("KBJBm9GVRAG9Ffe25odmmg"), DirectoryId.LOST, - DirectoryId.fromString("7DZNT5qBS7yFF7VMMHS7kw"), - DirectoryId.fromString("cJGPUZsMSEqbidOLYLOIXg") + Uuid.fromString("7DZNT5qBS7yFF7VMMHS7kw"), + Uuid.fromString("cJGPUZsMSEqbidOLYLOIXg") )); when(metadataVersion.isDirectoryAssignmentSupported()).thenReturn(true); } diff --git a/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java b/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java index 6c22d70f283ee..82cb64f4df461 100644 --- a/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java +++ b/server-common/src/main/java/org/apache/kafka/common/DirectoryId.java @@ -25,102 +25,52 @@ import java.util.Map; import java.util.Set; -public class DirectoryId extends Uuid { +public class DirectoryId { /** - * A DirectoryId that is used to identify new or unknown dir assignments. + * A Uuid that is used to identify new or unknown dir assignments. */ - public static final DirectoryId UNASSIGNED = new DirectoryId(0L, 0L); + public static final Uuid UNASSIGNED = new Uuid(0L, 0L); /** - * A DirectoryId that is used to represent unspecified offline dirs. + * A Uuid that is used to represent unspecified offline dirs. */ - public static final DirectoryId LOST = new DirectoryId(0L, 1L); + public static final Uuid LOST = new Uuid(0L, 1L); /** - * A DirectoryId that is used to represent and unspecified log directory, + * A Uuid that is used to represent and unspecified log directory, * that is expected to have been previously selected to host an * associated replica. This contrasts with {@code UNASSIGNED_DIR}, * which is associated with (typically new) replicas that may not * yet have been placed in any log directory. */ - public static final DirectoryId MIGRATING = new DirectoryId(0L, 2L); + public static final Uuid MIGRATING = new Uuid(0L, 2L); /** * The set of reserved UUIDs that will never be returned by the random method. */ - public static final Set RESERVED; + public static final Set RESERVED; static { - HashSet reserved = new HashSet<>(); - // The first 100 DirectoryIds are reserved for future use. + HashSet reserved = new HashSet<>(); + // The first 100 Uuids are reserved for future use. for (long i = 0L; i < 100L; i++) { - reserved.add(new DirectoryId(0L, i)); + reserved.add(new Uuid(0L, i)); } RESERVED = Collections.unmodifiableSet(reserved); } - /** - * Constructs a Directory ID from the underlying 128 bits, - * exactly as a {@link Uuid} is constructed. - */ - private DirectoryId(long mostSigBits, long leastSigBits) { - super(mostSigBits, leastSigBits); - } - - /** - * Creates a DirectoryId based on a base64 string encoding used in the toString() method. - */ - public static DirectoryId fromString(String str) { - return DirectoryId.fromUuid(Uuid.fromString(str)); - } - - /** - * Creates a DirectoryId based on a {@link Uuid}. - */ - public static DirectoryId fromUuid(Uuid uuid) { - return new DirectoryId(uuid.getMostSignificantBits(), uuid.getLeastSignificantBits()); - } - /** * Static factory to generate a directory ID. * * This will not generate a reserved UUID (first 100), or one whose string representation starts with a dash ("-") */ - public static DirectoryId random() { + public static Uuid random() { Uuid uuid = Uuid.randomUuid(); while (RESERVED.contains(uuid) || uuid.toString().startsWith("-")) { uuid = Uuid.randomUuid(); } - return DirectoryId.fromUuid(uuid); - } - - /** - * Convert a list of Uuid to an array of DirectoryId. - * - * @param list The input list - * @return The output array - */ - public static DirectoryId[] toArray(List list) { - if (list == null) return null; - DirectoryId[] array = new DirectoryId[list.size()]; - for (int i = 0; i < list.size(); i++) { - array[i] = DirectoryId.fromUuid(list.get(i)); - } - return array; - } - - /** - * Convert an array of DirectoryIds to a list of Uuid. - * - * @param array The input array - * @return The output list - */ - public static List toList(DirectoryId[] array) { - if (array == null) return null; - List list = new ArrayList<>(array.length); - list.addAll(Arrays.asList(array)); - return list; + return uuid; } /** @@ -134,10 +84,10 @@ public static List toList(DirectoryId[] array) { * @throws IllegalArgumentException If currentReplicas and currentDirectories have different lengths, * or if there are duplicate broker IDs in the replica lists */ - public static List createDirectoriesFrom(int[] currentReplicas, DirectoryId[] currentDirectories, List newReplicas) { + public static List createDirectoriesFrom(int[] currentReplicas, Uuid[] currentDirectories, List newReplicas) { if (currentReplicas == null) currentReplicas = new int[0]; - if (currentDirectories == null) currentDirectories = new DirectoryId[0]; - Map assignments = createAssignmentMap(currentReplicas, currentDirectories); + if (currentDirectories == null) currentDirectories = new Uuid[0]; + Map assignments = createAssignmentMap(currentReplicas, currentDirectories); List consolidated = new ArrayList<>(newReplicas.size()); for (int newReplica : newReplicas) { Uuid newDirectory = assignments.getOrDefault(newReplica, UNASSIGNED); @@ -154,14 +104,14 @@ public static List createDirectoriesFrom(int[] currentReplicas, DirectoryI * @throws IllegalArgumentException If replicas and directories have different lengths, * or if there are duplicate broker IDs in the replica list */ - public static Map createAssignmentMap(int[] replicas, DirectoryId[] directories) { + public static Map createAssignmentMap(int[] replicas, Uuid[] directories) { if (replicas.length != directories.length) { throw new IllegalArgumentException("The lengths for replicas and directories do not match."); } - Map assignments = new HashMap<>(); + Map assignments = new HashMap<>(); for (int i = 0; i < replicas.length; i++) { int brokerId = replicas[i]; - DirectoryId directory = directories[i]; + Uuid directory = directories[i]; if (assignments.put(brokerId, directory) != null) { throw new IllegalArgumentException("Duplicate broker ID in assignment"); } @@ -172,8 +122,8 @@ public static Map createAssignmentMap(int[] replicas, Dire /** * Create an array with the specified number of entries set to {@link #UNASSIGNED}. */ - public static DirectoryId[] unassignedArray(int length) { - DirectoryId[] array = new DirectoryId[length]; + public static Uuid[] unassignedArray(int length) { + Uuid[] array = new Uuid[length]; Arrays.fill(array, UNASSIGNED); return array; } diff --git a/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java b/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java index 8fe7ec2ea19d4..5db21931077d5 100644 --- a/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java +++ b/server-common/src/test/java/org/apache/kafka/common/DirectoryIdTest.java @@ -23,9 +23,7 @@ import java.util.HashSet; import java.util.Set; -import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -34,7 +32,7 @@ public class DirectoryIdTest { @Test void testReserved() { Set seen = new HashSet<>(100); - for (DirectoryId reservedId : DirectoryId.RESERVED) { + for (Uuid reservedId : DirectoryId.RESERVED) { assertEquals(0L, reservedId.getMostSignificantBits(), "Unexpected reserved msb value"); long lsb = reservedId.getLeastSignificantBits(); assertTrue(lsb >= 0 && lsb < 100L, "Unexpected reserved lsb value"); @@ -43,51 +41,25 @@ void testReserved() { assertEquals(100, DirectoryId.RESERVED.size()); } - @Test - void testToArray() { - assertNull(DirectoryId.toArray(null)); - assertArrayEquals( - new DirectoryId[]{ - DirectoryId.MIGRATING, DirectoryId.fromString("UXyU9i5ARn6W00ON2taeWA") - }, - DirectoryId.toArray(Arrays.asList( - DirectoryId.MIGRATING, DirectoryId.fromString("UXyU9i5ARn6W00ON2taeWA") - )) - ); - } - - @Test - void testToList() { - assertNull(DirectoryId.toList(null)); - assertEquals( - Arrays.asList( - DirectoryId.MIGRATING, DirectoryId.fromString("UXyU9i5ARn6W00ON2taeWA") - ), - DirectoryId.toList(new DirectoryId[]{ - DirectoryId.MIGRATING, DirectoryId.fromString("UXyU9i5ARn6W00ON2taeWA") - }) - ); - } - @Test void testCreateDirectoriesFrom() { assertThrows(IllegalArgumentException.class, () -> DirectoryId.createDirectoriesFrom( new int[] {1}, - new DirectoryId[] {DirectoryId.UNASSIGNED, DirectoryId.LOST}, + new Uuid[] {DirectoryId.UNASSIGNED, DirectoryId.LOST}, Arrays.asList(2, 3) )); assertEquals( Arrays.asList( - DirectoryId.fromString("YXY0bQYEQmmyOQ6ZDfGgSQ"), - DirectoryId.fromString("5SZij3DRQgaFbvzR9KooLg"), + Uuid.fromString("YXY0bQYEQmmyOQ6ZDfGgSQ"), + Uuid.fromString("5SZij3DRQgaFbvzR9KooLg"), DirectoryId.UNASSIGNED ), DirectoryId.createDirectoriesFrom( new int[] {1, 2, 3}, - new DirectoryId[] { - DirectoryId.fromString("MgVK5KSwTxe65eYATaoQrg"), - DirectoryId.fromString("YXY0bQYEQmmyOQ6ZDfGgSQ"), - DirectoryId.fromString("5SZij3DRQgaFbvzR9KooLg") + new Uuid[] { + Uuid.fromString("MgVK5KSwTxe65eYATaoQrg"), + Uuid.fromString("YXY0bQYEQmmyOQ6ZDfGgSQ"), + Uuid.fromString("5SZij3DRQgaFbvzR9KooLg") }, Arrays.asList(2, 3, 4) ) @@ -100,7 +72,7 @@ void testCreateDirectoriesFrom() { ), DirectoryId.createDirectoriesFrom( new int[] {1, 2}, - new DirectoryId[] { + new Uuid[] { DirectoryId.UNASSIGNED, DirectoryId.UNASSIGNED }, @@ -114,19 +86,19 @@ void testCreateAssignmentMap() { assertThrows(IllegalArgumentException.class, () -> DirectoryId.createAssignmentMap(new int[]{1, 2}, DirectoryId.unassignedArray(3))); assertEquals( - new HashMap() {{ - put(1, DirectoryId.fromString("upjfkCrUR9GNn1i94ip1wg")); - put(2, DirectoryId.fromString("bCF3l0RIQjOKhUqgbivHZA")); - put(3, DirectoryId.fromString("Fg3mFhcVQlqCWRk4dZazxw")); - put(4, DirectoryId.fromString("bv9TEYi4TqOm52hLmrxT5w")); + new HashMap() {{ + put(1, Uuid.fromString("upjfkCrUR9GNn1i94ip1wg")); + put(2, Uuid.fromString("bCF3l0RIQjOKhUqgbivHZA")); + put(3, Uuid.fromString("Fg3mFhcVQlqCWRk4dZazxw")); + put(4, Uuid.fromString("bv9TEYi4TqOm52hLmrxT5w")); }}, DirectoryId.createAssignmentMap( new int[] {1, 2, 3, 4}, - new DirectoryId[] { - DirectoryId.fromString("upjfkCrUR9GNn1i94ip1wg"), - DirectoryId.fromString("bCF3l0RIQjOKhUqgbivHZA"), - DirectoryId.fromString("Fg3mFhcVQlqCWRk4dZazxw"), - DirectoryId.fromString("bv9TEYi4TqOm52hLmrxT5w") + new Uuid[] { + Uuid.fromString("upjfkCrUR9GNn1i94ip1wg"), + Uuid.fromString("bCF3l0RIQjOKhUqgbivHZA"), + Uuid.fromString("Fg3mFhcVQlqCWRk4dZazxw"), + Uuid.fromString("bv9TEYi4TqOm52hLmrxT5w") }) ); } From 27e815f40405eec6800e4adcaa67c33affd1b425 Mon Sep 17 00:00:00 2001 From: Igor Soarez Date: Wed, 1 Nov 2023 16:46:35 +0000 Subject: [PATCH 6/6] Make constructors in InvalidMetadataException protected --- .../kafka/common/errors/InvalidMetadataException.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java b/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java index e6663dbf185d0..504e8f3cc782b 100644 --- a/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java +++ b/clients/src/main/java/org/apache/kafka/common/errors/InvalidMetadataException.java @@ -23,19 +23,19 @@ public abstract class InvalidMetadataException extends RetriableException { private static final long serialVersionUID = 1L; - public InvalidMetadataException() { + protected InvalidMetadataException() { super(); } - public InvalidMetadataException(String message) { + protected InvalidMetadataException(String message) { super(message); } - public InvalidMetadataException(String message, Throwable cause) { + protected InvalidMetadataException(String message, Throwable cause) { super(message, cause); } - public InvalidMetadataException(Throwable cause) { + protected InvalidMetadataException(Throwable cause) { super(cause); }