junrao commented on code in PR #22620:
URL: https://github.com/apache/kafka/pull/22620#discussion_r3530545826
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -55,17 +60,20 @@
*/
public final class UpdateVoterHandler {
private final KRaftControlRecordStateMachine partitionState;
- private final ListenerName defaultListenerName;
- private final Logger log;
+ private final RequestSender requestSender;
Review Comment:
Should we update the javadoc to reflect the newly added logic for
UpdateVoter?
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -76,8 +84,15 @@ public CompletionStage<UpdateRaftVoterResponseData>
handleUpdateVoterRequest(
UpdateRaftVoterRequestData.KRaftVersionFeature supportedKraftVersions,
long currentTimeMs
) {
+ var changeVoterState = leaderState.changeVoterState();
// Check if there are any pending voter change requests
- if (leaderState.changeVoterState().isOperationPending(currentTimeMs)) {
+ if (
+ changeVoterState.isOperationPending(
Review Comment:
Should this be merged into the previous line?
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -155,50 +140,202 @@ public CompletionStage<UpdateRaftVoterResponseData>
handleUpdateVoterRequest(
);
}
- // Check that endpoints includes the default listener
- if (voterEndpoints.address(defaultListenerName).isEmpty()) {
+ // Send API_VERSIONS request to new voter to test new default endpoint
+ var timeout = requestSender.send(
+ voterEndpoints
+ .address(requestSender.listenerName())
+ .map(address -> new Node(voterKey.id(), address.getHostName(),
address.getPort()))
+ .orElseThrow(
+ () -> new IllegalStateException(
+ String.format(
+ "Provided listeners %s do not contain a listener
for %s",
+ voterEndpoints,
+ requestSender.listenerName()
+ )
+ )
+ ),
+ this::buildApiVersionsRequest,
+ currentTimeMs
+ );
+ if (timeout.isEmpty()) {
return CompletableFuture.completedFuture(
RaftUtil.updateVoterResponse(
- Errors.INVALID_REQUEST,
+ Errors.REQUEST_TIMED_OUT,
requestListenerName,
leaderState.leaderAndEpoch(),
leaderState.leaderEndpoints()
)
);
}
+ var state = new UpdateVoterHandlerState(
+ voterKey,
+ voterEndpoints,
+ requestListenerName,
+ new SupportedVersionRange(
+ supportedKraftVersions.minSupportedVersion(),
+ supportedKraftVersions.maxSupportedVersion()
+ ),
+ time.timer(timeout.getAsLong())
+ );
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.UNKNOWN_SERVER_ERROR,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.of(state)
+ );
+
+ return state.future();
+ }
+
+ /**
+ * Handle the API_VERSIONS response for an update voter operation.
+ *
+ * @param leaderState the leader state
+ * @param source the node that sent the response
+ * @param error the error from the response
+ * @param supportedKraftVersions the supported kraft version range from
the response
+ * @param currentTimeMs the current time in milliseconds
+ * @return true if the update voter operation should continue, false if it
was aborted
+ */
+ public boolean handleApiVersionsResponse(
+ LeaderState<?> leaderState,
+ Node source,
+ Errors error,
+ Optional<ApiVersionsResponseData.SupportedFeatureKey>
supportedKraftVersions,
+ long currentTimeMs
+ ) {
+ var changeVoterState = leaderState.changeVoterState();
+ var handlerState = changeVoterState.updateVoterHandlerState();
+ if (handlerState.isEmpty()) {
+ // There are no pending add operation just ignore the api response
+ return true;
+ }
+
+ // Check that the API_VERSIONS response matches the id of the voter
getting added
+ var current = handlerState.get();
+ if (!current.expectingApiResponse(source.id())) {
+ logger.info(
+ "API_VERSIONS response is not expected from {}: voterKey is
{}, lastOffset is {}",
+ source,
+ current.voterKey(),
+ current.lastOffset()
+ );
+
+ return true;
+ } else if (error != Errors.NONE) {
+ // Abort operation if the API_VERSIONS returned an error
+ logger.info(
+ "Aborting update voter operation for {} at {} since
API_VERSIONS returned an error {}",
+ current.voterKey(),
+ current.voterEndpoints(),
+ error
+ );
+
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.REQUEST_TIMED_OUT,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
+ );
+
+ return false;
+ } else if (
+ !Optional.of(current.supportedKraftVersions())
+
.equals(supportedKraftVersions.map(this::convertToVersionRange))
+ ) {
+ // Check that the supported version from the ApiVersions response
matches the supported
+ // version from the UpdateVoter requet
Review Comment:
typo requet
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -219,56 +356,107 @@ private Optional<VoterSet> updateVoters(
voters.updateVoterIgnoringDirectoryId(updatedVoter);
}
- private CompletionStage<UpdateRaftVoterResponseData> storeUpdatedVoters(
+ private void storeUpdatedVoters(
LeaderState<?> leaderState,
- ReplicaKey voterKey,
+ UpdateVoterHandlerState current,
Optional<KRaftVersionUpgrade.Voters> inMemoryVoters,
VoterSet newVoters,
- ListenerName requestListenerName,
long currentTimeMs
) {
+ var changeVoterState = leaderState.changeVoterState();
+
if (inMemoryVoters.isEmpty()) {
- // Since the partition support reconfig then just write the update
voter set directly to the log
- leaderState.appendVotersRecord(newVoters, currentTimeMs);
+ /* Since the partition support reconfig then just write the update
voter set directly to the log.
+ *
+ * Complete the RPC but don't reset the handler state. This allows
the follower to send a FETCH
+ * request and help to commit the voter set change.
+ */
+ current.setLastOffset(leaderState.appendVotersRecord(newVoters,
currentTimeMs));
+ current.completeFuture(
+ RaftUtil.updateVoterResponse(
+ Errors.NONE,
+ current.requestListenerName(),
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints()
+ )
+ );
} else {
// Store the new voters set in the leader state since it cannot be
written to the log
var successful = leaderState.compareAndSetVolatileVoters(
inMemoryVoters.get(),
new KRaftVersionUpgrade.Voters(newVoters)
);
if (successful) {
- log.info(
+ logger.info(
"Updated in-memory voters from {} to {}",
inMemoryVoters.get().voters(),
newVoters
);
+
+ // Reset the check quorum state since the leader received a
successful request
+
leaderState.updateCheckQuorumForFollowingVoter(current.voterKey(),
currentTimeMs);
+
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.NONE,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
+ );
} else {
- log.info(
+ logger.info(
"Unable to update in-memory voters from {} to {}",
inMemoryVoters.get().voters(),
newVoters
);
- return CompletableFuture.completedFuture(
- RaftUtil.updateVoterResponse(
- Errors.REQUEST_TIMED_OUT,
- requestListenerName,
- leaderState.leaderAndEpoch(),
- leaderState.leaderEndpoints()
- )
+
+ // Fail the pending future if present
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.REQUEST_TIMED_OUT,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
);
}
}
+ }
- // Reset the check quorum state since the leader received a successful
request
- leaderState.updateCheckQuorumForFollowingVoter(voterKey,
currentTimeMs);
+ private ApiVersionsRequestData buildApiVersionsRequest() {
+ return new ApiVersionsRequest.Builder().build().data();
+ }
- return CompletableFuture.completedFuture(
- RaftUtil.updateVoterResponse(
- Errors.NONE,
- requestListenerName,
- leaderState.leaderAndEpoch(),
- leaderState.leaderEndpoints()
- )
+ private SupportedVersionRange convertToVersionRange(
+ ApiVersionsResponseData.SupportedFeatureKey supportedKraftVersions
+ ) {
+ return new SupportedVersionRange(
+ supportedKraftVersions.minVersion(),
+ supportedKraftVersions.maxVersion()
);
}
+
+ /**
+ * Called when the high watermark is updated to check if any pending
update voter operations
+ * can be completed.
+ *
+ * @param leaderState the leader state
+ * @param highWatermark the new high watermark offset
+ */
+ public void highWatermarkUpdated(LeaderState<?> leaderState, long
highWatermark) {
+ var changeVoterState = leaderState.changeVoterState();
+
+ changeVoterState
+ .updateVoterHandlerState()
+ .ifPresent(current -> {
+ current.lastOffset().ifPresent(lastOffset -> {
+ if (highWatermark > lastOffset) {
+ // VotersRecord with the added voter was committed;
complete the RPC
Review Comment:
This comment is mis-leading. The UpdateVoter RPC has already been completed
when processing the ApiVersionResponse. So, the following doesn't complete the
RPC.
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -155,50 +140,202 @@ public CompletionStage<UpdateRaftVoterResponseData>
handleUpdateVoterRequest(
);
}
- // Check that endpoints includes the default listener
- if (voterEndpoints.address(defaultListenerName).isEmpty()) {
+ // Send API_VERSIONS request to new voter to test new default endpoint
+ var timeout = requestSender.send(
+ voterEndpoints
+ .address(requestSender.listenerName())
+ .map(address -> new Node(voterKey.id(), address.getHostName(),
address.getPort()))
+ .orElseThrow(
+ () -> new IllegalStateException(
+ String.format(
+ "Provided listeners %s do not contain a listener
for %s",
+ voterEndpoints,
+ requestSender.listenerName()
+ )
+ )
+ ),
+ this::buildApiVersionsRequest,
+ currentTimeMs
+ );
+ if (timeout.isEmpty()) {
return CompletableFuture.completedFuture(
RaftUtil.updateVoterResponse(
- Errors.INVALID_REQUEST,
+ Errors.REQUEST_TIMED_OUT,
requestListenerName,
leaderState.leaderAndEpoch(),
leaderState.leaderEndpoints()
)
);
}
+ var state = new UpdateVoterHandlerState(
+ voterKey,
+ voterEndpoints,
+ requestListenerName,
+ new SupportedVersionRange(
+ supportedKraftVersions.minSupportedVersion(),
+ supportedKraftVersions.maxSupportedVersion()
+ ),
+ time.timer(timeout.getAsLong())
+ );
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.UNKNOWN_SERVER_ERROR,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.of(state)
+ );
+
+ return state.future();
+ }
+
+ /**
+ * Handle the API_VERSIONS response for an update voter operation.
+ *
+ * @param leaderState the leader state
+ * @param source the node that sent the response
+ * @param error the error from the response
+ * @param supportedKraftVersions the supported kraft version range from
the response
+ * @param currentTimeMs the current time in milliseconds
+ * @return true if the update voter operation should continue, false if it
was aborted
+ */
+ public boolean handleApiVersionsResponse(
+ LeaderState<?> leaderState,
+ Node source,
+ Errors error,
+ Optional<ApiVersionsResponseData.SupportedFeatureKey>
supportedKraftVersions,
+ long currentTimeMs
+ ) {
+ var changeVoterState = leaderState.changeVoterState();
+ var handlerState = changeVoterState.updateVoterHandlerState();
+ if (handlerState.isEmpty()) {
+ // There are no pending add operation just ignore the api response
+ return true;
+ }
+
+ // Check that the API_VERSIONS response matches the id of the voter
getting added
+ var current = handlerState.get();
+ if (!current.expectingApiResponse(source.id())) {
+ logger.info(
+ "API_VERSIONS response is not expected from {}: voterKey is
{}, lastOffset is {}",
+ source,
+ current.voterKey(),
+ current.lastOffset()
+ );
+
+ return true;
+ } else if (error != Errors.NONE) {
+ // Abort operation if the API_VERSIONS returned an error
+ logger.info(
+ "Aborting update voter operation for {} at {} since
API_VERSIONS returned an error {}",
+ current.voterKey(),
+ current.voterEndpoints(),
+ error
+ );
+
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.REQUEST_TIMED_OUT,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
+ );
+
+ return false;
+ } else if (
+ !Optional.of(current.supportedKraftVersions())
Review Comment:
Should this be merged into the previous line?
##########
raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandler.java:
##########
@@ -93,7 +93,13 @@ public CompletionStage<AddRaftVoterResponseData>
handleAddVoterRequest(
) {
var changeVoterState = leaderState.changeVoterState();
// Check if there are any pending voter change requests
- if (changeVoterState.isOperationPending(currentTimeMs)) {
+ if (
+ changeVoterState.isOperationPending(
Review Comment:
Could we merge this into the previous line?
##########
raft/src/main/java/org/apache/kafka/raft/internals/ChangeVoterHandlerState.java:
##########
@@ -150,7 +218,11 @@ private void updateUncommittedVoterChangeMetric() {
* @return the time in milliseconds until the next operation expires, or
Long.MAX_VALUE if
* no operations are pending
*/
- public long maybeExpirePendingOperation(long currentTimeMs) {
+ public long maybeExpirePendingOperation(
+ LeaderAndEpoch leaderAndEpoch,
+ Endpoints leaderEndpoints,
Review Comment:
Could we add the new params to javadoc?
##########
raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandler.java:
##########
@@ -82,7 +82,13 @@ public CompletionStage<RemoveRaftVoterResponseData>
handleRemoveVoterRequest(
) {
var changeVoterState = leaderState.changeVoterState();
// Check if there are any pending voter change requests
- if (changeVoterState.isOperationPending(currentTimeMs)) {
+ if (
+ changeVoterState.isOperationPending(
Review Comment:
Could this merged into the previous line?
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -155,50 +140,202 @@ public CompletionStage<UpdateRaftVoterResponseData>
handleUpdateVoterRequest(
);
}
- // Check that endpoints includes the default listener
- if (voterEndpoints.address(defaultListenerName).isEmpty()) {
+ // Send API_VERSIONS request to new voter to test new default endpoint
+ var timeout = requestSender.send(
+ voterEndpoints
+ .address(requestSender.listenerName())
+ .map(address -> new Node(voterKey.id(), address.getHostName(),
address.getPort()))
+ .orElseThrow(
+ () -> new IllegalStateException(
+ String.format(
+ "Provided listeners %s do not contain a listener
for %s",
+ voterEndpoints,
+ requestSender.listenerName()
+ )
+ )
+ ),
+ this::buildApiVersionsRequest,
+ currentTimeMs
+ );
+ if (timeout.isEmpty()) {
return CompletableFuture.completedFuture(
RaftUtil.updateVoterResponse(
- Errors.INVALID_REQUEST,
+ Errors.REQUEST_TIMED_OUT,
requestListenerName,
leaderState.leaderAndEpoch(),
leaderState.leaderEndpoints()
)
);
}
+ var state = new UpdateVoterHandlerState(
+ voterKey,
+ voterEndpoints,
+ requestListenerName,
+ new SupportedVersionRange(
+ supportedKraftVersions.minSupportedVersion(),
+ supportedKraftVersions.maxSupportedVersion()
+ ),
+ time.timer(timeout.getAsLong())
+ );
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.UNKNOWN_SERVER_ERROR,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.of(state)
+ );
+
+ return state.future();
+ }
+
+ /**
+ * Handle the API_VERSIONS response for an update voter operation.
+ *
+ * @param leaderState the leader state
+ * @param source the node that sent the response
+ * @param error the error from the response
+ * @param supportedKraftVersions the supported kraft version range from
the response
+ * @param currentTimeMs the current time in milliseconds
+ * @return true if the update voter operation should continue, false if it
was aborted
+ */
+ public boolean handleApiVersionsResponse(
+ LeaderState<?> leaderState,
+ Node source,
+ Errors error,
+ Optional<ApiVersionsResponseData.SupportedFeatureKey>
supportedKraftVersions,
+ long currentTimeMs
+ ) {
+ var changeVoterState = leaderState.changeVoterState();
+ var handlerState = changeVoterState.updateVoterHandlerState();
+ if (handlerState.isEmpty()) {
+ // There are no pending add operation just ignore the api response
+ return true;
+ }
+
+ // Check that the API_VERSIONS response matches the id of the voter
getting added
+ var current = handlerState.get();
+ if (!current.expectingApiResponse(source.id())) {
+ logger.info(
+ "API_VERSIONS response is not expected from {}: voterKey is
{}, lastOffset is {}",
+ source,
+ current.voterKey(),
+ current.lastOffset()
+ );
+
+ return true;
+ } else if (error != Errors.NONE) {
+ // Abort operation if the API_VERSIONS returned an error
+ logger.info(
+ "Aborting update voter operation for {} at {} since
API_VERSIONS returned an error {}",
+ current.voterKey(),
+ current.voterEndpoints(),
+ error
+ );
+
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.REQUEST_TIMED_OUT,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
+ );
+
+ return false;
+ } else if (
+ !Optional.of(current.supportedKraftVersions())
+
.equals(supportedKraftVersions.map(this::convertToVersionRange))
+ ) {
+ // Check that the supported version from the ApiVersions response
matches the supported
+ // version from the UpdateVoter requet
+ logger.error(
+ "The supported kraft version from UpdateVoters {} doesn't
match the supported " +
+ "kraft version from ApiVersions {}",
+ current.supportedKraftVersions(),
+ supportedKraftVersions
+ );
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.INVALID_REQUEST,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
+ );
+ return true;
+ }
+
+ // Check that the leader has established a HWM and committed the
current epoch
+ Optional<Long> highWatermark =
leaderState.highWatermark().map(LogOffsetMetadata::offset);
+ if (highWatermark.isEmpty()) {
+ // This cannot happen because the update voter request handler
already validated that
+ // the HWMN is known
Review Comment:
typo HWMN
--
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]
