josefk31 commented on code in PR #22620:
URL: https://github.com/apache/kafka/pull/22620#discussion_r3468616480
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
Review Comment:
I think we also need to update this comment with a new step after step 5?
##########
raft/src/main/java/org/apache/kafka/raft/internals/ChangeVoterHandlerState.java:
##########
@@ -168,41 +240,82 @@ public long maybeExpirePendingOperation(long
currentTimeMs) {
resetRemoveVoterHandlerState(Errors.REQUEST_TIMED_OUT, null,
Optional.empty());
}
+ long timeUntilUpdateVoterExpiration = updateVoterHandlerState()
+ .map(state -> state.timeUntilOperationExpiration(currentTimeMs))
+ .orElse(Long.MAX_VALUE);
+
+ if (timeUntilUpdateVoterExpiration == 0) {
+ resetUpdateVoterHandlerState(
+ Errors.REQUEST_TIMED_OUT,
+ leaderAndEpoch,
+ leaderEndpoints,
+ Optional.empty()
+ );
+ }
+
// Reread the timeouts and return the smaller of them
return Math.min(
addVoterHandlerState()
.map(state ->
state.timeUntilOperationExpiration(currentTimeMs))
.orElse(Long.MAX_VALUE),
- removeVoterHandlerState()
- .map(state ->
state.timeUntilOperationExpiration(currentTimeMs))
- .orElse(Long.MAX_VALUE)
+ Math.min(
+ removeVoterHandlerState()
+ .map(state ->
state.timeUntilOperationExpiration(currentTimeMs))
+ .orElse(Long.MAX_VALUE),
+ updateVoterHandlerState()
+ .map(state ->
state.timeUntilOperationExpiration(currentTimeMs))
+ .orElse(Long.MAX_VALUE)
+ )
);
}
/**
- * Resets all pending voter handler states, completing their futures with
the specified error.
+ * Resets all pending voter handler states with the given error.
* <p>
- * This method clears both add voter and remove voter handler states if
they exist. Each
- * pending operation's future is completed with the provided error.
+ * This method completes the futures of any pending add voter, remove
voter, and update voter
+ * operations with the provided error.
*
- * @param error the error to complete any pending operations with
+ * @param error the error to complete any existing operations with
+ * @param leaderAndEpoch the current leader and epoch information
+ * @param leaderEndpoints the current leader endpoints
*/
- public void maybeResetPendingVoterHandlerState(Errors error) {
+ public void maybeResetPendingVoterHandlerState(
+ Errors error,
+ LeaderAndEpoch leaderAndEpoch,
+ Endpoints leaderEndpoints
+ ) {
resetAddVoterHandlerState(error, null, Optional.empty());
resetRemoveVoterHandlerState(error, null, Optional.empty());
+ resetUpdateVoterHandlerState(error, leaderAndEpoch, leaderEndpoints,
Optional.empty());
}
/**
* Checks whether any voter change operation is currently pending.
* <p>
- * This method first expires any operations that have timed out at the
given timestamp,
- * then returns true if either an add voter or remove voter operation
remains pending.
+ * This method first expires any operations that have timed out, then
checks if any
+ * add voter, remove voter, or update voter operations remain active.
*
+ * @param leaderAndEpoch the current leader and epoch information
+ * @param leaderEndpoints the current leader endpoints
* @param currentTimeMs the current time in milliseconds
- * @return true if a voter change operation is pending, false otherwise
+ * @return true if any voter change operation is pending, false otherwise
*/
- public boolean isOperationPending(long currentTimeMs) {
- maybeExpirePendingOperation(currentTimeMs);
- return addVoterHandlerState.isPresent() ||
removeVoterHandlerState.isPresent();
+ public boolean isOperationPending(
+ LeaderAndEpoch leaderAndEpoch,
+ Endpoints leaderEndpoints,
+ long currentTimeMs
+ ) {
+ maybeExpirePendingOperation(leaderAndEpoch, leaderEndpoints,
currentTimeMs);
Review Comment:
I'm wondering if the following case is possible.
If we have added a `VoterRecord` to the `BatchAccumulator` in
`UpdateVoterHandler#storeUpdatedVoters` but then expire the pending operation
before the accumulator is drained could that technically lead to multiple voter
change requests at the same time?
In this case, we still have a pending UpdateVoterRequest since we need the
HWM to advance past the new voters record for the operation to be seen as
complete. But we will have reset the the `changeVoterSetState`.
--
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]
