kevin-wu24 commented on code in PR #22620:
URL: https://github.com/apache/kafka/pull/22620#discussion_r3461052421
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -155,50 +140,193 @@ 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
+ throw new IllegalStateException("Expected the high-watermark to be
known");
+ }
+
+ // Read the voter set from the log or leader state
+ KRaftVersion kraftVersion = partitionState.lastKraftVersion();
+ final Optional<KRaftVersionUpgrade.Voters> inMemoryVoters;
+ final Optional<VoterSet> voters;
+ if (kraftVersion.isReconfigSupported()) {
+ inMemoryVoters = Optional.empty();
+
+ // Check that there are no uncommitted VotersRecord
+ Optional<LogHistory.Entry<VoterSet>> votersEntry =
partitionState.lastVoterSetEntry();
+ if (votersEntry.isEmpty() || votersEntry.get().offset() >=
highWatermark.get()) {
+ voters = Optional.empty();
+ } else {
+ voters = votersEntry.map(LogHistory.Entry::value);
+ }
+ } else {
+ inMemoryVoters = leaderState.volatileVoters();
+ voters = inMemoryVoters.map(KRaftVersionUpgrade.Voters::voters);
+ }
+ if (voters.isEmpty()) {
+ logger.info("Unable to read the current voter set with kraft
version {}", kraftVersion);
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.REQUEST_TIMED_OUT,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
+ );
Review Comment:
When would this happen on the leader? Should handle this more strongly than
returning a `REQUEST_TIMED_OUT` error response?
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -155,50 +140,193 @@ 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))
Review Comment:
This looks like a different check than what we do for `AddVoter`'s
`APIVersionsResponse`.
This is saying the node being updated's kraft version range must match the
leader's exactly. The AddVoter check says the leader's kraft version value (1+)
must be in the supported range of the node being added.
If update voter is supported in kraft.version=0, why is there a kraft
version check?
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -101,50 +116,20 @@ public CompletionStage<UpdateRaftVoterResponseData>
handleUpdateVoterRequest(
);
}
- // Read the voter set from the log or leader state
- KRaftVersion kraftVersion = partitionState.lastKraftVersion();
- final Optional<KRaftVersionUpgrade.Voters> inMemoryVoters;
- final Optional<VoterSet> voters;
- if (kraftVersion.isReconfigSupported()) {
- inMemoryVoters = Optional.empty();
-
- // Check that there are no uncommitted VotersRecord
- Optional<LogHistory.Entry<VoterSet>> votersEntry =
partitionState.lastVoterSetEntry();
- if (votersEntry.isEmpty() || votersEntry.get().offset() >=
highWatermark.get()) {
- voters = Optional.empty();
- } else {
- voters = votersEntry.map(LogHistory.Entry::value);
- }
- } else {
- inMemoryVoters = leaderState.volatileVoters();
- if (inMemoryVoters.isEmpty()) {
- /* This can happen if the remote voter sends an update voter
request before the
- * updated kraft version has been written to the log
- */
- return CompletableFuture.completedFuture(
- RaftUtil.updateVoterResponse(
- Errors.REQUEST_TIMED_OUT,
- requestListenerName,
- leaderState.leaderAndEpoch(),
- leaderState.leaderEndpoints()
- )
- );
- }
- voters = inMemoryVoters.map(KRaftVersionUpgrade.Voters::voters);
- }
- if (voters.isEmpty()) {
- log.info("Unable to read the current voter set with kraft version
{}", kraftVersion);
+ // Check that the supported version range is valid
+ if (!validVersionRange(partitionState.lastKraftVersion(),
supportedKraftVersions)) {
return CompletableFuture.completedFuture(
RaftUtil.updateVoterResponse(
- Errors.REQUEST_TIMED_OUT,
+ Errors.INVALID_REQUEST,
requestListenerName,
leaderState.leaderAndEpoch(),
leaderState.leaderEndpoints()
)
);
}
- // Check that the supported version range is valid
- if (!validVersionRange(kraftVersion, supportedKraftVersions)) {
+
+ // Check that endpoints includes the default listener
+ if (voterEndpoints.address(requestSender.listenerName()).isEmpty()) {
Review Comment:
I think this check is already done in
`KafkaRaftClient#handleUpdateVoterRequest`:
```
Endpoints voterEndpoints =
Endpoints.fromUpdateVoterRequest(data.listeners());
if (voterEndpoints.address(channel.listenerName()).isEmpty()) {
return completedFuture(
RaftUtil.updateVoterResponse(
Errors.INVALID_REQUEST,
requestMetadata.listenerName(),
quorum.leaderAndEpoch(),
quorum.leaderEndpoints()
)
);
}
```
##########
raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java:
##########
@@ -219,56 +347,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 followr to send a FETCH
Review Comment:
```suggestion
* Complete the RPC but don't reset the handler state. This
allows the follower to send a FETCH
```
--
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]
