jsancio commented on code in PR #16079:
URL: https://github.com/apache/kafka/pull/16079#discussion_r1616359378
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -243,8 +243,9 @@ private boolean maybeUpdateHighWatermark() {
);
return true;
} else if (highWatermarkUpdateOffset <
currentHighWatermarkMetadata.offset) {
- log.error("The latest computed high watermark {} is
smaller than the current " +
- "value {}, which suggests that one of the
voters has lost committed data. " +
+ log.warn("The latest computed high watermark {} is
smaller than the current " +
+ "value {}, which should only happen when voter
set membership changes. If the voter " +
+ "set has not changed this suggests that one of
the voters has lost committed data. " +
Review Comment:
In general we should avoid having `error` or `warn` log messages. In most
cases we want to throw an exception if there is an error.
In this case, this state is expected but infrequent so `log.info` seems
appropriate.
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -445,6 +453,15 @@ private boolean isVoter(int remoteNodeId) {
return voterStates.containsKey(remoteNodeId);
}
+ // for testing purposes
+ boolean removeVoter(int nodeId) {
+ if (voterStates.containsKey(nodeId)) {
+ voterStates.remove(nodeId);
+ return true;
+ }
+ return false;
+ }
Review Comment:
I think we should remove this method. We should test and use the same
functionality that will be used by `KafkaRaftClient`. I don't think this is how
voter changes are going to get communicated to the replica state
(`LeaderState`).
I am thinking that we should replace the `Set<Integer> voters` parameter in
the constructor with `Supplier<VoterSet> latestVoterSet`. Every time the
`partitionState` gets updated, we should compare the current `LeaderState`
against the latest voter set and update the `LeaderState` accordingly.
##########
raft/src/main/java/org/apache/kafka/raft/LeaderState.java:
##########
@@ -341,9 +342,16 @@ public boolean updateReplicaState(
state.nodeId, currentEndOffset.offset,
fetchOffsetMetadata.offset);
}
});
-
- Optional<LogOffsetMetadata> leaderEndOffsetOpt =
- voterStates.get(localId).endOffset;
+ Optional<LogOffsetMetadata> leaderEndOffsetOpt;
+ if (voterStates.containsKey(localId)) {
+ leaderEndOffsetOpt = voterStates.get(localId).endOffset;
Review Comment:
Let's avoid the pattern `if (map.containsKey(key)) { T value = map.get(key);
... }` This is the same as:
```java
T value = map.get(key);
if (value != null) {
...
}
// Or
Optional.OfNullable(map.get(key)).ifPresent(value -> ...);
```
--
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
