jsancio commented on code in PR #22620:
URL: https://github.com/apache/kafka/pull/22620#discussion_r3467950420
##########
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 the kraft.version gets upgraded from 0 to 1 there is a small race
between when the volatileVoters state gets cleared and the kraft version and
voter set record get written to the log (and updates the partition state for
kraft). During this race the updated voter set will be in the batch accumulator
and won't be in the leader's volatile state or the log. The leader should
return a retriable error (REQUEST_TIMED_OUT) so that the follower can simply
retry the operation.
We used to have a comment for this case which I'll add back:
```java
/* This can happen if the remote voter sends an update voter
request before the
* updated kraft version has been written to the log
*/
```
##########
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 the kraft.version gets upgraded from 0 to 1 there is a small race
between when the volatileVoters state gets cleared and the kraft version and
voter set record get written to the log (and updates the partition state for
kraft). During this race the updated voter set will be in the batch accumulator
and won't be in the leader's volatile state or the log. The leader should
return a retriable error (REQUEST_TIMED_OUT) so that the follower can simply
retry the operation.
We used to have a comment for this case which I'll add back:
```java
/* This can happen if the remote voter sends an update voter
request before the
* updated kraft version has been written to the log
*/
```
--
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]
