junrao commented on code in PR #18277:
URL: https://github.com/apache/kafka/pull/18277#discussion_r1934835870
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -432,23 +443,29 @@ public ControllerResult<BrokerRegistrationReply>
registerBroker(
"epoch is {}.", brokerId, request.incarnationId(),
numRecordsAdded, newBrokerEpoch);
} else {
log.info("Registering a new incarnation of broker {}. Previous
incarnation ID " +
- "was {}; new incarnation ID is {}. Generated {}
record(s) to clean up " +
+ "was {}; new incarnation ID is {}. {}Generated {}
record(s) to clean up " +
"previous incarnations. Broker epoch will become {}.",
brokerId,
- existing.incarnationId(), request.incarnationId(),
numRecordsAdded,
+ existing.incarnationId(), request.incarnationId(),
+ uncleanShutdownDetectionEnabled ?
+ "Using unclean shutdown detection. Stored broker
epoch=" + storedBrokerEpoch +
+ " vs request broker epoch=" +
request.previousBrokerEpoch() + ". "
+ : "",
+ numRecordsAdded,
newBrokerEpoch);
}
record.setBrokerEpoch(newBrokerEpoch);
} else {
log.info("Amending registration of broker {}, incarnation ID {}.
Broker epoch remains {}.",
request.brokerId(), request.incarnationId(),
existing.epoch());
record.setFenced(existing.fenced());
- record.setInControlledShutdown(existing.inControlledShutdown());
+ record.setInControlledShutdown(isBrokerStartUp ? false :
existing.inControlledShutdown());
record.setBrokerEpoch(existing.epoch());
+ record.setIncarnationId(request.incarnationId());
}
records.add(new ApiMessageAndVersion(record,
featureControl.metadataVersion().
registerBrokerRecordVersion()));
- if (!request.incarnationId().equals(prevIncarnationId)) {
+ if (isUncleanShutdownRegistration) {
Review Comment:
Hmm, is this change correct? We should clear the heartbeat as long as the
incarnationId is different, right?
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -422,7 +425,15 @@ public ControllerResult<BrokerRegistrationReply>
registerBroker(
record.setLogDirs(request.logDirs());
}
- if (!request.incarnationId().equals(prevIncarnationId)) {
+ boolean isUncleanShutdownRegistration;
+ if (uncleanShutdownDetectionEnabled) {
+ isUncleanShutdownRegistration = storedBrokerEpoch !=
request.previousBrokerEpoch();
+ } else {
+ isUncleanShutdownRegistration =
!request.incarnationId().equals(prevIncarnationId);
+ }
+ boolean isBrokerStartUp =
!request.incarnationId().equals(prevIncarnationId);
+
+ if (isUncleanShutdownRegistration) {
Review Comment:
Hmm, is this correct? If the incarnationId is different, we always need to
generate new PartitionChangeRecords by removing the brokerId from ISR. It seems
that we are skipping that step if the previous shutdown is clean in ELR mode.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
