hachikuji commented on a change in pull request #10480:
URL: https://github.com/apache/kafka/pull/10480#discussion_r608140887
##########
File path: raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
##########
@@ -395,31 +393,35 @@ private void resetConnections() {
requestManager.resetAll();
}
- private void onBecomeLeader(long currentTimeMs) {
- LeaderState state = quorum.leaderStateOrThrow();
+ private void onBecomeLeader(long currentTimeMs) throws IOException {
+ long endOffset = log.endOffset().offset;
+
+ // Add 1 to the offset that the accumulator tracks since
appendLeaderChangeMessage
+ // will write a record from the new leader's epoch to advance the high
watermark below
+ BatchAccumulator<T> accumulator = new BatchAccumulator<>(
+ quorum.epoch(),
+ endOffset + 1,
Review comment:
Hmm.. This is a little strange. It would be nice if we could refactor
this so that we can write the LeaderChange message through LeaderState. This is
a little bit tricky, but it should be doable. Let me try to put this together.
--
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
