mumrah commented on code in PR #12050:
URL: https://github.com/apache/kafka/pull/12050#discussion_r851351864
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -883,15 +905,49 @@ public void handleLeaderChange(LeaderAndEpoch newLeader) {
curEpoch);
}
log.info(
- "Becoming the active controller at epoch {}, committed
offset {} and committed epoch {}.",
- newEpoch, lastCommittedOffset, lastCommittedEpoch
+ "Becoming the active controller at epoch {}, committed
offset {}, committed epoch {}, and metadata.version {}",
+ newEpoch, lastCommittedOffset, lastCommittedEpoch,
featureControl.metadataVersion()
);
curClaimEpoch = newEpoch;
controllerMetrics.setActive(true);
writeOffset = lastCommittedOffset;
clusterControl.activate();
+ // Check if we need to bootstrap a metadata.version into
the log. This must happen before we can
+ // write any records to the log since we need the
metadata.version to determine the correct
+ // record version
+
+ if (featureControl.metadataVersion() ==
MetadataVersion.UNINITIALIZED) {
+ final CompletableFuture<Map<String, ApiError>> future;
+ if (initialMetadataVersion ==
MetadataVersion.UNINITIALIZED) {
+ future = new CompletableFuture<>();
+ future.completeExceptionally(
+ new IllegalStateException("Cannot become
leader without an initial metadata.version to use."));
+ } else if (initialMetadataVersion ==
MetadataVersion.V1) {
+ future =
appendWriteEvent("initializeMetadataVersion", () -> {
+ log.info("Upgrading from KRaft preview.
Initializing metadata.version to 1");
+ return
featureControl.initializeMetadataVersion(MetadataVersion.V1.version());
+ });
+ } else {
+ future =
appendWriteEvent("initializeMetadataVersion", () -> {
+ log.info("Initializing metadata.version to
{}", initialMetadataVersion.version());
+ return
featureControl.initializeMetadataVersion(initialMetadataVersion.version());
+ });
+ }
+ future.whenComplete((result, exception) -> {
+ if (exception != null) {
+ log.error("Failed to initialize
metadata.version", exception);
+ appendRaftEvent("metadataVersionFailure[" +
curClaimEpoch + "]", () -> {
+ log.warn("Renouncing the leadership at
oldEpoch {} since we could not bootstrap" +
+ "a metadata.version. Reverting to
last committed offset {}.",
+ curClaimEpoch, lastCommittedOffset);
+ renounce();
Review Comment:
@cmccabe is this the correct way to renounce leadership? Originally, I was
calling renounce directly here, but that led to a
ConcurrentModificationException
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
