jsancio commented on code in PR #22620:
URL: https://github.com/apache/kafka/pull/22620#discussion_r3475541754
##########
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:
Yes, this is possible.
KRaft maintains an invariant where only a single voter change operation can
be pending at any given time. This operation is restricted to a single
modification—one addition, removal, or update—relative to the committed voter
set. This constraint is essential to prevent split-brain scenarios or
inconsistencies, ensuring that a majority of voters in both the old and new
configurations have committed to and agree upon the change.
Currently, KRaft violates this invariant by relying on the change voter
handler state to enforce the single-pending-operation rule. A problem arises
when an operation times out while the KRaft control records are still residing
in the log or batch accumulator; in these instances, KRaft may incorrectly
reset the change voter handler state.
To resolve this, the code must be modified so that the change voter handler
state is not reset once KRaft control records have been dispatched to the batch
accumulator. Following that dispatch, the state should only be reset through
replication, specifically via High Watermark (HWM) advancement.
I created https://issues.apache.org/jira/browse/KAFKA-20740
--
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]
