hachikuji commented on code in PR #12508:
URL: https://github.com/apache/kafka/pull/12508#discussion_r947308245
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -246,20 +264,30 @@ private List<ReplicaState>
followersByDescendingFetchOffset() {
.collect(Collectors.toList());
}
- private boolean updateEndOffset(ReplicaState state,
- LogOffsetMetadata endOffsetMetadata) {
+ private void verifyEndOffsetUpdate(
+ ReplicaState state,
+ LogOffsetMetadata endOffsetMetadata
+ ) {
state.endOffset.ifPresent(currentEndOffset -> {
if (currentEndOffset.offset > endOffsetMetadata.offset) {
if (state.nodeId == localId) {
throw new IllegalStateException("Detected non-monotonic
update of local " +
- "end offset: " + currentEndOffset.offset + " -> " +
endOffsetMetadata.offset);
+ "end offset: " + currentEndOffset.offset + " -> "
+ endOffsetMetadata.offset);
} else {
log.warn("Detected non-monotonic update of fetch offset
from nodeId {}: {} -> {}",
- state.nodeId, currentEndOffset.offset,
endOffsetMetadata.offset);
+ state.nodeId, currentEndOffset.offset,
endOffsetMetadata.offset);
}
}
});
-
+ }
+ private boolean updateEndOffset(
+ ReplicaState state,
+ LogOffsetMetadata endOffsetMetadata,
+ boolean verifyUpdate
Review Comment:
Why do we need this? Seems like `false` is the only value we ever pass.
##########
core/src/test/scala/integration/kafka/server/KRaftClusterTest.scala:
##########
@@ -804,28 +804,24 @@ class KRaftClusterTest {
val quorumState = admin.describeMetadataQuorum(new
DescribeMetadataQuorumOptions)
val quorumInfo = quorumState.quorumInfo.get()
- assertEquals(4, quorumInfo.observers.size)
- assertEquals(3, quorumInfo.voters.size)
+ assertEquals(cluster.controllers.asScala.keySet,
quorumInfo.voters.asScala.map(_.replicaId).toSet)
assertTrue(2999 < quorumInfo.leaderId && 3003 > quorumInfo.leaderId,
Review Comment:
We could make this assertion a little more generic by asserting that
`quorumInfo.leaderId` is contained in `cluster.controllers.asScala.keySet`.
--
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
