mumrah commented on code in PR #13643:
URL: https://github.com/apache/kafka/pull/13643#discussion_r1264405122
##########
metadata/src/main/java/org/apache/kafka/controller/QuorumController.java:
##########
@@ -740,22 +739,27 @@ public Long apply(List<ApiMessageAndVersion> records) {
// Start by trying to apply the record to our
in-memory state. This should always
// succeed; if it does not, that's a fatal error.
It is important to do this before
// scheduling the record for Raft replication.
- int i = 1;
+ int recordIndex = 0;
for (ApiMessageAndVersion message : records) {
+ long recordOffset = prevEndOffset + 1 +
recordIndex;
try {
- replay(message.message(),
Optional.empty(), prevEndOffset + records.size());
+ replay(message.message(),
Optional.empty(), recordOffset);
} catch (Throwable e) {
- String failureMessage =
String.format("Unable to apply %s record, which was " +
- "%d of %d record(s) in the batch
following last write offset %d.",
-
message.message().getClass().getSimpleName(), i, records.size(),
- prevEndOffset);
+ String failureMessage =
String.format("Unable to apply %s " +
+ "record at offset %d on active
controller, from the " +
+ "batch with baseOffset %d",
+
message.message().getClass().getSimpleName(),
+ recordOffset, prevEndOffset + 1);
throw
fatalFaultHandler.handleFault(failureMessage, e);
}
- i++;
+ recordIndex++;
}
- prevEndOffset =
raftClient.scheduleAtomicAppend(controllerEpoch, records);
-
snapshotRegistry.getOrCreateSnapshot(prevEndOffset);
- return prevEndOffset;
+ long nextEndOffset = prevEndOffset + recordIndex;
+ raftClient.scheduleAtomicAppend(controllerEpoch,
OptionalLong.of(nextEndOffset), records);
+
snapshotRegistry.getOrCreateSnapshot(nextEndOffset);
+
snapshotRegistry.getOrCreateSnapshot(nextEndOffset);
Review Comment:
is this duplicate line intentional?
--
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]
