You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2022/08/14 20:05:54 UTC
[GitHub] [kafka] cmccabe opened a new pull request, #12513: MINOR: some fixes to the KRaft bootstrap logic
cmccabe opened a new pull request, #12513:
URL: https://github.com/apache/kafka/pull/12513
The main changes here are ensuring that we always have a metadata.version record in the log, making
sure that the bootstrap file can be used for records other than the metadata.version record (for
example, we will want to put SCRAM initialization records there), and fixing some bugs.
In order to ensure that we always have a metadata.version record in the log, we have to make the
minimum KRaft version 3.3IV0, since this is the first version that supports that. Older KRaft
clusters that were on a pre-production version will be auto-upgraded to 3.3IV0.
Fix some issues related to initializing new clusters. If there are no records in the log at all,
then insert the bootstrap records in a single batch. If there are records, but no metadata version,
process the existing records as though they were metadata.version 3.3IV0 and then append a metadata
version record setting version 3.3IV0. Previously, we were not clearly distinguishing between the
case where the metadata log was empty, and the case where we just needed to add a metadata.version
record.
Refactor BootstrapMetadata into an immutable class which contains a 3-tuple of metadata version,
record list, and source. The source field is used to log where the bootstrap metadata was obtained
from. This could be a bootstrap file, the static configuration, or just the software defaults.
Move the logic for reading and writing bootstrap files into BootstrapDirectory.java.
Add LogReplayTracker, which tracks whether the log is empty, whether we've seen a metadata version
record, and the number of bytes we've replayed since the last snapshot.
Fix a bug in FeatureControlManager where it was possible to use a "downgrade" operation to
transition to a newer version. Do not store whether we have seen a metadata version or not in
FeatureControlManager, since that is now handled by LogReplayTracker.
Introduce BatchFileReader, which is a simple way of reading a file containing batches of snapshots
that does not require spawning a thread. Rename SnapshotFileWriter to BatchFileWriter to be
consistent, and to reflect the fact that bootstrap files aren't snapshots.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949665053
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
Review Comment:
I don't feel that strongly about it. But generally making a parameter non-nullable feels more natural than making it non-emptyable, and you want to choose one or the other
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: KAFKA-14177: Correctly support older kraft versions without FeatureLevelRecord
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r954317998
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -316,10 +326,12 @@ public boolean hasNext() {
public List<ApiMessageAndVersion> next() {
// Write the metadata.version first
if (!wroteVersion) {
- wroteVersion = true;
- return Collections.singletonList(new ApiMessageAndVersion(new FeatureLevelRecord()
- .setName(MetadataVersion.FEATURE_NAME)
- .setFeatureLevel(metadataVersion.featureLevel()), FEATURE_LEVEL_RECORD.lowestSupportedVersion()));
+ if (!metadataVersion.isLessThan(minimumBootstrapVersion)) {
Review Comment:
nit: i think this check reads easier as `metadataVersion.isAtLeast(minimumBootstrapVersion)`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
mumrah commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r950205140
##########
metadata/src/main/java/org/apache/kafka/controller/LogReplayTracker.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.controller;
+
+import java.util.Optional;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.metadata.bootstrap.BootstrapMetadata;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.slf4j.Logger;
+
+
+public class LogReplayTracker {
Review Comment:
Brief javadoc would be nice. Should probably mention this is not thread safe
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1118,6 +1059,65 @@ private void updateWriteOffset(long offset) {
}
}
+ private void claim(int epoch) {
+ try {
+ if (curClaimEpoch != -1) {
+ throw new RuntimeException("Cannot claim leadership because we are already the " +
+ "active controller.");
+ }
+ curClaimEpoch = epoch;
+ controllerMetrics.setActive(true);
+ updateWriteOffset(lastCommittedOffset);
+ clusterControl.activate();
+
+ // Before switching to active, create an in-memory snapshot at the last committed
+ // offset. This is required because the active controller assumes that there is always
+ // an in-memory snapshot at the last committed offset.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+
+ // Prepend the activate event. It is important that this event go at the beginning
+ // of the queue rather than the end (hence prepend rather than append). It's also
+ // important not to use prepend for anything else, to preserve the ordering here.
+ queue.prepend(new ControllerWriteEvent<>("completeActivation[" + epoch + "]",
+ new CompleteActivationEvent()));
+ } catch (Throwable e) {
+ fatalFaultHandler.handleFault("exception while claiming leadership", e);
+ }
+ }
+
+ class CompleteActivationEvent implements ControllerWriteOperation<Void> {
Review Comment:
"Activation" is sort of unclear here. How about CompleteLeadershipClaim or something to indicate this happens as a result of becoming leader.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949693026
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -222,52 +227,130 @@ public void testFeatureControlIterator() throws Exception {
manager.iterator(Long.MAX_VALUE));
}
+ private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
+ new FeatureControlManager.Builder().
+ setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
+ MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+
@Test
public void testApplyMetadataVersionChangeRecord() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).build();
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_0_IV1.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_0_IV1, manager.metadataVersion());
+ setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
+ assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
}
@Test
- public void testDowngradeMetadataVersion() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_2_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
- build();
- assertEquals(manager.metadataVersion(), MetadataVersion.IBP_3_3_IV0);
+ public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ assertEquals(ControllerResult.of(Collections.emptyList(),
+ singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
Review Comment:
Hmm... I have a different view about INVALID_REQUEST. I thought that error code was supposed to be a catch-all for requests that were malformed. I don't think we should be using it in APIs to indicate that the user messed up. In theory a well-programmed client should not send invalid requests.
I'm used to the way errno works in C, where there's a small number of error codes and they're heavily overloaded. For example, a bunch of APIs return ENOENT if the entity they're looking for doesn't exist, or EEXIST if it does and that's a problem.
I sort of look askance at Kafka's approach. Like did we really need a separate error code for DELEGATION_TOKEN_NOT_FOUND, GROUP_ID_NOT_FOUND, and RESOURCE_NOT_FOUND? Or could we just have a NOT_FOUND error code and return it where appropriate. It's not like you are going to get back DELEGATION_TOKEN_NOT_FOUND from an API that deals with groups, so what's the issue? You are not adding more descriptive power by having an X_NOT_FOUND, Y_NOT_FOUND, Z_NOT_FOUND If only one of them can be used in any given scenario.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r950354444
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
+ * if it is not configured.
+ */
+ public BootstrapDirectory(
+ String directoryPath,
+ String ibp
+ ) {
+ Objects.requireNonNull(directoryPath);
+ Objects.requireNonNull(ibp);
+ this.directoryPath = directoryPath;
+ this.ibp = ibp;
+ }
+
+ public BootstrapMetadata read() throws Exception {
+ if (!Files.isDirectory(Paths.get(directoryPath))) {
+ throw new RuntimeException("No such directory as " + directoryPath);
+ }
+ Path binaryBootstrapPath = Paths.get(directoryPath, BINARY_BOOTSTRAP);
+ if (!Files.exists(binaryBootstrapPath)) {
+ return readFromConfiguration();
+ } else {
+ return readFromBinaryFile(binaryBootstrapPath.toString());
+ }
+ }
+
+ BootstrapMetadata readFromConfiguration() {
+ if (ibp.isEmpty()) {
+ return BootstrapMetadata.fromVersion(MetadataVersion.latest(),
+ "the default bootstrap file which sets the latest metadata.version, " +
+ "since no bootstrap file was found, and " + INTER_BROKER_PROTOCOL_CONFIG_KEY +
+ " was not configured.");
+ }
+ MetadataVersion version = MetadataVersion.fromVersionString(ibp);
+ if (version.isLessThan(MINIMUM_KRAFT_VERSION)) {
+ return BootstrapMetadata.fromVersion(MINIMUM_KRAFT_VERSION,
+ "a default bootstrap file setting the minimum supported KRaft metadata " +
Review Comment:
My point is that it does not read coherently when inserted into the exception message. For example:
> Unable to load metadata from a default bootstrap file setting the minimum supported KRaft metadata version, since no bootstrap file was found, and inter.broker.protocol.version was XX which is not supported by KRaft.: bootstrap metadata versions less than 3.3.IV1 are not supported.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r950399632
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
+ * if it is not configured.
+ */
+ public BootstrapDirectory(
+ String directoryPath,
+ String ibp
+ ) {
+ Objects.requireNonNull(directoryPath);
+ Objects.requireNonNull(ibp);
+ this.directoryPath = directoryPath;
+ this.ibp = ibp;
+ }
+
+ public BootstrapMetadata read() throws Exception {
+ if (!Files.isDirectory(Paths.get(directoryPath))) {
+ throw new RuntimeException("No such directory as " + directoryPath);
+ }
+ Path binaryBootstrapPath = Paths.get(directoryPath, BINARY_BOOTSTRAP);
+ if (!Files.exists(binaryBootstrapPath)) {
+ return readFromConfiguration();
+ } else {
+ return readFromBinaryFile(binaryBootstrapPath.toString());
+ }
+ }
+
+ BootstrapMetadata readFromConfiguration() {
+ if (ibp.isEmpty()) {
+ return BootstrapMetadata.fromVersion(MetadataVersion.latest(),
+ "the default bootstrap file which sets the latest metadata.version, " +
+ "since no bootstrap file was found, and " + INTER_BROKER_PROTOCOL_CONFIG_KEY +
+ " was not configured.");
+ }
+ MetadataVersion version = MetadataVersion.fromVersionString(ibp);
+ if (version.isLessThan(MINIMUM_KRAFT_VERSION)) {
+ return BootstrapMetadata.fromVersion(MINIMUM_KRAFT_VERSION,
+ "a default bootstrap file setting the minimum supported KRaft metadata " +
Review Comment:
OK. I will reorganize this so that "source" is always at the end of the exception text, and make it a little bit shorter.
I still think we should err on the side of being a bit verbose and explaining exactly what is happening, even if it's a little janky. Having to guess about this stuff is very very bad. Often all you have to go on in these problems is an exception, log message, etc. so we should be sure that it actually does explain what happened.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949668350
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
+ * if it is not configured.
+ */
+ public BootstrapDirectory(
+ String directoryPath,
+ String ibp
+ ) {
+ Objects.requireNonNull(directoryPath);
+ Objects.requireNonNull(ibp);
+ this.directoryPath = directoryPath;
+ this.ibp = ibp;
+ }
+
+ public BootstrapMetadata read() throws Exception {
+ if (!Files.isDirectory(Paths.get(directoryPath))) {
+ throw new RuntimeException("No such directory as " + directoryPath);
Review Comment:
I will add a different message if it exists but not as a directory. I wish we had a better standard library (that had something more like Golang's `os.Stat`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949680410
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1118,6 +1059,65 @@ private void updateWriteOffset(long offset) {
}
}
+ private void claim(int epoch) {
+ try {
+ if (curClaimEpoch != -1) {
+ throw new RuntimeException("Cannot claim leadership because we are already the " +
+ "active controller.");
+ }
+ curClaimEpoch = epoch;
+ controllerMetrics.setActive(true);
+ updateWriteOffset(lastCommittedOffset);
+ clusterControl.activate();
+
+ // Before switching to active, create an in-memory snapshot at the last committed
+ // offset. This is required because the active controller assumes that there is always
+ // an in-memory snapshot at the last committed offset.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+
+ // Prepend the activate event. It is important that this event go at the beginning
+ // of the queue rather than the end (hence prepend rather than append). It's also
+ // important not to use prepend for anything else, to preserve the ordering here.
+ queue.prepend(new ControllerWriteEvent<>("completeActivation[" + epoch + "]",
+ new CompleteActivationEvent()));
+ } catch (Throwable e) {
+ fatalFaultHandler.handleFault("exception while claiming leadership", e);
+ }
+ }
+
+ class CompleteActivationEvent implements ControllerWriteOperation<Void> {
+ @Override
+ public ControllerResult<Void> generateRecordsAndResult() throws Exception {
+ List<ApiMessageAndVersion> records = new ArrayList<>();
+ if (logReplayTracker.empty()) {
+ // If no records have been replayed, we need to write out the bootstrap records.
+ // This will include the new metadata.version, as well as things like SCRAM
+ // initialization, etc.
+ log.warn("The metadata log appears to be empty. Appending {} bootstrap record(s) " +
Review Comment:
Ehh... `warn` seems appropriate. We are upgrading your `metadata.version`, it's kind of a big deal. You will also only ever see this once
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r952030392
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
Review Comment:
I don't feel strongly either. The main argument in favor of `null` is that it prevents the value from being used in a context where it might not apply. In other words, it provides some runtime safety. A better option is probably `Optional<String>`, which also provides compile time safety.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: KAFKA-14177: Correctly support older kraft versions without FeatureLevelRecord
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r952882843
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1118,6 +1059,65 @@ private void updateWriteOffset(long offset) {
}
}
+ private void claim(int epoch) {
+ try {
+ if (curClaimEpoch != -1) {
+ throw new RuntimeException("Cannot claim leadership because we are already the " +
+ "active controller.");
+ }
+ curClaimEpoch = epoch;
+ controllerMetrics.setActive(true);
+ updateWriteOffset(lastCommittedOffset);
+ clusterControl.activate();
+
+ // Before switching to active, create an in-memory snapshot at the last committed
+ // offset. This is required because the active controller assumes that there is always
+ // an in-memory snapshot at the last committed offset.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+
+ // Prepend the activate event. It is important that this event go at the beginning
+ // of the queue rather than the end (hence prepend rather than append). It's also
+ // important not to use prepend for anything else, to preserve the ordering here.
+ queue.prepend(new ControllerWriteEvent<>("completeActivation[" + epoch + "]",
+ new CompleteActivationEvent()));
+ } catch (Throwable e) {
+ fatalFaultHandler.handleFault("exception while claiming leadership", e);
+ }
+ }
+
+ class CompleteActivationEvent implements ControllerWriteOperation<Void> {
Review Comment:
> How about BecomeActiveControllerEvent?
One issue is that this event doesn't do everything related to activation. It just completes activation (finishes up some things)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949670514
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
Review Comment:
(I mean, I guess another point of view is that empty string is just another invalid input, that we throw an exception on)
As I said, I don't feel that strongly. If you think it should be changed I can do that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r951937315
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -229,6 +243,12 @@ private ApiError updateMetadataVersion(
return invalidMetadataVersion(newVersionLevel, "Unknown metadata.version.");
}
+ // We cannot set a version earlier than IBP_3_3_IV0, since that was the first version that contained
+ // FeatureLevelRecord itself.
+ if (newVersion.isLessThan(minimumBootstrapVersion)) {
+ return invalidMetadataVersion(newVersionLevel, "Unable to set a metadata.version less than " +
+ minimumBootstrapVersion);
+ }
Review Comment:
> Edit: considering 3.0-3.2 were pre-prod versions of KRaft, it's arguably fine to make 3.3 a point of no return. If we decide to go this route we need to include this in KIP-833.
I agree that it needs to be documented in the KIP. It's an awkward situation: ideally we would have introduced the metadata.version stuff earlier than we did. But we are offering people a path forward to production versions of KRaft clusters (although not backward to pre-prod).
To put this in perspective, downgrading IBP is not even officially supported in the ZK world. There are IBP bumps you just can't roll back from in the ZK world without manual intervention. This is the same kind of thing (although hopefully it's the last one for KRaft!)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r950388163
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1118,6 +1059,65 @@ private void updateWriteOffset(long offset) {
}
}
+ private void claim(int epoch) {
+ try {
+ if (curClaimEpoch != -1) {
+ throw new RuntimeException("Cannot claim leadership because we are already the " +
+ "active controller.");
+ }
+ curClaimEpoch = epoch;
+ controllerMetrics.setActive(true);
+ updateWriteOffset(lastCommittedOffset);
+ clusterControl.activate();
+
+ // Before switching to active, create an in-memory snapshot at the last committed
+ // offset. This is required because the active controller assumes that there is always
+ // an in-memory snapshot at the last committed offset.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+
+ // Prepend the activate event. It is important that this event go at the beginning
+ // of the queue rather than the end (hence prepend rather than append). It's also
+ // important not to use prepend for anything else, to preserve the ordering here.
+ queue.prepend(new ControllerWriteEvent<>("completeActivation[" + epoch + "]",
+ new CompleteActivationEvent()));
+ } catch (Throwable e) {
+ fatalFaultHandler.handleFault("exception while claiming leadership", e);
+ }
+ }
+
+ class CompleteActivationEvent implements ControllerWriteOperation<Void> {
Review Comment:
Hmm... we do use the terminology of "active / activation" in a lot of other places.
```
controllerMetrics.setActive
isActiveController
clusterControl.activate
```
it's even in a metric...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949590579
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
Review Comment:
Out of curiosity, why not use null as the sentinel? Seems less likely to be used mistakenly.
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
+ * if it is not configured.
+ */
+ public BootstrapDirectory(
+ String directoryPath,
+ String ibp
+ ) {
+ Objects.requireNonNull(directoryPath);
+ Objects.requireNonNull(ibp);
+ this.directoryPath = directoryPath;
+ this.ibp = ibp;
+ }
+
+ public BootstrapMetadata read() throws Exception {
+ if (!Files.isDirectory(Paths.get(directoryPath))) {
+ throw new RuntimeException("No such directory as " + directoryPath);
Review Comment:
nit: perhaps it's useful to check existence separately so that we can have a clear message when either the file does not exist or it is not a directory. This might also be a useful helper to add to `Utils` or some such place.
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+
+/**
+ * The bootstrap metadata. On startup, if the metadata log is empty, we will populate the log with
+ * these records. Alternately, if log is not empty, but the metadata version is not set, we will
+ * use the version specified here.
+ */
+public class BootstrapMetadata {
+ private final List<ApiMessageAndVersion> records;
+ private final MetadataVersion metadataVersion;
+ private final String source;
+
+ public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
+ List<ApiMessageAndVersion> records = Collections.singletonList(
+ new ApiMessageAndVersion(new FeatureLevelRecord().
+ setName(MetadataVersion.FEATURE_NAME).
+ setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
+ MetadataVersion metadataVersion = null;
+ for (ApiMessageAndVersion record : records) {
+ Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
+ if (version.isPresent()) {
+ metadataVersion = version.get();
+ }
+ }
+ if (metadataVersion == null) {
+ throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
+ " was found in the bootstrap metadata from " + source);
+ }
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ if (record instanceof FeatureLevelRecord) {
+ FeatureLevelRecord featureLevel = (FeatureLevelRecord) record;
+ if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
+ return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ }
+ }
+ return Optional.empty();
+ }
+
+ BootstrapMetadata(
+ List<ApiMessageAndVersion> records,
+ MetadataVersion metadataVersion,
+ String source
+ ) {
+ Objects.requireNonNull(records);
Review Comment:
nit: usual idiom is this:
```java
this.records = Objects.requireNonNull(records);
```
Also, is our requirement stricter? Do we need a non-empty list? Otherwise, `copyWithOnlyVersion` will fail below. Finally, do we need to check the consistency requirement between `metadataVersion` and the `FeatureLevelRecord` in `records`? Perhaps justifiable to skip extra validations if we can make the constructor private.
##########
metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.metadata.util;
+
+import org.apache.kafka.common.message.LeaderChangeMessage;
+import org.apache.kafka.common.message.SnapshotFooterRecord;
+import org.apache.kafka.common.message.SnapshotHeaderRecord;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.record.ControlRecordType;
+import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
+import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.metadata.MetadataRecordSerde;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+
+/**
+ * Reads a log file containing KRaft record batches.
+ */
+public final class BatchFileReader implements Iterator<BatchFileReader.BatchAndType>, AutoCloseable {
+ private static final Logger log = LoggerFactory.getLogger(BatchFileReader.class);
+
+ public static class Builder {
+ private String path = null;
+
+ public Builder setPath(String path) {
+ this.path = path;
+ return this;
+ }
+
+ public BatchFileReader build() throws Exception {
+ if (path == null) {
+ throw new RuntimeException("You must specify a path.");
+ }
+ FileRecords fileRecords = FileRecords.open(new File(path), false);
+ try {
+ return new BatchFileReader(fileRecords);
+ } catch (Throwable e) {
+ Utils.closeQuietly(fileRecords, "fileRecords");
+ throw e;
+ }
+ }
+ }
+
+ public static class BatchAndType {
+ private final Batch<ApiMessageAndVersion> batch;
+ private final boolean isControl;
+
+ public BatchAndType(Batch<ApiMessageAndVersion> batch, boolean isControl) {
+ this.batch = batch;
+ this.isControl = isControl;
+ }
+
+ public Batch<ApiMessageAndVersion> batch() {
+ return batch;
+ }
+
+ public boolean isControl() {
+ return isControl;
+ }
+ }
+
+ private final FileRecords fileRecords;
+ private Iterator<FileChannelRecordBatch> batchIterator;
+ private final MetadataRecordSerde serde;
+
+ private BatchFileReader(FileRecords fileRecords) {
+ this.fileRecords = fileRecords;
+ this.batchIterator = fileRecords.batchIterator();
+ this.serde = new MetadataRecordSerde();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.batchIterator.hasNext();
+ }
+
+ @Override
+ public BatchAndType next() {
+ FileChannelRecordBatch input = batchIterator.next();
+ if (input.isControlBatch()) {
+ return nextControlBatch(input);
+ } else {
+ return nextMetadataBatch(input);
+ }
+ }
+
+ private BatchAndType nextControlBatch(FileChannelRecordBatch input) {
+ List<ApiMessageAndVersion> messages = new ArrayList<>();
+ for (Iterator<Record> iter = input.iterator(); iter.hasNext(); ) {
+ Record record = iter.next();
+ try {
+ short typeId = ControlRecordType.parseTypeId(record.key());
+ ControlRecordType type = ControlRecordType.fromTypeId(typeId);
+ switch (type) {
+ case LEADER_CHANGE: {
+ LeaderChangeMessage message = new LeaderChangeMessage();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ case SNAPSHOT_HEADER: {
+ SnapshotHeaderRecord message = new SnapshotHeaderRecord();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ case SNAPSHOT_FOOTER: {
+ SnapshotFooterRecord message = new SnapshotFooterRecord();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ default:
+ throw new RuntimeException("Unsupported control record type " + type + " at offset " +
+ record.offset());
+ }
+ } catch (Throwable e) {
+ throw new RuntimeException("Unable to read control record at offset " + record.offset(), e);
+ }
+ }
+ return new BatchAndType(Batch.data(
+ input.baseOffset(),
+ input.partitionLeaderEpoch(),
+ input.maxTimestamp(),
+ input.sizeInBytes(),
+ messages), true);
+ }
+
+ private BatchAndType nextMetadataBatch(FileChannelRecordBatch input) {
+ List<ApiMessageAndVersion> messages = new ArrayList<>();
+ for (Record record : input) {
+ ByteBufferAccessor accessor = new ByteBufferAccessor(record.value());
+ try {
+ ApiMessageAndVersion messageAndVersion = serde.read(accessor, record.valueSize());
+ messages.add(messageAndVersion);
+ } catch (Throwable e) {
+ log.error("unable to read metadata record at offset {}", record.offset(), e);
Review Comment:
Hmm, why don't we want to propagate this? Seems dangerous to continue.
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1142,15 +1142,19 @@ private void renounce() {
purgatory.failAll(newNotControllerException());
if (snapshotRegistry.hasSnapshot(lastCommittedOffset)) {
- newBytesSinceLastSnapshot = 0;
snapshotRegistry.revertToSnapshot(lastCommittedOffset);
authorizer.ifPresent(a -> a.loadSnapshot(aclControlManager.idToAcl()));
} else {
- resetState();
+ log.warn("Unable to find last committed offset {} in snapshot registry; resetting " +
Review Comment:
Is not having a snapshot for the committed offset an expected or unexpected state?
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -222,52 +227,130 @@ public void testFeatureControlIterator() throws Exception {
manager.iterator(Long.MAX_VALUE));
}
+ private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
+ new FeatureControlManager.Builder().
+ setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
+ MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+
@Test
public void testApplyMetadataVersionChangeRecord() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).build();
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_0_IV1.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_0_IV1, manager.metadataVersion());
+ setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
+ assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
}
@Test
- public void testDowngradeMetadataVersion() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_2_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
- build();
- assertEquals(manager.metadataVersion(), MetadataVersion.IBP_3_3_IV0);
+ public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ assertEquals(ControllerResult.of(Collections.emptyList(),
+ singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
+ "Invalid update version 3 for feature metadata.version. Local controller 0 only " +
+ "supports versions 4-7"))),
+ manager.updateFeatures(
+ singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_2_IV0.featureLevel()),
+ singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.UNSAFE_DOWNGRADE),
+ emptyMap(),
+ true));
+ }
+
+ @Test
+ public void testCannotDowngradeToHigherVersion() {
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ assertEquals(ControllerResult.of(Collections.emptyList(),
+ singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
+ "Invalid update version 7 for feature metadata.version. Can't downgrade to a " +
+ "newer version."))),
+ manager.updateFeatures(
+ singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_3_IV3.featureLevel()),
+ singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.SAFE_DOWNGRADE),
Review Comment:
Is the downgrade feature supported yet?
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
+ * if it is not configured.
+ */
+ public BootstrapDirectory(
+ String directoryPath,
+ String ibp
+ ) {
+ Objects.requireNonNull(directoryPath);
+ Objects.requireNonNull(ibp);
+ this.directoryPath = directoryPath;
+ this.ibp = ibp;
+ }
+
+ public BootstrapMetadata read() throws Exception {
+ if (!Files.isDirectory(Paths.get(directoryPath))) {
+ throw new RuntimeException("No such directory as " + directoryPath);
+ }
+ Path binaryBootstrapPath = Paths.get(directoryPath, BINARY_BOOTSTRAP);
+ if (!Files.exists(binaryBootstrapPath)) {
+ return readFromConfiguration();
+ } else {
+ return readFromBinaryFile(binaryBootstrapPath.toString());
+ }
+ }
+
+ BootstrapMetadata readFromConfiguration() {
+ if (ibp.isEmpty()) {
+ return BootstrapMetadata.fromVersion(MetadataVersion.latest(),
+ "the default bootstrap file which sets the latest metadata.version, " +
+ "since no bootstrap file was found, and " + INTER_BROKER_PROTOCOL_CONFIG_KEY +
+ " was not configured.");
+ }
+ MetadataVersion version = MetadataVersion.fromVersionString(ibp);
+ if (version.isLessThan(MINIMUM_KRAFT_VERSION)) {
+ return BootstrapMetadata.fromVersion(MINIMUM_KRAFT_VERSION,
+ "a default bootstrap file setting the minimum supported KRaft metadata " +
Review Comment:
The source descriptions in here are pretty long-winded. I am not so sure it remains coherent when used in a case like the following:
```java
throw new RuntimeException("Unable to load metadata from " + source + ": bootstrap " +
"metadata versions less than " + MetadataVersion.MINIMUM_KRAFT_VERSION + " are " +
"not supported.");
```
Perhaps `source` should be concise and we can save the verbose message for logging?
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+
+/**
+ * The bootstrap metadata. On startup, if the metadata log is empty, we will populate the log with
+ * these records. Alternately, if log is not empty, but the metadata version is not set, we will
+ * use the version specified here.
+ */
+public class BootstrapMetadata {
+ private final List<ApiMessageAndVersion> records;
+ private final MetadataVersion metadataVersion;
+ private final String source;
+
+ public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
+ List<ApiMessageAndVersion> records = Collections.singletonList(
+ new ApiMessageAndVersion(new FeatureLevelRecord().
+ setName(MetadataVersion.FEATURE_NAME).
+ setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
+ MetadataVersion metadataVersion = null;
+ for (ApiMessageAndVersion record : records) {
+ Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
+ if (version.isPresent()) {
+ metadataVersion = version.get();
+ }
+ }
+ if (metadataVersion == null) {
+ throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
+ " was found in the bootstrap metadata from " + source);
+ }
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ if (record instanceof FeatureLevelRecord) {
+ FeatureLevelRecord featureLevel = (FeatureLevelRecord) record;
+ if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
+ return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ }
+ }
+ return Optional.empty();
+ }
+
+ BootstrapMetadata(
+ List<ApiMessageAndVersion> records,
+ MetadataVersion metadataVersion,
+ String source
+ ) {
+ Objects.requireNonNull(records);
+ this.records = records;
+ if (metadataVersion.isLessThan(MetadataVersion.MINIMUM_KRAFT_VERSION)) {
+ throw new RuntimeException("Unable to load metadata from " + source + ": bootstrap " +
+ "metadata versions less than " + MetadataVersion.MINIMUM_KRAFT_VERSION + " are " +
+ "not supported.");
+ }
+ this.metadataVersion = metadataVersion;
+ Objects.requireNonNull(source);
+ this.source = source;
+ }
+
+ public List<ApiMessageAndVersion> records() {
+ return records;
+ }
+
+ public MetadataVersion metadataVersion() {
+ return metadataVersion;
+ }
+
+ public String source() {
+ return source;
+ }
+
+ public BootstrapMetadata copyWithOnlyVersion() {
+ List<ApiMessageAndVersion> newRecords = new ArrayList<>();
+ for (ApiMessageAndVersion record : records) {
+ if (recordToMetadataVersion(record.message()).isPresent()) {
+ newRecords.clear();
Review Comment:
Not sure why we need a list if we clear every time before we add a record. Could it just be a variable for a single instance of `ApiMessageAndVersion`?
##########
metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.metadata.util;
+
+import org.apache.kafka.common.message.LeaderChangeMessage;
+import org.apache.kafka.common.message.SnapshotFooterRecord;
+import org.apache.kafka.common.message.SnapshotHeaderRecord;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.record.ControlRecordType;
+import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
+import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.metadata.MetadataRecordSerde;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+
+/**
+ * Reads a log file containing KRaft record batches.
+ */
+public final class BatchFileReader implements Iterator<BatchFileReader.BatchAndType>, AutoCloseable {
+ private static final Logger log = LoggerFactory.getLogger(BatchFileReader.class);
+
+ public static class Builder {
+ private String path = null;
+
+ public Builder setPath(String path) {
+ this.path = path;
Review Comment:
nit: perhaps we could require the path be non-null?
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+
+/**
+ * The bootstrap metadata. On startup, if the metadata log is empty, we will populate the log with
+ * these records. Alternately, if log is not empty, but the metadata version is not set, we will
+ * use the version specified here.
+ */
+public class BootstrapMetadata {
+ private final List<ApiMessageAndVersion> records;
+ private final MetadataVersion metadataVersion;
+ private final String source;
+
+ public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
+ List<ApiMessageAndVersion> records = Collections.singletonList(
+ new ApiMessageAndVersion(new FeatureLevelRecord().
+ setName(MetadataVersion.FEATURE_NAME).
+ setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
Review Comment:
Just wondering about the use of 0 here as the version. Is this a requirement for all bootstrap snapshots in order to maintain compatibility?
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -188,12 +182,16 @@ private ApiError updateFeature(
}
}
- if (currentVersion != null && newVersion < currentVersion) {
+ if (newVersion < currentVersion) {
if (upgradeType.equals(FeatureUpdate.UpgradeType.UPGRADE)) {
Review Comment:
Not from this patch, but I was trying to trace back where the `UpgradeType` is determined. I see logic in `QuorumController.updateFeatures` to convert from the code in the request, but it doesn't look like we are checking for the `UNKNOWN` type. Am I missing the validation somewhere?
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
Review Comment:
The comment mentions `bootstrap.snapshot`, but below we seem to use `bootstrap.checkpoint`. Which is right?
##########
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java:
##########
@@ -952,6 +966,7 @@ public void testEarlyControllerResults() throws Throwable {
}
}
+ @Disabled // TODO: need to fix leader election in LocalLog.
Review Comment:
Did this break from this patch or is it flaky?
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -222,52 +227,130 @@ public void testFeatureControlIterator() throws Exception {
manager.iterator(Long.MAX_VALUE));
}
+ private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
+ new FeatureControlManager.Builder().
+ setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
+ MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+
@Test
public void testApplyMetadataVersionChangeRecord() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).build();
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_0_IV1.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_0_IV1, manager.metadataVersion());
+ setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
+ assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
}
@Test
- public void testDowngradeMetadataVersion() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_2_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
- build();
- assertEquals(manager.metadataVersion(), MetadataVersion.IBP_3_3_IV0);
+ public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ assertEquals(ControllerResult.of(Collections.emptyList(),
+ singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
Review Comment:
By the way, the overloading of `INVALID_UPDATE_VERSION` is a little unfortunate. The error makes sense here and seems reasonable, but we also use it for "partition epoch" mismatches in `AlterPartition`. I almost wonder if it would be more consistent with other APIs to use `INVALID_REQUEST` here.
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+
+/**
+ * The bootstrap metadata. On startup, if the metadata log is empty, we will populate the log with
+ * these records. Alternately, if log is not empty, but the metadata version is not set, we will
+ * use the version specified here.
+ */
+public class BootstrapMetadata {
+ private final List<ApiMessageAndVersion> records;
+ private final MetadataVersion metadataVersion;
+ private final String source;
+
+ public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
+ List<ApiMessageAndVersion> records = Collections.singletonList(
+ new ApiMessageAndVersion(new FeatureLevelRecord().
+ setName(MetadataVersion.FEATURE_NAME).
+ setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
+ MetadataVersion metadataVersion = null;
+ for (ApiMessageAndVersion record : records) {
+ Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
+ if (version.isPresent()) {
+ metadataVersion = version.get();
+ }
+ }
+ if (metadataVersion == null) {
+ throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
+ " was found in the bootstrap metadata from " + source);
+ }
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ if (record instanceof FeatureLevelRecord) {
+ FeatureLevelRecord featureLevel = (FeatureLevelRecord) record;
+ if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
+ return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ }
+ }
+ return Optional.empty();
+ }
+
+ BootstrapMetadata(
+ List<ApiMessageAndVersion> records,
+ MetadataVersion metadataVersion,
+ String source
+ ) {
+ Objects.requireNonNull(records);
+ this.records = records;
+ if (metadataVersion.isLessThan(MetadataVersion.MINIMUM_KRAFT_VERSION)) {
+ throw new RuntimeException("Unable to load metadata from " + source + ": bootstrap " +
+ "metadata versions less than " + MetadataVersion.MINIMUM_KRAFT_VERSION + " are " +
+ "not supported.");
+ }
+ this.metadataVersion = metadataVersion;
+ Objects.requireNonNull(source);
+ this.source = source;
+ }
+
+ public List<ApiMessageAndVersion> records() {
+ return records;
+ }
+
+ public MetadataVersion metadataVersion() {
+ return metadataVersion;
+ }
+
+ public String source() {
+ return source;
+ }
+
+ public BootstrapMetadata copyWithOnlyVersion() {
+ List<ApiMessageAndVersion> newRecords = new ArrayList<>();
+ for (ApiMessageAndVersion record : records) {
+ if (recordToMetadataVersion(record.message()).isPresent()) {
+ newRecords.clear();
+ newRecords.add(record);
+ }
+ }
+ if (newRecords.isEmpty()) {
+ throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
+ " was found in the bootstrap metadata from " + source);
+ }
+ return new BootstrapMetadata(Collections.unmodifiableList(newRecords),
+ metadataVersion, source);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(records, metadataVersion, source);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || !(o instanceof BootstrapMetadata)) return false;
Review Comment:
nit: the first check `o == null` is covered by the second check
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1118,6 +1059,65 @@ private void updateWriteOffset(long offset) {
}
}
+ private void claim(int epoch) {
+ try {
+ if (curClaimEpoch != -1) {
+ throw new RuntimeException("Cannot claim leadership because we are already the " +
+ "active controller.");
+ }
+ curClaimEpoch = epoch;
+ controllerMetrics.setActive(true);
+ updateWriteOffset(lastCommittedOffset);
+ clusterControl.activate();
+
+ // Before switching to active, create an in-memory snapshot at the last committed
+ // offset. This is required because the active controller assumes that there is always
+ // an in-memory snapshot at the last committed offset.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+
+ // Prepend the activate event. It is important that this event go at the beginning
+ // of the queue rather than the end (hence prepend rather than append). It's also
+ // important not to use prepend for anything else, to preserve the ordering here.
+ queue.prepend(new ControllerWriteEvent<>("completeActivation[" + epoch + "]",
+ new CompleteActivationEvent()));
+ } catch (Throwable e) {
+ fatalFaultHandler.handleFault("exception while claiming leadership", e);
+ }
+ }
+
+ class CompleteActivationEvent implements ControllerWriteOperation<Void> {
+ @Override
+ public ControllerResult<Void> generateRecordsAndResult() throws Exception {
+ List<ApiMessageAndVersion> records = new ArrayList<>();
+ if (logReplayTracker.empty()) {
+ // If no records have been replayed, we need to write out the bootstrap records.
+ // This will include the new metadata.version, as well as things like SCRAM
+ // initialization, etc.
+ log.warn("The metadata log appears to be empty. Appending {} bootstrap record(s) " +
Review Comment:
Are these messages really warnings or could they be info?
##########
server-common/src/test/java/org/apache/kafka/server/common/MetadataVersionTest.java:
##########
@@ -72,20 +71,18 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
class MetadataVersionTest {
-
@Test
public void testFeatureLevel() {
- MetadataVersion[] metadataVersions = MetadataVersion.VERSIONS;
- int firstFeatureLevelIndex = Arrays.asList(metadataVersions).indexOf(MetadataVersion.MINIMUM_KRAFT_VERSION);
- for (int i = 0; i < firstFeatureLevelIndex; i++) {
- assertTrue(metadataVersions[i].featureLevel() < 0);
+ int i = 0;
+ while (i < MetadataVersion.VERSIONS.length &&
Review Comment:
nit: The new version of this test seems more obscure than the original. It seems more straightforward to assert 1) all versions below IBP_3_0_IV1 have -1 feature level, and 2) that all subsequent version increment feature level by 1. Also, it would be helpful to have a better name for the test so that we don't have to decipher what it's trying to do. For example, perhaps `testSequentialFeatureLevel` or something like that.
##########
metadata/src/main/java/org/apache/kafka/controller/LogReplayTracker.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.controller;
+
+import java.util.Optional;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.metadata.bootstrap.BootstrapMetadata;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.slf4j.Logger;
+
+
+public class LogReplayTracker {
+ public static class Builder {
+ private LogContext logContext = null;
+
+ Builder setLogContext(LogContext logContext) {
+ this.logContext = logContext;
+ return this;
+ }
+
+ public LogReplayTracker build() {
+ if (logContext == null) logContext = new LogContext();
+ return new LogReplayTracker(logContext);
+ }
+ }
+
+ /**
+ * The slf4j logger.
+ */
+ private final Logger log;
+
+ /**
+ * True if we haven't replayed any records yet.
+ */
+ private boolean empty;
+
+ /**
+ * True if we haven't replayed any metadata.version records yet.
+ */
+ private boolean versionless;
Review Comment:
nit: the name confused me a little bit. How about `hasReadMetadataVersion` or something like that? Or perhaps even simpler, we could record the `MetadataVersion` itself.
##########
metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.metadata.util;
+
+import org.apache.kafka.common.message.LeaderChangeMessage;
+import org.apache.kafka.common.message.SnapshotFooterRecord;
+import org.apache.kafka.common.message.SnapshotHeaderRecord;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.record.ControlRecordType;
+import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
+import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.metadata.MetadataRecordSerde;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+
+/**
+ * Reads a log file containing KRaft record batches.
+ */
+public final class BatchFileReader implements Iterator<BatchFileReader.BatchAndType>, AutoCloseable {
+ private static final Logger log = LoggerFactory.getLogger(BatchFileReader.class);
+
+ public static class Builder {
+ private String path = null;
+
+ public Builder setPath(String path) {
+ this.path = path;
+ return this;
+ }
+
+ public BatchFileReader build() throws Exception {
+ if (path == null) {
+ throw new RuntimeException("You must specify a path.");
+ }
+ FileRecords fileRecords = FileRecords.open(new File(path), false);
+ try {
+ return new BatchFileReader(fileRecords);
+ } catch (Throwable e) {
+ Utils.closeQuietly(fileRecords, "fileRecords");
+ throw e;
+ }
+ }
+ }
+
+ public static class BatchAndType {
+ private final Batch<ApiMessageAndVersion> batch;
+ private final boolean isControl;
+
+ public BatchAndType(Batch<ApiMessageAndVersion> batch, boolean isControl) {
+ this.batch = batch;
+ this.isControl = isControl;
+ }
+
+ public Batch<ApiMessageAndVersion> batch() {
+ return batch;
+ }
+
+ public boolean isControl() {
+ return isControl;
+ }
+ }
+
+ private final FileRecords fileRecords;
+ private Iterator<FileChannelRecordBatch> batchIterator;
+ private final MetadataRecordSerde serde;
+
+ private BatchFileReader(FileRecords fileRecords) {
+ this.fileRecords = fileRecords;
+ this.batchIterator = fileRecords.batchIterator();
+ this.serde = new MetadataRecordSerde();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.batchIterator.hasNext();
+ }
+
+ @Override
+ public BatchAndType next() {
+ FileChannelRecordBatch input = batchIterator.next();
+ if (input.isControlBatch()) {
+ return nextControlBatch(input);
+ } else {
+ return nextMetadataBatch(input);
+ }
+ }
+
+ private BatchAndType nextControlBatch(FileChannelRecordBatch input) {
+ List<ApiMessageAndVersion> messages = new ArrayList<>();
+ for (Iterator<Record> iter = input.iterator(); iter.hasNext(); ) {
+ Record record = iter.next();
+ try {
+ short typeId = ControlRecordType.parseTypeId(record.key());
+ ControlRecordType type = ControlRecordType.fromTypeId(typeId);
+ switch (type) {
+ case LEADER_CHANGE: {
+ LeaderChangeMessage message = new LeaderChangeMessage();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ case SNAPSHOT_HEADER: {
+ SnapshotHeaderRecord message = new SnapshotHeaderRecord();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ case SNAPSHOT_FOOTER: {
+ SnapshotFooterRecord message = new SnapshotFooterRecord();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ default:
+ throw new RuntimeException("Unsupported control record type " + type + " at offset " +
+ record.offset());
+ }
+ } catch (Throwable e) {
+ throw new RuntimeException("Unable to read control record at offset " + record.offset(), e);
+ }
+ }
+ return new BatchAndType(Batch.data(
+ input.baseOffset(),
+ input.partitionLeaderEpoch(),
+ input.maxTimestamp(),
+ input.sizeInBytes(),
+ messages), true);
+ }
+
+ private BatchAndType nextMetadataBatch(FileChannelRecordBatch input) {
+ List<ApiMessageAndVersion> messages = new ArrayList<>();
+ for (Record record : input) {
+ ByteBufferAccessor accessor = new ByteBufferAccessor(record.value());
+ try {
+ ApiMessageAndVersion messageAndVersion = serde.read(accessor, record.valueSize());
+ messages.add(messageAndVersion);
+ } catch (Throwable e) {
+ log.error("unable to read metadata record at offset {}", record.offset(), e);
+ }
+ }
+ return new BatchAndType(Batch.data(
+ input.baseOffset(),
+ input.partitionLeaderEpoch(),
+ input.maxTimestamp(),
+ input.sizeInBytes(),
+ messages), false);
+ }
+
+ @Override
+ public void close() {
+ try {
+ fileRecords.closeHandlers();
Review Comment:
Why not use `close()` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r950502736
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -188,12 +182,16 @@ private ApiError updateFeature(
}
}
- if (currentVersion != null && newVersion < currentVersion) {
+ if (newVersion < currentVersion) {
if (upgradeType.equals(FeatureUpdate.UpgradeType.UPGRADE)) {
Review Comment:
Ah, ok. I missed the check.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r950450852
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
+ * if it is not configured.
+ */
+ public BootstrapDirectory(
+ String directoryPath,
+ String ibp
+ ) {
+ Objects.requireNonNull(directoryPath);
+ Objects.requireNonNull(ibp);
+ this.directoryPath = directoryPath;
+ this.ibp = ibp;
+ }
+
+ public BootstrapMetadata read() throws Exception {
+ if (!Files.isDirectory(Paths.get(directoryPath))) {
+ throw new RuntimeException("No such directory as " + directoryPath);
+ }
+ Path binaryBootstrapPath = Paths.get(directoryPath, BINARY_BOOTSTRAP);
+ if (!Files.exists(binaryBootstrapPath)) {
+ return readFromConfiguration();
+ } else {
+ return readFromBinaryFile(binaryBootstrapPath.toString());
+ }
+ }
+
+ BootstrapMetadata readFromConfiguration() {
+ if (ibp.isEmpty()) {
+ return BootstrapMetadata.fromVersion(MetadataVersion.latest(),
+ "the default bootstrap file which sets the latest metadata.version, " +
+ "since no bootstrap file was found, and " + INTER_BROKER_PROTOCOL_CONFIG_KEY +
+ " was not configured.");
+ }
+ MetadataVersion version = MetadataVersion.fromVersionString(ibp);
+ if (version.isLessThan(MINIMUM_KRAFT_VERSION)) {
+ return BootstrapMetadata.fromVersion(MINIMUM_KRAFT_VERSION,
+ "a default bootstrap file setting the minimum supported KRaft metadata " +
Review Comment:
I don't mind having verbose messages, but I don't think it's asking too much for them to be grammatical.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949671940
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+
+/**
+ * The bootstrap metadata. On startup, if the metadata log is empty, we will populate the log with
+ * these records. Alternately, if log is not empty, but the metadata version is not set, we will
+ * use the version specified here.
+ */
+public class BootstrapMetadata {
+ private final List<ApiMessageAndVersion> records;
+ private final MetadataVersion metadataVersion;
+ private final String source;
+
+ public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
+ List<ApiMessageAndVersion> records = Collections.singletonList(
+ new ApiMessageAndVersion(new FeatureLevelRecord().
+ setName(MetadataVersion.FEATURE_NAME).
+ setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
+ MetadataVersion metadataVersion = null;
+ for (ApiMessageAndVersion record : records) {
+ Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
+ if (version.isPresent()) {
+ metadataVersion = version.get();
+ }
+ }
+ if (metadataVersion == null) {
+ throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
+ " was found in the bootstrap metadata from " + source);
+ }
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ if (record instanceof FeatureLevelRecord) {
+ FeatureLevelRecord featureLevel = (FeatureLevelRecord) record;
+ if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
+ return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ }
+ }
+ return Optional.empty();
+ }
+
+ BootstrapMetadata(
+ List<ApiMessageAndVersion> records,
+ MetadataVersion metadataVersion,
+ String source
+ ) {
+ Objects.requireNonNull(records);
Review Comment:
> nit: usual idiom is this: this.records = Objects.requireNonNull(records);
Fixed
> Perhaps justifiable to skip extra validations if we can make the constructor private.
It's package-private, which in practice will keep anyone from doing something weird because the users are outside `package org.apache.kafka.metadata.bootstrap`
I wanted to do something slightly weird in a unit test, which is why it was package-private rather than fully private. I can't recall what at the moment. Should be ok?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: KAFKA-14177: Correctly support older kraft versions without FeatureLevelRecord
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r952893463
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
Review Comment:
I will just make this an `Optional<MetadataVersion>` for simplicity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
mumrah commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r951824786
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -229,6 +243,12 @@ private ApiError updateMetadataVersion(
return invalidMetadataVersion(newVersionLevel, "Unknown metadata.version.");
}
+ // We cannot set a version earlier than IBP_3_3_IV0, since that was the first version that contained
+ // FeatureLevelRecord itself.
+ if (newVersion.isLessThan(minimumBootstrapVersion)) {
+ return invalidMetadataVersion(newVersionLevel, "Unable to set a metadata.version less than " +
+ minimumBootstrapVersion);
+ }
Review Comment:
If we don't allow downgrading the flag, do we have a way of unsetting it?
If a downgraded broker comes up with 3.2, it will use whatever IBP is set to determine its compatibility. The non-downgraded brokers could be using newer replication protocols that the 3.2 broker doesn't support. In the IBP world, an operator would first do a rolling downgrade of the IBP before lowering the software version.
Another issue is registration. If the cluster is on 3.3 and `metadata.version` is set to 3.3IV0 and a user wants to downgrade to 3.2, when the first downgraded broker comes up it will register with an empty set of features and the controller will infer (minSupported=3.0IV1, maxSupported=3.0IV1) and reject the registration.
If we let the operator unset the `metadata.version`, we can let the absence of the record infer 3.0IV1 in the cluster and then safely downgrade brokers.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] mumrah commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
mumrah commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r951824786
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -229,6 +243,12 @@ private ApiError updateMetadataVersion(
return invalidMetadataVersion(newVersionLevel, "Unknown metadata.version.");
}
+ // We cannot set a version earlier than IBP_3_3_IV0, since that was the first version that contained
+ // FeatureLevelRecord itself.
+ if (newVersion.isLessThan(minimumBootstrapVersion)) {
+ return invalidMetadataVersion(newVersionLevel, "Unable to set a metadata.version less than " +
+ minimumBootstrapVersion);
+ }
Review Comment:
If we don't allow downgrading the flag, do we have a way of unsetting it?
If a downgraded broker comes up with 3.2, it will use whatever IBP is set to determine its compatibility. The non-downgraded brokers could be using newer replication protocols that the 3.2 broker doesn't support. In the IBP world, an operator would first do a rolling downgrade of the IBP before lowering the software version.
Another issue is registration. If the cluster is on 3.3 and `metadata.version` is set to 3.3IV0 and a user wants to downgrade to 3.2, when the first downgraded broker comes up it will register with an empty set of features and the controller will infer (minSupported=3.0IV1, maxSupported=3.0IV1) and reject the registration.
If we let the operator unset the `metadata.version`, we can let the absence of the record infer 3.0IV1 in the cluster and then safely downgrade brokers.
Edit: considering 3.0-3.2 were pre-prod versions of KRaft, it's arguably fine to make 3.3 a point of no return. If we decide to go this route we need to include this in KIP-833
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r952055798
##########
server-common/src/main/java/org/apache/kafka/server/common/MetadataVersion.java:
##########
@@ -167,8 +167,18 @@ public enum MetadataVersion {
public static final String FEATURE_NAME = "metadata.version";
+ /**
+ * The first version we currently support in KRaft. We chose 3.3IV0 since it is the first
Review Comment:
This comment needs to be fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r950458081
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1118,6 +1059,65 @@ private void updateWriteOffset(long offset) {
}
}
+ private void claim(int epoch) {
+ try {
+ if (curClaimEpoch != -1) {
+ throw new RuntimeException("Cannot claim leadership because we are already the " +
+ "active controller.");
+ }
+ curClaimEpoch = epoch;
+ controllerMetrics.setActive(true);
+ updateWriteOffset(lastCommittedOffset);
+ clusterControl.activate();
+
+ // Before switching to active, create an in-memory snapshot at the last committed
+ // offset. This is required because the active controller assumes that there is always
+ // an in-memory snapshot at the last committed offset.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+
+ // Prepend the activate event. It is important that this event go at the beginning
+ // of the queue rather than the end (hence prepend rather than append). It's also
+ // important not to use prepend for anything else, to preserve the ordering here.
+ queue.prepend(new ControllerWriteEvent<>("completeActivation[" + epoch + "]",
+ new CompleteActivationEvent()));
+ } catch (Throwable e) {
+ fatalFaultHandler.handleFault("exception while claiming leadership", e);
+ }
+ }
+
+ class CompleteActivationEvent implements ControllerWriteOperation<Void> {
Review Comment:
How about `BecomeActiveControllerEvent`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949686749
##########
server-common/src/test/java/org/apache/kafka/server/common/MetadataVersionTest.java:
##########
@@ -72,20 +71,18 @@
import static org.junit.jupiter.api.Assertions.assertTrue;
class MetadataVersionTest {
-
@Test
public void testFeatureLevel() {
- MetadataVersion[] metadataVersions = MetadataVersion.VERSIONS;
- int firstFeatureLevelIndex = Arrays.asList(metadataVersions).indexOf(MetadataVersion.MINIMUM_KRAFT_VERSION);
- for (int i = 0; i < firstFeatureLevelIndex; i++) {
- assertTrue(metadataVersions[i].featureLevel() < 0);
+ int i = 0;
+ while (i < MetadataVersion.VERSIONS.length &&
Review Comment:
Good points. I will separate this out into `testKRaftFeatureLevelsBefore3_0_IV1` and `testKRaftFeatureLevelsAtAndAfter3_0_IV1`, which are hopefully self-explanatory.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on PR #12513:
URL: https://github.com/apache/kafka/pull/12513#issuecomment-1216957153
There was some kind of build issue where Apache Rat (a license scanner) tried to look at a jenkins build output file.
```
[2022-08-16T04:06:02.647Z] > Task :rat
[2022-08-16T04:06:02.647Z] Files with unapproved licenses:
[2022-08-16T04:06:02.647Z] - /home/jenkins/workspace/Kafka_kafka-pr_PR-12513/buildoutput.log
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r950487364
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1118,6 +1059,65 @@ private void updateWriteOffset(long offset) {
}
}
+ private void claim(int epoch) {
+ try {
+ if (curClaimEpoch != -1) {
+ throw new RuntimeException("Cannot claim leadership because we are already the " +
+ "active controller.");
+ }
+ curClaimEpoch = epoch;
+ controllerMetrics.setActive(true);
+ updateWriteOffset(lastCommittedOffset);
+ clusterControl.activate();
+
+ // Before switching to active, create an in-memory snapshot at the last committed
+ // offset. This is required because the active controller assumes that there is always
+ // an in-memory snapshot at the last committed offset.
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+
+ // Prepend the activate event. It is important that this event go at the beginning
+ // of the queue rather than the end (hence prepend rather than append). It's also
+ // important not to use prepend for anything else, to preserve the ordering here.
+ queue.prepend(new ControllerWriteEvent<>("completeActivation[" + epoch + "]",
+ new CompleteActivationEvent()));
+ } catch (Throwable e) {
+ fatalFaultHandler.handleFault("exception while claiming leadership", e);
+ }
+ }
+
+ class CompleteActivationEvent implements ControllerWriteOperation<Void> {
+ @Override
+ public ControllerResult<Void> generateRecordsAndResult() throws Exception {
+ List<ApiMessageAndVersion> records = new ArrayList<>();
+ if (logReplayTracker.empty()) {
+ // If no records have been replayed, we need to write out the bootstrap records.
+ // This will include the new metadata.version, as well as things like SCRAM
+ // initialization, etc.
+ log.warn("The metadata log appears to be empty. Appending {} bootstrap record(s) " +
Review Comment:
It is an important event, but warning level messages tend to be interpreted by users as problems, so I always hesitate to use them if it is an expected event.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949665682
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
Review Comment:
Sorry, this should read `bootstrap.checkpoint` (although now that I think about it more, I question whether we should be using the "checkpoint" suffix here...) I will change it to `bootstrap.checkpoint` for now since that's what we had.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949678938
##########
metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.metadata.util;
+
+import org.apache.kafka.common.message.LeaderChangeMessage;
+import org.apache.kafka.common.message.SnapshotFooterRecord;
+import org.apache.kafka.common.message.SnapshotHeaderRecord;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.record.ControlRecordType;
+import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
+import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.metadata.MetadataRecordSerde;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+
+/**
+ * Reads a log file containing KRaft record batches.
+ */
+public final class BatchFileReader implements Iterator<BatchFileReader.BatchAndType>, AutoCloseable {
+ private static final Logger log = LoggerFactory.getLogger(BatchFileReader.class);
+
+ public static class Builder {
+ private String path = null;
+
+ public Builder setPath(String path) {
+ this.path = path;
+ return this;
+ }
+
+ public BatchFileReader build() throws Exception {
+ if (path == null) {
+ throw new RuntimeException("You must specify a path.");
+ }
+ FileRecords fileRecords = FileRecords.open(new File(path), false);
+ try {
+ return new BatchFileReader(fileRecords);
+ } catch (Throwable e) {
+ Utils.closeQuietly(fileRecords, "fileRecords");
+ throw e;
+ }
+ }
+ }
+
+ public static class BatchAndType {
+ private final Batch<ApiMessageAndVersion> batch;
+ private final boolean isControl;
+
+ public BatchAndType(Batch<ApiMessageAndVersion> batch, boolean isControl) {
+ this.batch = batch;
+ this.isControl = isControl;
+ }
+
+ public Batch<ApiMessageAndVersion> batch() {
+ return batch;
+ }
+
+ public boolean isControl() {
+ return isControl;
+ }
+ }
+
+ private final FileRecords fileRecords;
+ private Iterator<FileChannelRecordBatch> batchIterator;
+ private final MetadataRecordSerde serde;
+
+ private BatchFileReader(FileRecords fileRecords) {
+ this.fileRecords = fileRecords;
+ this.batchIterator = fileRecords.batchIterator();
+ this.serde = new MetadataRecordSerde();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.batchIterator.hasNext();
+ }
+
+ @Override
+ public BatchAndType next() {
+ FileChannelRecordBatch input = batchIterator.next();
+ if (input.isControlBatch()) {
+ return nextControlBatch(input);
+ } else {
+ return nextMetadataBatch(input);
+ }
+ }
+
+ private BatchAndType nextControlBatch(FileChannelRecordBatch input) {
+ List<ApiMessageAndVersion> messages = new ArrayList<>();
+ for (Iterator<Record> iter = input.iterator(); iter.hasNext(); ) {
+ Record record = iter.next();
+ try {
+ short typeId = ControlRecordType.parseTypeId(record.key());
+ ControlRecordType type = ControlRecordType.fromTypeId(typeId);
+ switch (type) {
+ case LEADER_CHANGE: {
+ LeaderChangeMessage message = new LeaderChangeMessage();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ case SNAPSHOT_HEADER: {
+ SnapshotHeaderRecord message = new SnapshotHeaderRecord();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ case SNAPSHOT_FOOTER: {
+ SnapshotFooterRecord message = new SnapshotFooterRecord();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ default:
+ throw new RuntimeException("Unsupported control record type " + type + " at offset " +
+ record.offset());
+ }
+ } catch (Throwable e) {
+ throw new RuntimeException("Unable to read control record at offset " + record.offset(), e);
+ }
+ }
+ return new BatchAndType(Batch.data(
+ input.baseOffset(),
+ input.partitionLeaderEpoch(),
+ input.maxTimestamp(),
+ input.sizeInBytes(),
+ messages), true);
+ }
+
+ private BatchAndType nextMetadataBatch(FileChannelRecordBatch input) {
+ List<ApiMessageAndVersion> messages = new ArrayList<>();
+ for (Record record : input) {
+ ByteBufferAccessor accessor = new ByteBufferAccessor(record.value());
+ try {
+ ApiMessageAndVersion messageAndVersion = serde.read(accessor, record.valueSize());
+ messages.add(messageAndVersion);
+ } catch (Throwable e) {
+ log.error("unable to read metadata record at offset {}", record.offset(), e);
Review Comment:
good catch. this should not be ignored.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949687350
##########
metadata/src/main/java/org/apache/kafka/controller/LogReplayTracker.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.controller;
+
+import java.util.Optional;
+
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.common.utils.LogContext;
+import org.apache.kafka.metadata.bootstrap.BootstrapMetadata;
+import org.apache.kafka.server.common.MetadataVersion;
+import org.slf4j.Logger;
+
+
+public class LogReplayTracker {
+ public static class Builder {
+ private LogContext logContext = null;
+
+ Builder setLogContext(LogContext logContext) {
+ this.logContext = logContext;
+ return this;
+ }
+
+ public LogReplayTracker build() {
+ if (logContext == null) logContext = new LogContext();
+ return new LogReplayTracker(logContext);
+ }
+ }
+
+ /**
+ * The slf4j logger.
+ */
+ private final Logger log;
+
+ /**
+ * True if we haven't replayed any records yet.
+ */
+ private boolean empty;
+
+ /**
+ * True if we haven't replayed any metadata.version records yet.
+ */
+ private boolean versionless;
Review Comment:
I usually don't push back on name suggestions but I really liked "versionless" ... it's a log without a version. Seems apt?
Same for "empty" -- it just feels right.
Of course these things are mutable, will change over time, but as a description of the current state it seems correct?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949676920
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+
+/**
+ * The bootstrap metadata. On startup, if the metadata log is empty, we will populate the log with
+ * these records. Alternately, if log is not empty, but the metadata version is not set, we will
+ * use the version specified here.
+ */
+public class BootstrapMetadata {
+ private final List<ApiMessageAndVersion> records;
+ private final MetadataVersion metadataVersion;
+ private final String source;
+
+ public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
+ List<ApiMessageAndVersion> records = Collections.singletonList(
+ new ApiMessageAndVersion(new FeatureLevelRecord().
+ setName(MetadataVersion.FEATURE_NAME).
+ setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
Review Comment:
This is an issue which is not really specific to `bootstrap.checkpoint`... it's kind of similar to should we bump the version of `ApiVersionRequest`?
I do think the `FeatureLevelRecord` which contains the initial metadata version should always remain at version 0. Otherwise old Kafkas won't be able to determine the version of newer metadata logs. (They wouldn't have been able to read them in either case, but if they can't even read the first record, the error message is bad)
I guess this seemed obvious to me, but we should probably add a comment to `FeatureLevelRecord.json` telling people not to bump the version. I will add one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949685089
##########
metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java:
##########
@@ -188,12 +182,16 @@ private ApiError updateFeature(
}
}
- if (currentVersion != null && newVersion < currentVersion) {
+ if (newVersion < currentVersion) {
if (upgradeType.equals(FeatureUpdate.UpgradeType.UPGRADE)) {
Review Comment:
We do check it in `updateFeature`, right?
```
if (upgradeType.equals(FeatureUpdate.UpgradeType.UNKNOWN)) {
return invalidUpdateVersion(featureName, newVersion,
"The controller does not support the given upgrade type.");
}
```
I do kind of wish we did less in `QuorumController#updateFeatures` and more in the `FeatureControlManager` (because it's easier to unit test the latter) but this particular case is covered, I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji merged pull request #12513: KAFKA-14177: Correctly support older kraft versions without FeatureLevelRecord
Posted by GitBox <gi...@apache.org>.
hachikuji merged PR #12513:
URL: https://github.com/apache/kafka/pull/12513
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: KAFKA-14177: Correctly support older kraft versions without FeatureLevelRecord
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r952893463
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
Review Comment:
I will just make this an `Optional<MetadataVersion>` for simplicity.
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
Review Comment:
let's leave this as-is for now and we can refactor later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949693026
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -222,52 +227,130 @@ public void testFeatureControlIterator() throws Exception {
manager.iterator(Long.MAX_VALUE));
}
+ private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
+ new FeatureControlManager.Builder().
+ setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
+ MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+
@Test
public void testApplyMetadataVersionChangeRecord() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).build();
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_0_IV1.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_0_IV1, manager.metadataVersion());
+ setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
+ assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
}
@Test
- public void testDowngradeMetadataVersion() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_2_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
- build();
- assertEquals(manager.metadataVersion(), MetadataVersion.IBP_3_3_IV0);
+ public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ assertEquals(ControllerResult.of(Collections.emptyList(),
+ singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
Review Comment:
Hmm... I definitely disagree about INVALID_REQUEST. That error code is supposed to be a catch-all for requests that are malformed. I don't think we should be using it in APIs to indicate that the user messed up. In theory a well-programmed client should not send invalid requests.
I'm used to the way errno works in C, where there's a small number of error codes and they're heavily overloaded. For example, a bunch of APIs return ENOENT if the entity they're looking for doesn't exist, or EEXIST if it does and that's a problem.
I sort of look askance at Kafka's approach. Like did we really need a separate error code for DELEGATION_TOKEN_NOT_FOUND, GROUP_ID_NOT_FOUND, and RESOURCE_NOT_FOUND? Or could we just have a NOT_FOUND error code and return it where appropriate. It's not like you are going to get back DELEGATION_TOKEN_NOT_FOUND from an API that deals with groups, so what's the issue? You are not adding more descriptive power by having an X_NOT_FOUND, Y_NOT_FOUND, Z_NOT_FOUND If only one of them can be used in any given scenario.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949694015
##########
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java:
##########
@@ -952,6 +966,7 @@ public void testEarlyControllerResults() throws Throwable {
}
}
+ @Disabled // TODO: need to fix leader election in LocalLog.
Review Comment:
It's not flaky, but this PR did break it.
This PR exposes a bug in `LocalLogManager`, which is that it doesn't really do leader election properly. I started to fix that bug, but then I ended up rewriting big parts of LocalLogManager, which doesn't seem appropriate for 3.3, so I shelved it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949674532
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapMetadata.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.common.metadata.FeatureLevelRecord;
+import org.apache.kafka.common.protocol.ApiMessage;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+
+/**
+ * The bootstrap metadata. On startup, if the metadata log is empty, we will populate the log with
+ * these records. Alternately, if log is not empty, but the metadata version is not set, we will
+ * use the version specified here.
+ */
+public class BootstrapMetadata {
+ private final List<ApiMessageAndVersion> records;
+ private final MetadataVersion metadataVersion;
+ private final String source;
+
+ public static BootstrapMetadata fromVersion(MetadataVersion metadataVersion, String source) {
+ List<ApiMessageAndVersion> records = Collections.singletonList(
+ new ApiMessageAndVersion(new FeatureLevelRecord().
+ setName(MetadataVersion.FEATURE_NAME).
+ setFeatureLevel(metadataVersion.featureLevel()), (short) 0));
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static BootstrapMetadata fromRecords(List<ApiMessageAndVersion> records, String source) {
+ MetadataVersion metadataVersion = null;
+ for (ApiMessageAndVersion record : records) {
+ Optional<MetadataVersion> version = recordToMetadataVersion(record.message());
+ if (version.isPresent()) {
+ metadataVersion = version.get();
+ }
+ }
+ if (metadataVersion == null) {
+ throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
+ " was found in the bootstrap metadata from " + source);
+ }
+ return new BootstrapMetadata(records, metadataVersion, source);
+ }
+
+ public static Optional<MetadataVersion> recordToMetadataVersion(ApiMessage record) {
+ if (record instanceof FeatureLevelRecord) {
+ FeatureLevelRecord featureLevel = (FeatureLevelRecord) record;
+ if (featureLevel.name().equals(MetadataVersion.FEATURE_NAME)) {
+ return Optional.of(MetadataVersion.fromFeatureLevel(featureLevel.featureLevel()));
+ }
+ }
+ return Optional.empty();
+ }
+
+ BootstrapMetadata(
+ List<ApiMessageAndVersion> records,
+ MetadataVersion metadataVersion,
+ String source
+ ) {
+ Objects.requireNonNull(records);
+ this.records = records;
+ if (metadataVersion.isLessThan(MetadataVersion.MINIMUM_KRAFT_VERSION)) {
+ throw new RuntimeException("Unable to load metadata from " + source + ": bootstrap " +
+ "metadata versions less than " + MetadataVersion.MINIMUM_KRAFT_VERSION + " are " +
+ "not supported.");
+ }
+ this.metadataVersion = metadataVersion;
+ Objects.requireNonNull(source);
+ this.source = source;
+ }
+
+ public List<ApiMessageAndVersion> records() {
+ return records;
+ }
+
+ public MetadataVersion metadataVersion() {
+ return metadataVersion;
+ }
+
+ public String source() {
+ return source;
+ }
+
+ public BootstrapMetadata copyWithOnlyVersion() {
+ List<ApiMessageAndVersion> newRecords = new ArrayList<>();
+ for (ApiMessageAndVersion record : records) {
+ if (recordToMetadataVersion(record.message()).isPresent()) {
+ newRecords.clear();
+ newRecords.add(record);
+ }
+ }
+ if (newRecords.isEmpty()) {
+ throw new RuntimeException("No FeatureLevelRecord for " + MetadataVersion.FEATURE_NAME +
+ " was found in the bootstrap metadata from " + source);
+ }
+ return new BootstrapMetadata(Collections.unmodifiableList(newRecords),
+ metadataVersion, source);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(records, metadataVersion, source);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o == null || !(o instanceof BootstrapMetadata)) return false;
Review Comment:
oh, I wanted to write `!o.getClass().equals(this.getClass()))`. Let me do that.
Sigh. Record types can't come soon enough...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a diff in pull request #12513: KAFKA-14177: Correctly support older kraft versions without FeatureLevelRecord
Posted by GitBox <gi...@apache.org>.
hachikuji commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r954300741
##########
core/src/test/scala/unit/kafka/tools/StorageToolTest.scala:
##########
@@ -191,18 +191,26 @@ Found problem:
@Test
def testDefaultMetadataVersion(): Unit = {
val namespace = StorageTool.parseArguments(Array("format", "-c", "config.props", "-t", "XcZZOzUqS4yHOjhMQB6JLQ"))
- val mv = StorageTool.getMetadataVersion(namespace)
+ val mv = StorageTool.getMetadataVersion(namespace, "")
assertEquals(MetadataVersion.latest().featureLevel(), mv.featureLevel(),
"Expected the default metadata.version to be the latest version")
}
+ @Test
+ def testConfiguredMetadataVersion(): Unit = {
+ val namespace = StorageTool.parseArguments(Array("format", "-c", "config.props", "-t", "XcZZOzUqS4yHOjhMQB6JLQ"))
+ val mv = StorageTool.getMetadataVersion(namespace, MetadataVersion.IBP_3_3_IV2.toString)
+ assertEquals(MetadataVersion.IBP_3_3_IV2.featureLevel(), mv.featureLevel(),
+ "Expected the default metadata.version to be 3.3-IV2")
+ }
+
@Test
def testMetadataVersionFlags(): Unit = {
def parseMetadataVersion(strings: String*): MetadataVersion = {
var args = mutable.Seq("format", "-c", "config.props", "-t", "XcZZOzUqS4yHOjhMQB6JLQ")
args ++= strings
val namespace = StorageTool.parseArguments(args.toArray)
- StorageTool.getMetadataVersion(namespace)
+ StorageTool.getMetadataVersion(namespace, "")
Review Comment:
This test fails because `""` is not a valid IBP. I think this gets back to my point about the dubious use of `""` as a sentinel. The method here expects `null` instead. I think we should try to be consistent.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949666958
##########
metadata/src/main/java/org/apache/kafka/metadata/bootstrap/BootstrapDirectory.java:
##########
@@ -0,0 +1,134 @@
+/*
+ * 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.metadata.bootstrap;
+
+import org.apache.kafka.metadata.util.BatchFileReader;
+import org.apache.kafka.metadata.util.BatchFileReader.BatchAndType;
+import org.apache.kafka.metadata.util.BatchFileWriter;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.apache.kafka.server.common.MetadataVersion;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static org.apache.kafka.server.common.MetadataVersion.MINIMUM_KRAFT_VERSION;
+
+
+/**
+ * A read-only class that holds the controller bootstrap metadata. A file named "bootstrap.snapshot" is used and the
+ * format is the same as a KRaft snapshot.
+ */
+public class BootstrapDirectory {
+ final static String INTER_BROKER_PROTOCOL_CONFIG_KEY = "inter.broker.protocol.version";
+ final static String BINARY_BOOTSTRAP = "bootstrap.checkpoint";
+
+ public static String ibpStringFromConfigMap(Map<String, Object> staticConfig) {
+ Object value = staticConfig.get(INTER_BROKER_PROTOCOL_CONFIG_KEY);
+ return value == null ? "" : value.toString();
+ }
+
+ private final String directoryPath;
+ private final String ibp;
+
+ /**
+ * Create a new BootstrapDirectory object.
+ *
+ * @param directoryPath The path to the directory with the bootstrap file.
+ * @param ibp The configured value of inter.broker.protocol, or the empty string
+ * if it is not configured.
+ */
+ public BootstrapDirectory(
+ String directoryPath,
+ String ibp
+ ) {
+ Objects.requireNonNull(directoryPath);
+ Objects.requireNonNull(ibp);
+ this.directoryPath = directoryPath;
+ this.ibp = ibp;
+ }
+
+ public BootstrapMetadata read() throws Exception {
+ if (!Files.isDirectory(Paths.get(directoryPath))) {
+ throw new RuntimeException("No such directory as " + directoryPath);
+ }
+ Path binaryBootstrapPath = Paths.get(directoryPath, BINARY_BOOTSTRAP);
+ if (!Files.exists(binaryBootstrapPath)) {
+ return readFromConfiguration();
+ } else {
+ return readFromBinaryFile(binaryBootstrapPath.toString());
+ }
+ }
+
+ BootstrapMetadata readFromConfiguration() {
+ if (ibp.isEmpty()) {
+ return BootstrapMetadata.fromVersion(MetadataVersion.latest(),
+ "the default bootstrap file which sets the latest metadata.version, " +
+ "since no bootstrap file was found, and " + INTER_BROKER_PROTOCOL_CONFIG_KEY +
+ " was not configured.");
+ }
+ MetadataVersion version = MetadataVersion.fromVersionString(ibp);
+ if (version.isLessThan(MINIMUM_KRAFT_VERSION)) {
+ return BootstrapMetadata.fromVersion(MINIMUM_KRAFT_VERSION,
+ "a default bootstrap file setting the minimum supported KRaft metadata " +
Review Comment:
Well, this is an exception that will terminate the startup process. Making it concise is probably not what we want.
"source" is also used in logging in the case where startup succeeded. It's useful to know things like "you configured X, but actually you got Y, because of these reasons" or "you didn't configure anything, so you got X, have a nice day"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949679883
##########
metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileReader.java:
##########
@@ -0,0 +1,183 @@
+/*
+ * 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.metadata.util;
+
+import org.apache.kafka.common.message.LeaderChangeMessage;
+import org.apache.kafka.common.message.SnapshotFooterRecord;
+import org.apache.kafka.common.message.SnapshotHeaderRecord;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
+import org.apache.kafka.common.record.ControlRecordType;
+import org.apache.kafka.common.record.FileLogInputStream.FileChannelRecordBatch;
+import org.apache.kafka.common.record.FileRecords;
+import org.apache.kafka.common.record.Record;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.metadata.MetadataRecordSerde;
+import org.apache.kafka.raft.Batch;
+import org.apache.kafka.server.common.ApiMessageAndVersion;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+
+
+/**
+ * Reads a log file containing KRaft record batches.
+ */
+public final class BatchFileReader implements Iterator<BatchFileReader.BatchAndType>, AutoCloseable {
+ private static final Logger log = LoggerFactory.getLogger(BatchFileReader.class);
+
+ public static class Builder {
+ private String path = null;
+
+ public Builder setPath(String path) {
+ this.path = path;
+ return this;
+ }
+
+ public BatchFileReader build() throws Exception {
+ if (path == null) {
+ throw new RuntimeException("You must specify a path.");
+ }
+ FileRecords fileRecords = FileRecords.open(new File(path), false);
+ try {
+ return new BatchFileReader(fileRecords);
+ } catch (Throwable e) {
+ Utils.closeQuietly(fileRecords, "fileRecords");
+ throw e;
+ }
+ }
+ }
+
+ public static class BatchAndType {
+ private final Batch<ApiMessageAndVersion> batch;
+ private final boolean isControl;
+
+ public BatchAndType(Batch<ApiMessageAndVersion> batch, boolean isControl) {
+ this.batch = batch;
+ this.isControl = isControl;
+ }
+
+ public Batch<ApiMessageAndVersion> batch() {
+ return batch;
+ }
+
+ public boolean isControl() {
+ return isControl;
+ }
+ }
+
+ private final FileRecords fileRecords;
+ private Iterator<FileChannelRecordBatch> batchIterator;
+ private final MetadataRecordSerde serde;
+
+ private BatchFileReader(FileRecords fileRecords) {
+ this.fileRecords = fileRecords;
+ this.batchIterator = fileRecords.batchIterator();
+ this.serde = new MetadataRecordSerde();
+ }
+
+ @Override
+ public boolean hasNext() {
+ return this.batchIterator.hasNext();
+ }
+
+ @Override
+ public BatchAndType next() {
+ FileChannelRecordBatch input = batchIterator.next();
+ if (input.isControlBatch()) {
+ return nextControlBatch(input);
+ } else {
+ return nextMetadataBatch(input);
+ }
+ }
+
+ private BatchAndType nextControlBatch(FileChannelRecordBatch input) {
+ List<ApiMessageAndVersion> messages = new ArrayList<>();
+ for (Iterator<Record> iter = input.iterator(); iter.hasNext(); ) {
+ Record record = iter.next();
+ try {
+ short typeId = ControlRecordType.parseTypeId(record.key());
+ ControlRecordType type = ControlRecordType.fromTypeId(typeId);
+ switch (type) {
+ case LEADER_CHANGE: {
+ LeaderChangeMessage message = new LeaderChangeMessage();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ case SNAPSHOT_HEADER: {
+ SnapshotHeaderRecord message = new SnapshotHeaderRecord();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ case SNAPSHOT_FOOTER: {
+ SnapshotFooterRecord message = new SnapshotFooterRecord();
+ message.read(new ByteBufferAccessor(record.value()), (short) 0);
+ messages.add(new ApiMessageAndVersion(message, (short) 0));
+ break;
+ }
+ default:
+ throw new RuntimeException("Unsupported control record type " + type + " at offset " +
+ record.offset());
+ }
+ } catch (Throwable e) {
+ throw new RuntimeException("Unable to read control record at offset " + record.offset(), e);
+ }
+ }
+ return new BatchAndType(Batch.data(
+ input.baseOffset(),
+ input.partitionLeaderEpoch(),
+ input.maxTimestamp(),
+ input.sizeInBytes(),
+ messages), true);
+ }
+
+ private BatchAndType nextMetadataBatch(FileChannelRecordBatch input) {
+ List<ApiMessageAndVersion> messages = new ArrayList<>();
+ for (Record record : input) {
+ ByteBufferAccessor accessor = new ByteBufferAccessor(record.value());
+ try {
+ ApiMessageAndVersion messageAndVersion = serde.read(accessor, record.valueSize());
+ messages.add(messageAndVersion);
+ } catch (Throwable e) {
+ log.error("unable to read metadata record at offset {}", record.offset(), e);
+ }
+ }
+ return new BatchAndType(Batch.data(
+ input.baseOffset(),
+ input.partitionLeaderEpoch(),
+ input.maxTimestamp(),
+ input.sizeInBytes(),
+ messages), false);
+ }
+
+ @Override
+ public void close() {
+ try {
+ fileRecords.closeHandlers();
Review Comment:
`FileRecords#close` tries to do a flush and trim, which fails because the file descriptor was opened read-only.
`FileRecords#close` probably should be fixed to remove the assumption that all files are writable, but I didn't want to expand the scope of this change even more.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949687749
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -1142,15 +1142,19 @@ private void renounce() {
purgatory.failAll(newNotControllerException());
if (snapshotRegistry.hasSnapshot(lastCommittedOffset)) {
- newBytesSinceLastSnapshot = 0;
snapshotRegistry.revertToSnapshot(lastCommittedOffset);
authorizer.ifPresent(a -> a.loadSnapshot(aclControlManager.idToAcl()));
} else {
- resetState();
+ log.warn("Unable to find last committed offset {} in snapshot registry; resetting " +
Review Comment:
it is expected, unfortunately. we have some ideas for fixing this, but it's outside the scope of this PR.
I guess you're hinting that this should be an INFO? :) I'll make it an INFO.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949693026
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -222,52 +227,130 @@ public void testFeatureControlIterator() throws Exception {
manager.iterator(Long.MAX_VALUE));
}
+ private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
+ new FeatureControlManager.Builder().
+ setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
+ MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+
@Test
public void testApplyMetadataVersionChangeRecord() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).build();
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_0_IV1.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_0_IV1, manager.metadataVersion());
+ setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
+ assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
}
@Test
- public void testDowngradeMetadataVersion() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_2_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
- build();
- assertEquals(manager.metadataVersion(), MetadataVersion.IBP_3_3_IV0);
+ public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ assertEquals(ControllerResult.of(Collections.emptyList(),
+ singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
Review Comment:
Hmm... I definitely disagree about INVALID_REQUEST. That error code is supposed to be a catch-all for requests that are malformed. I don't think we should be using it in APIs to indicate that the user messed up. In theory a well-programmed client should not send invalid requests.
I'm used to the way errno works in C, where there's a small number of error codes and they're heavily overloaded. For example, a bunch of APIs return ENOENT if the entity they're looking for doesn't exist, or EEXIST if it does and that's a problem.
I sort of look askance at Kafka's approach. Like did we really need a separate error code for DELEGATION_TOKEN_NOT_FOUND, GROUP_ID_NOT_FOUND, and RESOURCE_NOT_FOUND? Or could we just have a NOT_FOUND error code and return it where appropriate. It's not like you are going to get back DELEGATION_TOKEN_NOT_FOUND from an API that deals with groups, so what's the issue?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949693455
##########
metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java:
##########
@@ -222,52 +227,130 @@ public void testFeatureControlIterator() throws Exception {
manager.iterator(Long.MAX_VALUE));
}
+ private static final FeatureControlManager.Builder TEST_MANAGER_BUILDER1 =
+ new FeatureControlManager.Builder().
+ setQuorumFeatures(features(MetadataVersion.FEATURE_NAME,
+ MetadataVersion.IBP_3_3_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV3.featureLevel())).
+ setMetadataVersion(MetadataVersion.IBP_3_3_IV2);
+
@Test
public void testApplyMetadataVersionChangeRecord() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_0_IV1.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).build();
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
manager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
- setFeatureLevel(MetadataVersion.IBP_3_0_IV1.featureLevel()));
- assertEquals(MetadataVersion.IBP_3_0_IV1, manager.metadataVersion());
+ setFeatureLevel(MetadataVersion.IBP_3_3_IV3.featureLevel()));
+ assertEquals(MetadataVersion.IBP_3_3_IV3, manager.metadataVersion());
}
@Test
- public void testDowngradeMetadataVersion() {
- QuorumFeatures features = features(MetadataVersion.FEATURE_NAME,
- MetadataVersion.IBP_3_2_IV0.featureLevel(), MetadataVersion.IBP_3_3_IV0.featureLevel());
- FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(features).
- setMetadataVersion(MetadataVersion.IBP_3_3_IV0).
- build();
- assertEquals(manager.metadataVersion(), MetadataVersion.IBP_3_3_IV0);
+ public void testCannotDowngradeToVersionBeforeMinimumSupportedKraftVersion() {
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ assertEquals(ControllerResult.of(Collections.emptyList(),
+ singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
+ "Invalid update version 3 for feature metadata.version. Local controller 0 only " +
+ "supports versions 4-7"))),
+ manager.updateFeatures(
+ singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_2_IV0.featureLevel()),
+ singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.UNSAFE_DOWNGRADE),
+ emptyMap(),
+ true));
+ }
+
+ @Test
+ public void testCannotDowngradeToHigherVersion() {
+ FeatureControlManager manager = TEST_MANAGER_BUILDER1.build();
+ assertEquals(ControllerResult.of(Collections.emptyList(),
+ singletonMap(MetadataVersion.FEATURE_NAME, new ApiError(Errors.INVALID_UPDATE_VERSION,
+ "Invalid update version 7 for feature metadata.version. Can't downgrade to a " +
+ "newer version."))),
+ manager.updateFeatures(
+ singletonMap(MetadataVersion.FEATURE_NAME, MetadataVersion.IBP_3_3_IV3.featureLevel()),
+ singletonMap(MetadataVersion.FEATURE_NAME, FeatureUpdate.UpgradeType.SAFE_DOWNGRADE),
Review Comment:
yes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] cmccabe commented on a diff in pull request #12513: MINOR: some fixes to the KRaft bootstrap logic
Posted by GitBox <gi...@apache.org>.
cmccabe commented on code in PR #12513:
URL: https://github.com/apache/kafka/pull/12513#discussion_r949694015
##########
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java:
##########
@@ -952,6 +966,7 @@ public void testEarlyControllerResults() throws Throwable {
}
}
+ @Disabled // TODO: need to fix leader election in LocalLog.
Review Comment:
This PR exposes a bug in `LocalLogManager`, which is that it doesn't really do leader election properly. I started to fix that bug, but then I ended up rewriting big parts of LocalLogManager, which doesn't seem appropriate for 3.3, so I shelved it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org