You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by js...@apache.org on 2026/08/19 17:40:15 UTC
(kafka) branch trunk updated: KAFKA-20661; Implement checking for voter default endpoint (#22620)
This is an automated email from the ASF dual-hosted git repository.
jsancio pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 7565048a5c7 KAFKA-20661; Implement checking for voter default endpoint (#22620)
7565048a5c7 is described below
commit 7565048a5c7da092d1540bdb03f9d0846336e9d7
Author: José Armando García Sancio <js...@users.noreply.github.com>
AuthorDate: Wed Aug 19 13:40:08 2026 -0400
KAFKA-20661; Implement checking for voter default endpoint (#22620)
When a KRaft controller is misconfigured with advertised listeners
unreachable by other controllers, the cluster can become permanently
unavailable. If the active controller persists these unreachable
endpoints in the cluster metadata and then loses leadership, no new
leader can be established because controllers cannot reach each other
via VOTE requests.
This change adds endpoint reachability validation to the UPDATE_VOTER
handler. Before accepting an UPDATE_VOTER request, the leader now sends
an API_VERSIONS request to the voter's default listener to verify it can
communicate with that endpoint. The operation proceeds only if the
API_VERSIONS response is received successfully, confirming the endpoint
is reachable. If the request fails or times out, the UPDATE_VOTER
operation is rejected with REQUEST_TIMED_OUT before any configuration
change is written to the log.
This validation prevents unreachable endpoint configurations from being
persisted into the cluster metadata, ensuring the quorum can continue to
elect leaders even when individual controllers are temporarily
misconfigured. The implementation introduces UpdateVoterHandlerState to
track pending validation requests and extends ChangeVoterHandlerState to
manage UPDATE_VOTER operations alongside the existing ADD_VOTER and
REMOVE_VOTER handlers.
Reviewers: Kevin Wu <kw...@confluent.io>, Jonah Hooper
<jh...@confluent.io>, Jun Rao <ju...@gmail.com>
---
.../main/scala/kafka/raft/KafkaRaftManager.scala | 2 +-
.../java/org/apache/kafka/raft/FollowerState.java | 11 +-
.../org/apache/kafka/raft/KafkaRaftClient.java | 75 +--
.../java/org/apache/kafka/raft/LeaderState.java | 6 +-
.../kafka/raft/internals/AddVoterHandler.java | 99 +++-
.../kafka/raft/internals/AddVoterHandlerState.java | 58 ++-
.../raft/internals/ChangeVoterHandlerState.java | 231 ++++++++--
.../kafka/raft/internals/RemoveVoterHandler.java | 70 ++-
.../raft/internals/RemoveVoterHandlerState.java | 33 +-
.../kafka/raft/internals/UpdateVoterHandler.java | 511 ++++++++++++++++-----
.../raft/internals/UpdateVoterHandlerState.java | 162 +++++++
.../kafka/raft/KafkaRaftClientReconfigTest.java | 366 +++++++++++++--
.../org/apache/kafka/raft/QuorumStateTest.java | 2 +-
.../apache/kafka/raft/RaftEventSimulationTest.java | 46 +-
.../raft/internals/AddVoterHandlerStateTest.java | 133 ++++++
.../internals/ChangeVoterHandlerStateTest.java | 275 ++++++++++-
.../kafka/raft/internals/KafkaRaftMetricsTest.java | 2 +-
.../internals/RemoveVoterHandlerStateTest.java | 64 +++
.../internals/UpdateVoterHandlerStateTest.java | 146 ++++++
.../apache/kafka/raft/RaftClientTestContext.java | 4 +-
.../org/apache/kafka/raft/VoterSetTestUtil.java | 2 +-
.../org/apache/kafka/server/common/Feature.java | 4 +-
.../org/apache/kafka/server/BrokerFeatures.java | 10 +-
23 files changed, 2024 insertions(+), 288 deletions(-)
diff --git a/core/src/main/scala/kafka/raft/KafkaRaftManager.scala b/core/src/main/scala/kafka/raft/KafkaRaftManager.scala
index b01ba8ce1a4..f0f4bd1f05f 100644
--- a/core/src/main/scala/kafka/raft/KafkaRaftManager.scala
+++ b/core/src/main/scala/kafka/raft/KafkaRaftManager.scala
@@ -177,7 +177,7 @@ class KafkaRaftManager[T](
clusterId,
bootstrapServers,
localListeners,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(config.unstableFeatureVersionsEnabled),
raftConfig
)
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/FollowerState.java b/raft/src/main/java/org/apache/kafka/raft/FollowerState.java
index 2ca35a892e7..600f14a2bb2 100644
--- a/raft/src/main/java/org/apache/kafka/raft/FollowerState.java
+++ b/raft/src/main/java/org/apache/kafka/raft/FollowerState.java
@@ -155,9 +155,14 @@ public final class FollowerState implements EpochState {
}
private long updateVoterPeriodMs() {
- // Allow for a few rounds of fetch request before attempting to update
- // the voter state
- return fetchTimeoutMs;
+ /* Allow for a few rounds of fetch requests before attempting to update the voter state.
+ *
+ * In the default configuration, both the request timeout and the fetch timeout are set to 2
+ * seconds. For this operation, they should not be the same, otherwise the leader can be
+ * prevented from accepting voter changes if the ApiVersions request times out for as long as
+ * the follower keeps retrying.
+ */
+ return 2 * fetchTimeoutMs;
}
public boolean hasUpdateVoterSetPeriodExpired(long currentTimeMs) {
diff --git a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
index 9a9dbda1051..da2b8ef11ca 100644
--- a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
+++ b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
@@ -397,6 +397,7 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
private void maybeNotifyVoterHandlerOnHWmUpdate(LeaderState<T> state, long highWatermark) {
addVoterHandler.highWatermarkUpdated(state, highWatermark);
removeVoterHandler.highWatermarkUpdated(state, highWatermark);
+ updateVoterHandler.highWatermarkUpdated(state, highWatermark);
}
private void updateListenersProgress(long highWatermark) {
@@ -578,15 +579,17 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
onBecomeFollower(currentTimeMs);
}
+ var requestSender = new DefaultRequestSender(
+ requestManager,
+ channel,
+ messageQueue,
+ logContext
+ );
+
// Specialized add voter handler
this.addVoterHandler = new AddVoterHandler(
partitionState,
- new DefaultRequestSender(
- requestManager,
- channel,
- messageQueue,
- logContext
- ),
+ requestSender,
time,
logContext
);
@@ -604,7 +607,8 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
// Specialized update voter handler
this.updateVoterHandler = new UpdateVoterHandler(
partitionState,
- channel.listenerName(),
+ requestSender,
+ time,
logContext
);
}
@@ -2316,19 +2320,37 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
return true;
}
+ var leaderState = quorum.leaderStateOrThrow();
+
ApiVersionsResponseData response = (ApiVersionsResponseData) responseMetadata.data();
Errors error = Errors.forCode(response.errorCode());
Optional<ApiVersionsResponseData.SupportedFeatureKey> supportedKraftVersions =
Optional.ofNullable(response.supportedFeatures().find(KRaftVersion.FEATURE_NAME));
- return addVoterHandler.handleApiVersionsResponse(
- quorum.leaderStateOrThrow(),
- responseMetadata.source(),
- error,
- supportedKraftVersions,
- currentTimeMs
- );
+ if (leaderState.changeVoterState().addVoterHandlerState().isPresent()) {
+ return addVoterHandler.handleApiVersionsResponse(
+ leaderState,
+ responseMetadata.source(),
+ error,
+ supportedKraftVersions,
+ currentTimeMs
+ );
+ } else if (leaderState.changeVoterState().updateVoterHandlerState().isPresent()) {
+ return updateVoterHandler.handleApiVersionsResponse(
+ leaderState,
+ responseMetadata.source(),
+ error,
+ supportedKraftVersions,
+ currentTimeMs
+ );
+ } else {
+ logger.debug(
+ "Received API_VERSIONS response from {} but no voter change operation is pending",
+ responseMetadata.source()
+ );
+ return true;
+ }
}
private boolean handleAddVoterResponse(
@@ -2510,18 +2532,6 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
);
}
- 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()
- )
- );
- }
-
UpdateRaftVoterRequestData.KRaftVersionFeature supportedKraftVersions = data.kRaftVersionFeature();
if (supportedKraftVersions.minSupportedVersion() < 0 ||
supportedKraftVersions.maxSupportedVersion() < 0 ||
@@ -2541,7 +2551,7 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
quorum.leaderStateOrThrow(),
requestMetadata.listenerName(),
voter.get(),
- voterEndpoints,
+ Endpoints.fromUpdateVoterRequest(data.listeners()),
supportedKraftVersions,
currentTimeMs
);
@@ -3179,9 +3189,14 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
return 0L;
}
- long timeUntilVoterChangeExpires = state
- .changeVoterState()
- .maybeExpirePendingOperation(currentTimeMs);
+ var changeVoterState = state.changeVoterState();
+ long timeUntilVoterChangeExpires = changeVoterState.hasPendingOperation() ?
+ changeVoterState.maybeExpirePendingOperation(
+ quorum.leaderAndEpoch(),
+ quorum.leaderEndpoints(),
+ currentTimeMs
+ ) :
+ Long.MAX_VALUE;
long timeUntilFlush = maybeAppendBatches(
state,
diff --git a/raft/src/main/java/org/apache/kafka/raft/LeaderState.java b/raft/src/main/java/org/apache/kafka/raft/LeaderState.java
index 0891e952024..d7c0e8b5ade 100644
--- a/raft/src/main/java/org/apache/kafka/raft/LeaderState.java
+++ b/raft/src/main/java/org/apache/kafka/raft/LeaderState.java
@@ -1077,7 +1077,11 @@ public final class LeaderState<T> implements EpochState {
@Override
public void close() {
- changeVoterState.maybeResetPendingVoterHandlerState(Errors.NOT_LEADER_OR_FOLLOWER);
+ changeVoterState.maybeResetPendingVoterHandlerState(
+ Errors.NOT_LEADER_OR_FOLLOWER,
+ leaderAndEpoch(),
+ leaderEndpoints()
+ );
kafkaRaftMetrics.removeLeaderMetrics();
accumulator.close();
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandler.java b/raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandler.java
index 3dac2d5046d..c1dc1f9611a 100644
--- a/raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandler.java
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandler.java
@@ -52,16 +52,29 @@ import java.util.concurrent.CompletionStage;
* 4. Check that the new voter's id is not part of the existing voter set, otherwise return the
* DUPLICATE_VOTER error.
* 5. Send an API_VERSIONS RPC to the first (default) listener to discover the supported
- * kraft.version of the new voter.
+ * kraft.version of the new voter, and record the operation as pending, returning the
+ * REQUEST_TIMED_OUT error if the request cannot be sent.
+ *
+ * The response is not sent synchronously; the rest of the algorithm runs once the API_VERSIONS
+ * response is handled by {@link #handleApiVersionsResponse}:
+ *
* 6. Check that the new voter supports the current kraft.version, otherwise return the
* INVALID_REQUEST error.
* 7. Check that the new voter is caught up to the log end offset of the leader, otherwise return
* a REQUEST_TIMED_OUT error.
* 8. Append the updated VotersRecord to the log. The KRaft internal listener will read this
* uncommitted record from the log and add the new voter to the set of voters.
- * 9. Wait for the VotersRecord to commit using the majority of the new set of voters. Return a
- * REQUEST_TIMED_OUT error if it doesn't commit in time.
- * 10. Send the AddVoter successful response to the client.
+ * 9. If the request did not ask to wait for the change to commit ({@code ackWhenCommitted} is
+ * false), send the AddVoter successful response immediately, without waiting for the record
+ * to commit.
+ * 10. Otherwise, wait for the VotersRecord to commit using the majority of the new set of
+ * voters, then send the AddVoter successful response; return a REQUEST_TIMED_OUT error if it
+ * doesn't commit in time. Either way, the pending operation is only cleared, via
+ * {@link #highWatermarkUpdated}, once the HWM advances past the offset of the appended
+ * record, which allows the next voter change operation to proceed.
+ *
+ * A pending operation that doesn't complete before its timeout expires is also aborted with the
+ * REQUEST_TIMED_OUT error, by {@link ChangeVoterHandlerState#maybeExpirePendingOperation}.
*
* The algorithm above could be improved as part of KAFKA-17147. Instead of returning an error
* immediately for 1., 2. and 7., KRaft can wait with a timeout until those invariants are true.
@@ -72,6 +85,15 @@ public final class AddVoterHandler {
private final Time time;
private final Logger logger;
+ /**
+ * Creates a new handler for add voter requests.
+ *
+ * @param partitionState the KRaft partition state, used to read the currently finalized
+ * kraft.version and the log's voter set
+ * @param requestSender used to send the API_VERSIONS request to the voter being added
+ * @param time the time implementation, used to create the timer that bounds a pending operation
+ * @param logContext used to create this class's logger
+ */
public AddVoterHandler(
KRaftControlRecordStateMachine partitionState,
RequestSender requestSender,
@@ -84,6 +106,22 @@ public final class AddVoterHandler {
this.logger = logContext.logger(AddVoterHandler.class);
}
+ /**
+ * Handle an AddVoter request.
+ * <p>
+ * See the class documentation for the full set of steps that this method and
+ * {@link #handleApiVersionsResponse} perform together.
+ *
+ * @param leaderState the leader state
+ * @param voterKey the id and directory id of the voter to add
+ * @param voterEndpoints the endpoints of the voter to add
+ * @param ackWhenCommitted if true, the response is withheld until the voter change commits;
+ * if false, the response is sent as soon as the change is appended to the log
+ * @param currentTimeMs the current time in milliseconds
+ * @return a future for the AddVoter response; it completes immediately if the request is
+ * rejected outright, or later, once the API_VERSIONS round trip (and, depending on
+ * {@code ackWhenCommitted}, the commit) finishes
+ */
public CompletionStage<AddRaftVoterResponseData> handleAddVoterRequest(
LeaderState<?> leaderState,
ReplicaKey voterKey,
@@ -93,7 +131,12 @@ public final class AddVoterHandler {
) {
var changeVoterState = leaderState.changeVoterState();
// Check if there are any pending voter change requests
- if (changeVoterState.isOperationPending(currentTimeMs)) {
+ if (changeVoterState.isOperationPending(
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ currentTimeMs
+ )
+ ) {
return CompletableFuture.completedFuture(
RaftUtil.addVoterResponse(
Errors.REQUEST_TIMED_OUT,
@@ -199,6 +242,26 @@ public final class AddVoterHandler {
return state.future();
}
+ /**
+ * Handle the API_VERSIONS response for a pending add voter operation.
+ * <p>
+ * This may abort the pending operation, completing its future with an error, if the response
+ * doesn't come from the expected voter, if the API_VERSIONS request failed, if the new
+ * voter's supported kraft.version range doesn't cover the cluster's finalized kraft.version,
+ * or if the voter isn't caught up to the leader's log end offset. Otherwise, it appends the
+ * updated VotersRecord to the log and, if {@code ackWhenCommitted} is false, completes the
+ * response immediately.
+ *
+ * @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 false only when the API_VERSIONS request itself failed, which is the only case
+ * where the caller (see {@code KafkaRaftClient#handleResponse}) should treat this as
+ * an unsuccessful response for request-tracking purposes; true otherwise, including
+ * when this method aborts the pending add voter operation for another reason
+ */
public boolean handleApiVersionsResponse(
LeaderState<?> leaderState,
Node source,
@@ -209,7 +272,7 @@ public final class AddVoterHandler {
var changeVoterState = leaderState.changeVoterState();
var handlerState = changeVoterState.addVoterHandlerState();
if (handlerState.isEmpty()) {
- // There are no pending add operation just ignore the api response
+ // There is no pending add operation; just ignore the API_VERSIONS response
return true;
}
@@ -238,7 +301,8 @@ public final class AddVoterHandler {
changeVoterState.resetAddVoterHandlerState(
Errors.REQUEST_TIMED_OUT,
String.format(
- "Aborted add voter operation for since API_VERSIONS returned an error %s",
+ "Aborted add voter operation for %s since API_VERSIONS returned an error %s",
+ current.voterKey(),
error
),
Optional.empty()
@@ -329,11 +393,25 @@ public final class AddVoterHandler {
if (!current.ackWhenCommitted()) {
// complete the future to send response, but do not reset the state,
// since the new voter set is not yet committed
- current.future().complete(RaftUtil.addVoterResponse(Errors.NONE, null));
+ current.completeFuture(RaftUtil.addVoterResponse(Errors.NONE, null));
}
return true;
}
+ /**
+ * Called when the high watermark advances to check if a pending add voter operation can be
+ * cleared.
+ * <p>
+ * If the AddVoter request asked to wait for the commit ({@code ackWhenCommitted} is true),
+ * this is also when the response is completed with success. Otherwise, the response was
+ * already sent by {@link #handleApiVersionsResponse} as soon as the VotersRecord was
+ * appended, and this method only clears the pending operation, allowing the next voter
+ * change request to be accepted, once the high watermark advances past the offset of that
+ * record.
+ *
+ * @param leaderState the leader state
+ * @param highWatermark the new high watermark offset
+ */
public void highWatermarkUpdated(LeaderState<?> leaderState, long highWatermark) {
var changeVoterState = leaderState.changeVoterState();
@@ -342,7 +420,10 @@ public final class AddVoterHandler {
.ifPresent(current ->
current.lastOffset().ifPresent(lastOffset -> {
if (highWatermark > lastOffset) {
- // VotersRecord with the added voter was committed; complete the RPC
+ // The VotersRecord with the added voter was committed. If the request
+ // asked to wait for the commit, this completes the RPC with success;
+ // otherwise the response was already sent and this just clears the
+ // pending operation.
changeVoterState
.resetAddVoterHandlerState(Errors.NONE, null, Optional.empty());
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandlerState.java b/raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandlerState.java
index 5cd9dec47d5..c5bd67f2457 100644
--- a/raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandlerState.java
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/AddVoterHandlerState.java
@@ -24,7 +24,15 @@ import org.apache.kafka.raft.ReplicaKey;
import java.util.OptionalLong;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+/**
+ * Tracks the state of a single pending add voter operation.
+ * <p>
+ * An instance is created by {@link AddVoterHandler#handleAddVoterRequest} once the API_VERSIONS
+ * request has been sent to the new voter, and is held by {@link ChangeVoterHandlerState} until
+ * the operation completes, is aborted, or expires.
+ */
public final class AddVoterHandlerState {
private final ReplicaKey voterKey;
private final Endpoints voterEndpoints;
@@ -46,15 +54,33 @@ public final class AddVoterHandlerState {
this.timeout = timeout;
}
+ /**
+ * Returns the time in milliseconds until this operation expires.
+ *
+ * @param currentTimeMs the current time in milliseconds
+ * @return the remaining time in milliseconds until expiration
+ */
public long timeUntilOperationExpiration(long currentTimeMs) {
timeout.update(currentTimeMs);
return timeout.remainingMs();
}
+ /**
+ * Checks whether this handler state is expecting an API_VERSIONS response from the given replica.
+ *
+ * @param replicaId the replica id to check
+ * @return true if expecting a response from this replica, false otherwise
+ */
public boolean expectingApiResponse(int replicaId) {
return lastOffset.isEmpty() && replicaId == voterKey.id();
}
+ /**
+ * Sets the offset of the VotersRecord that was appended to the log for this add voter operation.
+ *
+ * @param lastOffset the offset of the VotersRecord that was appended to the log
+ * @throws IllegalStateException if the last offset has already been set
+ */
public void setLastOffset(long lastOffset) {
if (this.lastOffset.isPresent()) {
throw new IllegalStateException(
@@ -71,23 +97,53 @@ public final class AddVoterHandlerState {
this.lastOffset = OptionalLong.of(lastOffset);
}
+ /**
+ * Returns the voter key for the voter being added.
+ *
+ * @return the voter key
+ */
public ReplicaKey voterKey() {
return voterKey;
}
+ /**
+ * Returns the endpoints for the voter being added.
+ *
+ * @return the voter endpoints
+ */
public Endpoints voterEndpoints() {
return voterEndpoints;
}
+ /**
+ * Returns whether the AddVoter response should be withheld until the voter change commits.
+ *
+ * @return true if the response should be sent only after the change commits, false if it
+ * should be sent as soon as the change is appended to the log
+ */
public boolean ackWhenCommitted() {
return ackWhenCommitted;
}
+ /**
+ * Returns the offset of the VotersRecord if it has been appended to the log.
+ *
+ * @return the last offset, or empty if not yet appended
+ */
public OptionalLong lastOffset() {
return lastOffset;
}
- public CompletableFuture<AddRaftVoterResponseData> future() {
+ /**
+ * Completes the future with the provided response.
+ *
+ * @param response the response to complete the future with
+ */
+ public void completeFuture(AddRaftVoterResponseData response) {
+ future.complete(response);
+ }
+
+ CompletionStage<AddRaftVoterResponseData> future() {
return future;
}
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/ChangeVoterHandlerState.java b/raft/src/main/java/org/apache/kafka/raft/internals/ChangeVoterHandlerState.java
index 7325cbf8d70..84a65af31d4 100644
--- a/raft/src/main/java/org/apache/kafka/raft/internals/ChangeVoterHandlerState.java
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/ChangeVoterHandlerState.java
@@ -17,17 +17,19 @@
package org.apache.kafka.raft.internals;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.raft.Endpoints;
+import org.apache.kafka.raft.LeaderAndEpoch;
import org.apache.kafka.raft.RaftUtil;
import java.util.Optional;
/**
- * Manages the state of add and remove voter operations.
+ * Manages the state of add, remove, and update voter operations.
* <p>
- * This class maintains at most one pending voter change operation at a time. Add voter and
- * remove voter operations are mutually exclusive - only one type can be in progress at any
- * given time. When an operation is reset or expires, its associated future is completed with
- * an appropriate error response.
+ * This class maintains at most one pending voter change operation at a time. Add voter, remove
+ * voter, and update voter operations are mutually exclusive - only one type can be in progress
+ * at any given time. When an operation is reset or expires, its associated future is completed
+ * with an appropriate error response.
* <p>
* The class also updates the uncommitted voter change metric to reflect whether a voter
* change operation is currently pending.
@@ -35,12 +37,14 @@ import java.util.Optional;
public final class ChangeVoterHandlerState {
private Optional<AddVoterHandlerState> addVoterHandlerState = Optional.empty();
private Optional<RemoveVoterHandlerState> removeVoterHandlerState = Optional.empty();
+ private Optional<UpdateVoterHandlerState> updateVoterHandlerState = Optional.empty();
+
private final KafkaRaftMetrics kafkaRaftMetrics;
/**
- * Constructs a new ChangeVoterHandlerState.
+ * Creates a new change-voter handler state tracker.
*
- * @param kafkaRaftMetrics the metrics instance to update when voter change state changes
+ * @param kafkaRaftMetrics used to report whether a voter change operation is currently pending
*/
public ChangeVoterHandlerState(KafkaRaftMetrics kafkaRaftMetrics) {
this.kafkaRaftMetrics = kafkaRaftMetrics;
@@ -61,29 +65,23 @@ public final class ChangeVoterHandlerState {
* <p>
* If an add voter handler state already exists, its future will be completed with the
* provided error and message before being replaced. If the new state is non-empty and a
- * remove voter handler state is currently present, this method throws an IllegalStateException
- * to enforce mutual exclusivity.
+ * remove voter or update voter handler state is currently present, this method throws an
+ * IllegalStateException to enforce mutual exclusivity.
*
* @param error the error to complete any existing add voter operation with
* @param message the error message to include in the response, or null for no message
* @param state the new add voter handler state, or empty to clear the state
* @throws IllegalStateException if attempting to set a non-empty add voter state while a
- * remove voter state is already present
+ * remove voter or update voter state is already present
*/
public void resetAddVoterHandlerState(
Errors error,
String message,
Optional<AddVoterHandlerState> state
) {
- if (state.isPresent() && removeVoterHandlerState.isPresent()) {
- throw new IllegalStateException(
- "Cannot set add voter handler state when remove voter handler state is already present"
- );
- }
+ validateMutualExclusivity(state, removeVoterHandlerState, updateVoterHandlerState);
addVoterHandlerState.ifPresent(
- handlerState -> handlerState
- .future()
- .complete(RaftUtil.addVoterResponse(error, message))
+ handlerState -> handlerState.completeFuture(RaftUtil.addVoterResponse(error, message))
);
addVoterHandlerState = state;
updateUncommittedVoterChangeMetric();
@@ -104,53 +102,143 @@ public final class ChangeVoterHandlerState {
* <p>
* If a remove voter handler state already exists, its future will be completed with the
* provided error and message before being replaced. If the new state is non-empty and an
- * add voter handler state is currently present, this method throws an IllegalStateException
- * to enforce mutual exclusivity.
+ * add voter or update voter handler state is currently present, this method throws an
+ * IllegalStateException to enforce mutual exclusivity.
*
* @param error the error to complete any existing remove voter operation with
* @param message the error message to include in the response, or null for no message
* @param state the new remove voter handler state, or empty to clear the state
* @throws IllegalStateException if attempting to set a non-empty remove voter state while an
- * add voter state is already present
+ * add voter or update voter state is already present
*/
public void resetRemoveVoterHandlerState(
Errors error,
String message,
Optional<RemoveVoterHandlerState> state
) {
- if (state.isPresent() && addVoterHandlerState.isPresent()) {
- throw new IllegalStateException(
- "Cannot set remove voter handler state when add voter handler state is already present"
- );
- }
+ validateMutualExclusivity(addVoterHandlerState, state, updateVoterHandlerState);
removeVoterHandlerState.ifPresent(
- handlerState -> handlerState
- .future()
- .complete(RaftUtil.removeVoterResponse(error, message))
+ handlerState -> handlerState.completeFuture(RaftUtil.removeVoterResponse(error, message))
);
removeVoterHandlerState = state;
updateUncommittedVoterChangeMetric();
}
- private void updateUncommittedVoterChangeMetric() {
- kafkaRaftMetrics.updateUncommittedVoterChange(
- addVoterHandlerState.isPresent() || removeVoterHandlerState.isPresent()
+ /**
+ * Returns the current update voter handler state, if one exists.
+ *
+ * @return an Optional containing the update voter handler state, or empty if no update voter
+ * operation is pending
+ */
+ public Optional<UpdateVoterHandlerState> updateVoterHandlerState() {
+ return updateVoterHandlerState;
+ }
+
+ /**
+ * Resets the update voter handler state to the specified state.
+ * <p>
+ * If an update voter handler state already exists, its future will be completed with the
+ * provided error before being replaced. If the new state is non-empty and an add voter or
+ * remove voter handler state is currently present, this method throws an IllegalStateException
+ * to enforce mutual exclusivity.
+ *
+ * @param error the error to complete any existing update voter operation with
+ * @param leaderAndEpoch the current leader and epoch information
+ * @param leaderEndpoints the current leader endpoints
+ * @param state the new update voter handler state, or empty to clear the state
+ * @throws IllegalStateException if attempting to set a non-empty update voter state while an
+ * add voter or remove voter state is already present
+ */
+ public void resetUpdateVoterHandlerState(
+ Errors error,
+ LeaderAndEpoch leaderAndEpoch,
+ Endpoints leaderEndpoints,
+ Optional<UpdateVoterHandlerState> state
+ ) {
+ validateMutualExclusivity(addVoterHandlerState, removeVoterHandlerState, state);
+ updateVoterHandlerState.ifPresent(
+ handlerState -> handlerState.completeFuture(
+ RaftUtil.updateVoterResponse(
+ error,
+ handlerState.requestListenerName(),
+ leaderAndEpoch,
+ leaderEndpoints
+ )
+ )
);
+ updateVoterHandlerState = state;
+ updateUncommittedVoterChangeMetric();
+ }
+
+ /**
+ * Validates that at most one voter change operation is active.
+ * <p>
+ * This enforces mutual exclusivity between add, remove, and update voter operations.
+ *
+ * @param newAdd the new add voter state being set (if any)
+ * @param newRemove the new remove voter state being set (if any)
+ * @param newUpdate the new update voter state being set (if any)
+ * @throws IllegalStateException if more than one operation would be active
+ */
+ private void validateMutualExclusivity(
+ Optional<AddVoterHandlerState> newAdd,
+ Optional<RemoveVoterHandlerState> newRemove,
+ Optional<UpdateVoterHandlerState> newUpdate
+ ) {
+ int activeCount = 0;
+ if (newAdd.isPresent()) activeCount++;
+ if (newRemove.isPresent()) activeCount++;
+ if (newUpdate.isPresent()) activeCount++;
+
+ if (activeCount > 1) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot have multiple voter change operations active simultaneously: " +
+ "add=%s, remove=%s, update=%s",
+ newAdd.isPresent(),
+ newRemove.isPresent(),
+ newUpdate.isPresent()
+ )
+ );
+ }
+ }
+
+ /**
+ * Returns whether an add, remove, or update voter operation is currently pending.
+ *
+ * @return true if an operation is pending, false otherwise
+ */
+ public boolean hasPendingOperation() {
+ return addVoterHandlerState.isPresent() ||
+ removeVoterHandlerState.isPresent() ||
+ updateVoterHandlerState.isPresent();
+ }
+
+ private void updateUncommittedVoterChangeMetric() {
+ kafkaRaftMetrics.updateUncommittedVoterChange(hasPendingOperation());
}
/**
* Checks for and expires any pending voter change operations that have timed out.
* <p>
- * This method evaluates both add voter and remove voter operations. Any operation that
- * has expired (timeUntilOperationExpiration returns 0) is reset with a REQUEST_TIMED_OUT
- * error. The method then returns the minimum time remaining until the next operation
- * expiration.
+ * This method evaluates the add voter, remove voter, and update voter operations. Any
+ * operation that has expired (timeUntilOperationExpiration returns 0) is reset with a
+ * REQUEST_TIMED_OUT error. The method then returns the minimum time remaining until the next
+ * operation expiration.
*
+ * @param leaderAndEpoch the current leader and epoch information, used to complete the
+ * future of an expired update voter operation
+ * @param leaderEndpoints the current leader endpoints, used to complete the future of an
+ * expired update voter operation
* @param currentTimeMs the current time in milliseconds
* @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,
+ long currentTimeMs
+ ) {
// First abort any expired operations
long timeUntilAddVoterExpiration = addVoterHandlerState()
.map(state -> state.timeUntilOperationExpiration(currentTimeMs))
@@ -168,41 +256,82 @@ public final class ChangeVoterHandlerState {
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);
+ return addVoterHandlerState.isPresent() || removeVoterHandlerState.isPresent() || updateVoterHandlerState.isPresent();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "ChangeVoterHandlerState(addVoterHandlerState=%s, removeVoterHandlerState=%s, updateVoterHandlerState=%s)",
+ addVoterHandlerState,
+ removeVoterHandlerState,
+ updateVoterHandlerState
+ );
}
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandler.java b/raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandler.java
index 713221ceeea..5ee4abc9f77 100644
--- a/raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandler.java
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandler.java
@@ -38,18 +38,25 @@ import java.util.concurrent.CompletionStage;
/**
* This type implements the protocol for removing a voter from a KRaft partition.
*
- * The general algorithm for removing voter to the voter set is:
+ * The general algorithm for removing a voter from the voter set is:
*
* 1. Check that the leader has fenced the previous leader(s) by checking that the HWM is known,
* otherwise return the REQUEST_TIMED_OUT error.
* 2. Check that the cluster supports kraft.version 1, otherwise return the UNSUPPORTED_VERSION error.
* 3. Check that there are no uncommitted voter changes, otherwise return the REQUEST_TIMED_OUT error.
- * 4. Append the updated VotersRecord to the log. The KRaft internal listener will read this
+ * 4. Check that the voter being removed is part of the current voter set, otherwise return the
+ * VOTER_NOT_FOUND error.
+ * 5. Append the updated VotersRecord to the log. The KRaft internal listener will read this
* uncommitted record from the log and remove the voter from the set of voters.
- * 5. Wait for the VotersRecord to commit using the majority of the new set of voters. Return a
- * REQUEST_TIMED_OUT error if it doesn't commit in time.
- * 6. Send the RemoveVoter successful response to the client.
- * 7. Resign the leadership if the leader is not in the new voter set
+ * 6. Wait for the VotersRecord to commit using the majority of the new set of voters, see
+ * {@link #highWatermarkUpdated}. Return a REQUEST_TIMED_OUT error if it doesn't commit in
+ * time, see {@link ChangeVoterHandlerState#maybeExpirePendingOperation}.
+ * 7. Send the RemoveVoter successful response to the client.
+ * 8. Resign the leadership if the leader is not in the new voter set.
+ *
+ * Unlike {@link AddVoterHandler} and {@link UpdateVoterHandler}, this operation does not need to
+ * contact the removed voter, so the VotersRecord is appended synchronously within
+ * {@link #handleRemoveVoterRequest} instead of waiting for an API_VERSIONS round trip.
*/
public final class RemoveVoterHandler {
private final Optional<ReplicaKey> localReplicaKey;
@@ -58,6 +65,20 @@ public final class RemoveVoterHandler {
private final long requestTimeoutMs;
private final Logger logger;
+ /**
+ * Creates a new handler for remove voter requests.
+ *
+ * @param nodeId this replica's node id, if it is eligible to be a voter; used together with
+ * {@code nodeDirectoryId} to detect when the leader itself is being removed, so that
+ * it can resign
+ * @param nodeDirectoryId this replica's directory id, used together with {@code nodeId} to
+ * build its {@link ReplicaKey}
+ * @param partitionState the KRaft partition state, used to read the currently finalized
+ * kraft.version and the log's voter set
+ * @param time the time implementation, used to create the timer that bounds a pending operation
+ * @param requestTimeoutMs the timeout, in milliseconds, for the appended VotersRecord to commit
+ * @param logContext used to create this class's logger
+ */
public RemoveVoterHandler(
OptionalInt nodeId,
Uuid nodeDirectoryId,
@@ -75,6 +96,19 @@ public final class RemoveVoterHandler {
this.logger = logContext.logger(RemoveVoterHandler.class);
}
+ /**
+ * Handle a RemoveVoter request.
+ * <p>
+ * See the class documentation for the full set of steps that this method and
+ * {@link #highWatermarkUpdated} perform together.
+ *
+ * @param leaderState the leader state
+ * @param voterKey the id and directory id of the voter to remove
+ * @param currentTimeMs the current time in milliseconds
+ * @return a future for the RemoveVoter response; it completes immediately if the request is
+ * rejected outright, or later, once the appended VotersRecord commits, via
+ * {@link #highWatermarkUpdated}
+ */
public CompletionStage<RemoveRaftVoterResponseData> handleRemoveVoterRequest(
LeaderState<?> leaderState,
ReplicaKey voterKey,
@@ -82,7 +116,13 @@ public final class RemoveVoterHandler {
) {
var changeVoterState = leaderState.changeVoterState();
// Check if there are any pending voter change requests
- if (changeVoterState.isOperationPending(currentTimeMs)) {
+ if (
+ changeVoterState.isOperationPending(
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ currentTimeMs
+ )
+ ) {
return CompletableFuture.completedFuture(
RaftUtil.removeVoterResponse(
Errors.REQUEST_TIMED_OUT,
@@ -132,7 +172,8 @@ public final class RemoveVoterHandler {
);
}
- // Remove the voter from the set of voters
+ // Remove the voter from the set of voters, returning VOTER_NOT_FOUND if it isn't a
+ // current voter
Optional<VoterSet> newVoters = votersEntry.get().value().removeVoter(voterKey);
if (newVoters.isEmpty()) {
return CompletableFuture.completedFuture(
@@ -157,6 +198,17 @@ public final class RemoveVoterHandler {
return state.future();
}
+ /**
+ * Called when the high watermark advances to check if a pending remove voter operation has
+ * committed.
+ * <p>
+ * Once the high watermark advances past the offset of the appended VotersRecord, this
+ * completes the pending RemoveVoter response with success, and resigns the leadership if the
+ * leader itself is no longer part of the committed voter set.
+ *
+ * @param leaderState the leader state
+ * @param highWatermark the new high watermark offset
+ */
public void highWatermarkUpdated(LeaderState<?> leaderState, long highWatermark) {
var changeVoterState = leaderState.changeVoterState();
@@ -172,7 +224,7 @@ public final class RemoveVoterHandler {
ReplicaKey localKey = localReplicaKey.orElseThrow(
() -> new IllegalStateException(
String.format(
- "Leaders mush have an id and directory id %s",
+ "Leaders must have an id and directory id %s",
localReplicaKey
)
)
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandlerState.java b/raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandlerState.java
index bb9ef4cb2cc..ead356d3979 100644
--- a/raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandlerState.java
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/RemoveVoterHandlerState.java
@@ -20,7 +20,15 @@ import org.apache.kafka.common.message.RemoveRaftVoterResponseData;
import org.apache.kafka.common.utils.Timer;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+/**
+ * Tracks the state of a single pending remove voter operation.
+ * <p>
+ * An instance is created by {@link RemoveVoterHandler#handleRemoveVoterRequest} once the
+ * updated VotersRecord has been appended to the log, and is held by
+ * {@link ChangeVoterHandlerState} until the record commits or the operation expires.
+ */
public final class RemoveVoterHandlerState {
private final long lastOffset;
private final Timer timeout;
@@ -31,16 +39,37 @@ public final class RemoveVoterHandlerState {
this.timeout = timeout;
}
+ /**
+ * Returns the time in milliseconds until this operation expires.
+ *
+ * @param currentTimeMs the current time in milliseconds
+ * @return the remaining time in milliseconds until expiration
+ */
public long timeUntilOperationExpiration(long currentTimeMs) {
timeout.update(currentTimeMs);
return timeout.remainingMs();
}
- public CompletableFuture<RemoveRaftVoterResponseData> future() {
- return future;
+ /**
+ * Completes the future with the provided response.
+ *
+ * @param response the response to complete the future with
+ */
+ public void completeFuture(RemoveRaftVoterResponseData response) {
+ future.complete(response);
}
+ /**
+ * Returns the offset of the VotersRecord that was appended to the log for this remove voter
+ * operation.
+ *
+ * @return the offset of the appended VotersRecord
+ */
public long lastOffset() {
return lastOffset;
}
+
+ CompletionStage<RemoveRaftVoterResponseData> future() {
+ return future;
+ }
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java b/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java
index 962e35c6cb4..4d75f920839 100644
--- a/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandler.java
@@ -16,11 +16,16 @@
*/
package org.apache.kafka.raft.internals;
+import org.apache.kafka.common.Node;
import org.apache.kafka.common.feature.SupportedVersionRange;
+import org.apache.kafka.common.message.ApiVersionsRequestData;
+import org.apache.kafka.common.message.ApiVersionsResponseData;
import org.apache.kafka.common.message.UpdateRaftVoterRequestData;
import org.apache.kafka.common.message.UpdateRaftVoterResponseData;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.requests.ApiVersionsRequest;
+import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.internals.LogContext;
import org.apache.kafka.raft.Endpoints;
import org.apache.kafka.raft.LeaderState;
@@ -37,37 +42,97 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
/**
- * This type implements the protocol for updating a voter from a KRaft partition.
+ * This type implements the protocol for updating a voter's endpoints and supported kraft.version
+ * in a KRaft partition.
*
- * 1. Check that the leader has fenced the previous leader(s) by checking that the HWM is known,
+ * Unlike {@link AddVoterHandler} and {@link RemoveVoterHandler}, this operation is not restricted
+ * to clusters that support kraft.version 1: it can also update the leader's in-memory voter
+ * information for clusters still running kraft.version 0.
+ *
+ * Handling the UpdateVoter request ({@link #handleUpdateVoterRequest}) does the following:
+ *
+ * 1. Check that there are no other pending voter change operations (add, remove or update),
* otherwise return the REQUEST_TIMED_OUT error.
- * 2. Check that the cluster supports kraft.version 1, otherwise return the UNSUPPORTED_VERSION error.
- * 3. Check that there are no uncommitted voter changes, otherwise return the REQUEST_TIMED_OUT error.
- * 4. Check that the updated voter still supports the currently finalized kraft.version, otherwise
- * return the INVALID_REQUEST error.
- * 5. Check that the updated voter is still listening on the default listener.
- * 6. Update voter set with new voter configuration.
- * a. If reconfiguration is supported, append the updated VotersRecord to the log. The KRaft internal listener will read this
- * uncommitted record from the log and update the voter in the set of voters.
- * b. If reconfiguration is not supported, update the in-memory information for the voter. This will get
- * appended to the log when the cluster is upgraded to a kraft version that supports reconfiguration.
- * 7. Send the UpdateVoter successful response to the voter.
+ * 2. Check that the leader has established a HWM, otherwise return the REQUEST_TIMED_OUT error.
+ * 3. Check that the request's supported kraft.version range covers the cluster's currently
+ * finalized kraft.version, otherwise return the INVALID_REQUEST error.
+ * 4. Check that the updated endpoints include the default listener, otherwise return the
+ * INVALID_REQUEST error.
+ * 5. Send an API_VERSIONS request to the voter's new default endpoint and record the operation as
+ * pending, returning the REQUEST_TIMED_OUT error if the request cannot be sent.
+ *
+ * The response is not sent synchronously; the RPC only completes once the API_VERSIONS response is
+ * handled by {@link #handleApiVersionsResponse}:
+ *
+ * 6. Ignore the response if it doesn't come from the voter this operation is waiting on.
+ * 7. Abort with the REQUEST_TIMED_OUT error if the API_VERSIONS request itself failed.
+ * 8. Abort with the INVALID_REQUEST error if the supported kraft.version range in the response
+ * doesn't match the range from the UpdateVoter request.
+ * 9. Read the current voter set: from the last committed VotersRecord in the log if the cluster
+ * supports kraft.version 1, or from the leader's in-memory voter set otherwise (this in-memory
+ * state is used to carry voter updates across a future kraft.version upgrade). Abort with the
+ * REQUEST_TIMED_OUT error if neither is available yet, e.g. because the kraft.version was just
+ * upgraded but the corresponding record has not been written to the log.
+ * 10. Update the matching voter's entry with the new endpoints and supported kraft.version, aborting
+ * with the VOTER_NOT_FOUND error if the voter is no longer part of the set.
+ * 11. Store the updated voter set:
+ * a. If the cluster supports kraft.version 1, append the updated VotersRecord to the log and
+ * immediately complete the RPC with success, without waiting for the record to commit. The
+ * pending operation is only cleared once the HWM advances past the appended record (see
+ * {@link #highWatermarkUpdated}), which allows the next voter change operation to proceed.
+ * b. Otherwise, compare-and-set the leader's in-memory voter set, completing the RPC
+ * immediately with success, or with the REQUEST_TIMED_OUT error if the in-memory state was
+ * concurrently changed by another operation.
+ * In either successful case, this also resets the leader's check quorum tracking for the
+ * voter (see {@link LeaderState#updateCheckQuorumForFollowingVoter}), since a successful
+ * UpdateVoter request is evidence that the voter is following the leader.
+ *
+ * A pending operation that doesn't complete before its timeout expires is also aborted with the
+ * REQUEST_TIMED_OUT error, by {@link ChangeVoterHandlerState#maybeExpirePendingOperation}.
*/
public final class UpdateVoterHandler {
private final KRaftControlRecordStateMachine partitionState;
- private final ListenerName defaultListenerName;
- private final Logger log;
+ private final RequestSender requestSender;
+ private final Time time;
+ private final Logger logger;
+ /**
+ * Creates a new handler for update voter requests.
+ *
+ * @param partitionState the KRaft partition state, used to read the currently finalized
+ * kraft.version and the log's voter set
+ * @param requestSender used to send the API_VERSIONS request to the voter being updated
+ * @param time the time implementation, used to create the timer that bounds a pending operation
+ * @param logContext used to create this class's logger
+ */
public UpdateVoterHandler(
KRaftControlRecordStateMachine partitionState,
- ListenerName defaultListenerName,
+ RequestSender requestSender,
+ Time time,
LogContext logContext
) {
this.partitionState = partitionState;
- this.defaultListenerName = defaultListenerName;
- this.log = logContext.logger(getClass());
+ this.requestSender = requestSender;
+ this.time = time;
+ this.logger = logContext.logger(getClass());
}
+ /**
+ * Handle an UpdateVoter request.
+ * <p>
+ * See the class documentation for the full set of steps that this method and
+ * {@link #handleApiVersionsResponse} perform together.
+ *
+ * @param leaderState the leader state
+ * @param requestListenerName the listener the request was received on, used to build the response
+ * @param voterKey the id and directory id of the voter to update
+ * @param voterEndpoints the updated endpoints for the voter
+ * @param supportedKraftVersions the kraft.version range supported by the voter
+ * @param currentTimeMs the current time in milliseconds
+ * @return a future for the UpdateVoter response; it completes immediately if the request is
+ * rejected outright, or later, once the API_VERSIONS round trip finishes, via
+ * {@link #handleApiVersionsResponse}
+ */
public CompletionStage<UpdateRaftVoterResponseData> handleUpdateVoterRequest(
LeaderState<?> leaderState,
ListenerName requestListenerName,
@@ -76,8 +141,14 @@ public final class UpdateVoterHandler {
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(
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ currentTimeMs
+ )
+ ) {
return CompletableFuture.completedFuture(
RaftUtil.updateVoterResponse(
Errors.REQUEST_TIMED_OUT,
@@ -89,8 +160,7 @@ public final class UpdateVoterHandler {
}
// Check that the leader has established a HWM and committed the current epoch
- Optional<Long> highWatermark = leaderState.highWatermark().map(LogOffsetMetadata::offset);
- if (highWatermark.isEmpty()) {
+ if (leaderState.highWatermark().isEmpty()) {
return CompletableFuture.completedFuture(
RaftUtil.updateVoterResponse(
Errors.REQUEST_TIMED_OUT,
@@ -101,53 +171,39 @@ public final class UpdateVoterHandler {
);
}
- // 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 endpoints includes the default listener
+ if (voterEndpoints.address(requestSender.listenerName()).isEmpty()) {
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)) {
+
+ // 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()
@@ -155,16 +211,192 @@ public final class UpdateVoterHandler {
);
}
- // Check that endpoints includes the default listener
- if (voterEndpoints.address(defaultListenerName).isEmpty()) {
- return CompletableFuture.completedFuture(
- RaftUtil.updateVoterResponse(
- Errors.INVALID_REQUEST,
- 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 a pending update voter operation.
+ * <p>
+ * This may abort the pending operation, completing its future with an error, if the response
+ * doesn't come from the expected voter, if the API_VERSIONS request failed, if the supported
+ * kraft.version range doesn't match the one from the UpdateVoter request, if the current voter
+ * set can't be read yet, or if the voter is no longer part of the voter set. Otherwise, it
+ * stores the updated voter set, see {@link #storeUpdatedVoters}.
+ *
+ * @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 false only when the API_VERSIONS request itself failed, which is the only case where
+ * the caller (see {@code KafkaRaftClient#handleResponse}) should treat this as an
+ * unsuccessful response for request-tracking purposes; true otherwise, including when
+ * this method aborts the pending update voter operation for another reason
+ */
+ 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 is no pending update operation; just ignore the API_VERSIONS response
+ return true;
+ }
+
+ // Check that the API_VERSIONS response matches the id of the voter getting updated
+ 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 request
+ 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;
+ }
+
+ completeUpdateVoter(leaderState, changeVoterState, current, currentTimeMs);
+ return true;
+ }
+
+ /**
+ * Validates the kraft.version range and current voter set, then applies the update, once the
+ * API_VERSIONS response has already been matched to a pending update voter operation.
+ */
+ private void completeUpdateVoter(
+ LeaderState<?> leaderState,
+ ChangeVoterHandlerState changeVoterState,
+ UpdateVoterHandlerState current,
+ long currentTimeMs
+ ) {
+ var kraftVersion = partitionState.lastKraftVersion();
+ if (!validVersionRange(kraftVersion, current.supportedKraftVersions())) {
+ // Check that the supported version range is valid
+ logger.info(
+ "Aborting update voter operation for {} at {} since kraft.version range {} doesn't match {}",
+ current.voterKey(),
+ current.voterEndpoints(),
+ current.supportedKraftVersions(),
+ kraftVersion
+ );
+
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.INVALID_REQUEST,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
+ );
+ return;
+ }
+
+ // 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 HWM is known
+ throw new IllegalStateException("Expected the high-watermark to be known");
+ }
+
+ // Read the voter set from the log if the cluster supports kraft.version 1, otherwise from
+ // the leader's in-memory voter set
+ final Optional<KRaftVersionUpgrade.Voters> inMemoryVoters;
+ final Optional<VoterSet> voters;
+ if (kraftVersion.isReconfigSupported()) {
+ inMemoryVoters = Optional.empty();
+
+ // Only use the last voter set entry if it has already committed; an uncommitted
+ // entry is treated as not yet available
+ 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()) {
+ /* This can happen for two reasons:
+ *
+ * 1. The cluster just upgraded to a kraft.version that supports reconfiguration, but
+ * the updated kraft.version and voter set have only been written to the batch
+ * accumulator, not yet the log. During this time the leader's volatile voter set
+ * has already been cleared, but the log doesn't have a committed VotersRecord yet.
+ * The KRaft replica's partition state is only updated once the control record has
+ * been written to the log (disk).
+ * 2. The cluster already supports kraft.version 1, but the last VotersRecord in the
+ * log has not committed yet.
+ *
+ * In both cases the leader doesn't have a definitive voter set to update yet, so ask
+ * the voter to retry.
+ */
+ 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()
+ );
+ return;
}
// Update the voter
@@ -172,41 +404,38 @@ public final class UpdateVoterHandler {
voters.get(),
kraftVersion,
VoterSet.VoterNode.of(
- voterKey,
- voterEndpoints,
- new SupportedVersionRange(
- supportedKraftVersions.minSupportedVersion(),
- supportedKraftVersions.maxSupportedVersion()
- )
+ current.voterKey(),
+ current.voterEndpoints(),
+ current.supportedKraftVersions()
)
);
if (updatedVoters.isEmpty()) {
- return CompletableFuture.completedFuture(
- RaftUtil.updateVoterResponse(
- Errors.VOTER_NOT_FOUND,
- requestListenerName,
- leaderState.leaderAndEpoch(),
- leaderState.leaderEndpoints()
- )
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.VOTER_NOT_FOUND,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
);
+
+ return;
}
- return storeUpdatedVoters(
+ storeUpdatedVoters(
leaderState,
- voterKey,
+ current,
inMemoryVoters,
updatedVoters.get(),
- requestListenerName,
currentTimeMs
);
+ return;
}
private boolean validVersionRange(
KRaftVersion finalizedVersion,
- UpdateRaftVoterRequestData.KRaftVersionFeature supportedKraftVersions
+ SupportedVersionRange supportedKraftVersions
) {
- return supportedKraftVersions.minSupportedVersion() <= finalizedVersion.featureLevel() &&
- supportedKraftVersions.maxSupportedVersion() >= finalizedVersion.featureLevel();
+ return supportedKraftVersions.min() <= finalizedVersion.featureLevel() &&
+ supportedKraftVersions.max() >= finalizedVersion.featureLevel();
}
private Optional<VoterSet> updateVoters(
@@ -219,17 +448,34 @@ public final class UpdateVoterHandler {
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 supports reconfig, write the updated 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));
+
+ // Reset the check quorum state since the leader received a successful request
+ leaderState.updateCheckQuorumForFollowingVoter(current.voterKey(), 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(
@@ -237,38 +483,85 @@ public final class UpdateVoterHandler {
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 so the client can retry
+ 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 advances to check if a pending update voter operation can be
+ * cleared.
+ * <p>
+ * When the cluster supports kraft.version 1, {@link #storeUpdatedVoters} completes the
+ * UpdateVoter RPC as soon as the updated VotersRecord is appended to the log, without waiting
+ * for it to commit, but leaves the operation marked as pending so that no other voter change
+ * operation can start. This method clears that pending state, allowing the next voter change
+ * request to be accepted, once the high watermark advances past the offset of that record.
+ *
+ * @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) {
+ // The VotersRecord with the updated voter was committed; clear the
+ // pending operation to allow other voter changes. The RPC response was
+ // already sent when the record was appended, so this does not complete
+ // the future again.
+ changeVoterState.resetUpdateVoterHandlerState(
+ Errors.NONE,
+ leaderState.leaderAndEpoch(),
+ leaderState.leaderEndpoints(),
+ Optional.empty()
+ );
+ }
+ });
+ });
+ }
}
diff --git a/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandlerState.java b/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandlerState.java
new file mode 100644
index 00000000000..6979bdc463b
--- /dev/null
+++ b/raft/src/main/java/org/apache/kafka/raft/internals/UpdateVoterHandlerState.java
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.feature.SupportedVersionRange;
+import org.apache.kafka.common.message.UpdateRaftVoterResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.Timer;
+import org.apache.kafka.raft.Endpoints;
+import org.apache.kafka.raft.ReplicaKey;
+
+import java.util.OptionalLong;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+
+/**
+ * Tracks the state of a single pending update voter operation.
+ * <p>
+ * An instance is created by {@link UpdateVoterHandler#handleUpdateVoterRequest} once the
+ * API_VERSIONS request has been sent to the voter being updated, and is held by
+ * {@link ChangeVoterHandlerState} until the operation completes, is aborted, or expires.
+ */
+public final class UpdateVoterHandlerState {
+ private final ReplicaKey voterKey;
+ private final Endpoints voterEndpoints;
+ private final ListenerName requestListenerName;
+ private final SupportedVersionRange supportedKraftVersions;
+ private final Timer timeout;
+ private final CompletableFuture<UpdateRaftVoterResponseData> future = new CompletableFuture<>();
+
+ private OptionalLong lastOffset = OptionalLong.empty();
+
+ UpdateVoterHandlerState(
+ ReplicaKey voterKey,
+ Endpoints voterEndpoints,
+ ListenerName requestListenerName,
+ SupportedVersionRange supportedKraftVersions,
+ Timer timeout
+ ) {
+ this.voterKey = voterKey;
+ this.voterEndpoints = voterEndpoints;
+ this.requestListenerName = requestListenerName;
+ this.supportedKraftVersions = supportedKraftVersions;
+ this.timeout = timeout;
+ }
+
+ /**
+ * Returns the time in milliseconds until this operation expires.
+ *
+ * @param currentTimeMs the current time in milliseconds
+ * @return the remaining time in milliseconds until expiration
+ */
+ public long timeUntilOperationExpiration(long currentTimeMs) {
+ timeout.update(currentTimeMs);
+ return timeout.remainingMs();
+ }
+
+ /**
+ * Checks whether this handler state is expecting an API_VERSIONS response from the given replica.
+ *
+ * @param replicaId the replica id to check
+ * @return true if expecting a response from this replica, false otherwise
+ */
+ public boolean expectingApiResponse(int replicaId) {
+ return lastOffset.isEmpty() && replicaId == voterKey.id();
+ }
+
+ /**
+ * Sets the last offset for this update voter operation.
+ *
+ * @param lastOffset the offset of the VotersRecord that was appended to the log
+ * @throws IllegalStateException if the last offset has already been set
+ */
+ public void setLastOffset(long lastOffset) {
+ if (this.lastOffset.isPresent()) {
+ throw new IllegalStateException(
+ String.format(
+ "Cannot override last offset to %s for updating voter %s because it is " +
+ "already set to %s",
+ lastOffset,
+ voterKey,
+ this.lastOffset
+ )
+ );
+ }
+
+ this.lastOffset = OptionalLong.of(lastOffset);
+ }
+
+ /**
+ * Returns the voter key for the voter being updated.
+ *
+ * @return the voter key
+ */
+ public ReplicaKey voterKey() {
+ return voterKey;
+ }
+
+ /**
+ * Returns the endpoints for the voter being updated.
+ *
+ * @return the voter endpoints
+ */
+ public Endpoints voterEndpoints() {
+ return voterEndpoints;
+ }
+
+ /**
+ * Returns the listener name from the update voter request.
+ *
+ * @return the listener name
+ */
+ public ListenerName requestListenerName() {
+ return requestListenerName;
+ }
+
+ /**
+ * Returns the kraft version range supported by the voter being updated.
+ *
+ * @return the supported kraft version range
+ */
+ public SupportedVersionRange supportedKraftVersions() {
+ return supportedKraftVersions;
+ }
+
+ /**
+ * Returns the offset of the VotersRecord if it has been appended to the log.
+ *
+ * @return the last offset, or empty if not yet appended
+ */
+ public OptionalLong lastOffset() {
+ return lastOffset;
+ }
+
+ /**
+ * Completes the future with the provided response.
+ *
+ * @param response the response to complete the future with
+ */
+ public void completeFuture(UpdateRaftVoterResponseData response) {
+ future.complete(response);
+ }
+
+ CompletionStage<UpdateRaftVoterResponseData> future() {
+ return future;
+ }
+}
diff --git a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
index 08f3d63833b..9c20a09b051 100644
--- a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
@@ -350,7 +350,7 @@ public class KafkaRaftClientReconfigTest {
// Attempt to add new voter to the quorum
context.deliverRequest(context.addVoterRequest(Integer.MAX_VALUE, newVoter, newListeners));
- completeApiVersionsForAddVoter(context, newVoter, newAddress);
+ completeApiVersions(context, newVoter, newAddress);
// Handle the API_VERSIONS response
context.poll();
@@ -402,7 +402,7 @@ public class KafkaRaftClientReconfigTest {
).setAckWhenCommitted(false)
);
- completeApiVersionsForAddVoter(context, newVoter, newAddress);
+ completeApiVersions(context, newVoter, newAddress);
// Handle the API_VERSIONS response
context.poll();
@@ -442,16 +442,16 @@ public class KafkaRaftClientReconfigTest {
checkLeaderMetricValues(2, 1, 0, context);
}
- private void completeApiVersionsForAddVoter(
+ private void completeApiVersions(
RaftClientTestContext context,
- ReplicaKey newVoter,
- InetSocketAddress newAddress
+ ReplicaKey remoteVoter,
+ InetSocketAddress remoteAddress
) throws Exception {
- // Leader should send an API_VERSIONS request to the new voter's endpoint
+ // Leader should send an API_VERSIONS request to the remote voter's endpoint
context.pollUntilRequest();
RaftRequest.Outbound apiVersionRequest = context.assertSentApiVersionsRequest();
assertEquals(
- new Node(newVoter.id(), newAddress.getHostString(), newAddress.getPort()),
+ new Node(remoteVoter.id(), remoteAddress.getHostString(), remoteAddress.getPort()),
apiVersionRequest.destination()
);
@@ -1665,11 +1665,11 @@ public class KafkaRaftClientReconfigTest {
// Attempt to update the follower
InetSocketAddress defaultAddress = InetSocketAddress.createUnresolved(
- "localhost",
- 9990 + follower.id()
+ "new_host",
+ 7770 + follower.id()
);
InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
- "localhost",
+ "new_host",
8990 + follower.id()
);
HashMap<ListenerName, InetSocketAddress> listenersMap = new HashMap<>(2);
@@ -1679,11 +1679,13 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
follower,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
newListeners
)
);
+ completeApiVersions(context, follower, defaultAddress);
+
// Expect reply for UpdateVoter request without committing the record
context.pollUntilResponse();
context.assertSentUpdateVoterResponse(
@@ -1696,6 +1698,250 @@ public class KafkaRaftClientReconfigTest {
assertTrue(context.client.quorum().isVoter(follower));
}
+ @Test
+ void testUpdateVoterResetsCheckQuorumTimer() throws Exception {
+ ReplicaKey local = replicaKey(randomReplicaId(), true);
+ ReplicaKey follower1 = replicaKey(local.id() + 1, true);
+ ReplicaKey follower2 = replicaKey(local.id() + 2, true);
+
+ VoterSet voters = VoterSetTest.voterSet(Stream.of(local, follower1, follower2));
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(local.id(), local.directoryId().get())
+ .withKip853Rpc(true)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+ int resignLeadershipTimeout = context.checkQuorumTimeoutMs;
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ // Establish a HWM; this also resets the check quorum timer
+ context.deliverRequest(
+ context.fetchRequest(epoch, follower1, context.log.endOffset().offset(), epoch, 0)
+ );
+ context.pollUntilResponse();
+ context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(local.id()));
+
+ // check quorum timeout has not expired, the leader should not resign
+ context.time.sleep(resignLeadershipTimeout / 2);
+ context.poll();
+ assertFalse(context.client.quorum().isResigned());
+
+ // The leader periodically re-sends BeginQuorumEpoch heartbeats; drain and answer them
+ // so the connection to follower2 is free for the UpdateVoter's ApiVersions probe below
+ for (RaftRequest.Outbound request : context.collectBeginEpochRequests(epoch)) {
+ context.deliverResponse(
+ request.correlationId(),
+ request.destination(),
+ context.beginEpochResponse(epoch, local.id())
+ );
+ }
+ context.poll();
+
+ // follower2 successfully completes an UpdateVoter request; this should reset the
+ // check quorum timer even though follower2 never sends a FETCH request
+ InetSocketAddress defaultAddress = InetSocketAddress.createUnresolved(
+ "new_host",
+ 7770 + follower2.id()
+ );
+ Endpoints newListeners = Endpoints.fromInetSocketAddresses(
+ Map.of(context.channel.listenerName(), defaultAddress)
+ );
+ context.deliverRequest(
+ context.updateVoterRequest(
+ follower2,
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
+ newListeners
+ )
+ );
+ completeApiVersions(context, follower2, defaultAddress);
+ context.pollUntilResponse();
+ context.assertSentUpdateVoterResponse(Errors.NONE, OptionalInt.of(local.id()), epoch);
+
+ // If the UpdateVoter request had not reset the check quorum timer, the leader would
+ // resign once the timeout from the original FETCH-based reset elapses. Sleeping past
+ // that point here, with no further FETCH request from either follower, verifies the
+ // reset actually happened: the leader should still not resign.
+ context.time.sleep((resignLeadershipTimeout / 2) + 2);
+ context.poll();
+ assertFalse(context.client.quorum().isResigned());
+
+ // Confirm the timer was actually reset and not simply broken: without any further
+ // FETCH or UpdateVoter request, sleeping past the (reset) timeout now causes the
+ // leader to resign
+ context.time.sleep(resignLeadershipTimeout);
+ context.poll();
+ assertTrue(context.client.quorum().isResigned());
+ context.assertResignedLeader(epoch, local.id());
+ }
+
+ @Test
+ void testUpdateVoterWithUnreachableListener() throws Exception {
+ ReplicaKey local = replicaKey(randomReplicaId(), true);
+ ReplicaKey follower = replicaKey(local.id() + 1, true);
+
+ VoterSet voters = VoterSetTest.voterSet(Stream.of(local, follower));
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(local.id(), local.directoryId().get())
+ .withRaftProtocol(RaftProtocol.KIP_1186_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ // Establish a HWM
+ context.deliverRequest(
+ context.fetchRequest(epoch, follower, context.log.endOffset().offset(), epoch, 0)
+ );
+ context.pollUntilResponse();
+ context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(local.id()));
+
+ // Attempt to update voter with new listeners
+ context.deliverRequest(
+ context.updateVoterRequest(
+ follower,
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
+ voters.listeners(follower.id())
+ )
+ );
+
+ // Leader sends API_VERSIONS to verify reachability
+ context.pollUntilRequest();
+ context.assertSentApiVersionsRequest();
+
+ // Simulate unreachable voter by not responding and letting it timeout
+ context.time.sleep(context.requestTimeoutMs());
+ context.pollUntilResponse();
+
+ // UpdateVoter should fail since voter is unreachable
+ context.assertSentUpdateVoterResponse(
+ Errors.REQUEST_TIMED_OUT,
+ OptionalInt.of(local.id()),
+ epoch
+ );
+
+ // Voter set should not be updated
+ assertEquals(voters, context.listener.lastCommittedVoterSet().get());
+ }
+
+ @Test
+ void testUpdateVoterWithBrokerNotAvailable() throws Exception {
+ ReplicaKey local = replicaKey(randomReplicaId(), true);
+ ReplicaKey follower = replicaKey(local.id() + 1, true);
+
+ VoterSet voters = VoterSetTest.voterSet(Stream.of(local, follower));
+ InetSocketAddress defaultAddress = InetSocketAddress.createUnresolved("localhost", 9990 + follower.id());
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(local.id(), local.directoryId().get())
+ .withRaftProtocol(RaftProtocol.KIP_1186_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ // Establish a HWM
+ context.deliverRequest(
+ context.fetchRequest(epoch, follower, context.log.endOffset().offset(), epoch, 0)
+ );
+ context.pollUntilResponse();
+ context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(local.id()));
+
+ // Attempt to update voter with new listeners
+ context.deliverRequest(
+ context.updateVoterRequest(
+ follower,
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
+ voters.listeners(follower.id())
+ )
+ );
+
+ // Leader sends API_VERSIONS to verify reachability
+ context.pollUntilRequest();
+ RaftRequest.Outbound apiVersionRequest = context.assertSentApiVersionsRequest();
+ assertEquals(
+ new Node(follower.id(), defaultAddress.getHostString(), defaultAddress.getPort()),
+ apiVersionRequest.destination()
+ );
+
+ // Reply with API_VERSIONS response indicating broker not available
+ context.deliverResponse(
+ apiVersionRequest.correlationId(),
+ apiVersionRequest.destination(),
+ apiVersionsResponse(Errors.BROKER_NOT_AVAILABLE)
+ );
+ context.pollUntilResponse();
+
+ // UpdateVoter should fail since voter is not available
+ context.assertSentUpdateVoterResponse(
+ Errors.REQUEST_TIMED_OUT,
+ OptionalInt.of(local.id()),
+ epoch
+ );
+
+ // Voter set should not be updated
+ assertEquals(voters, context.listener.lastCommittedVoterSet().get());
+ }
+
+ @Test
+ void testUpdateVoterWithUnsupportedKraftVersion() throws Exception {
+ ReplicaKey local = replicaKey(randomReplicaId(), true);
+ ReplicaKey follower = replicaKey(local.id() + 1, true);
+
+ VoterSet voters = VoterSetTest.voterSet(Stream.of(local, follower));
+ InetSocketAddress defaultAddress = InetSocketAddress.createUnresolved("localhost", 9990 + follower.id());
+
+ RaftClientTestContext context = new RaftClientTestContext.Builder(local.id(), local.directoryId().get())
+ .withRaftProtocol(RaftProtocol.KIP_1186_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ // Establish a HWM
+ context.deliverRequest(
+ context.fetchRequest(epoch, follower, context.log.endOffset().offset(), epoch, 0)
+ );
+ context.pollUntilResponse();
+ context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(local.id()));
+
+ // Attempt to update voter claiming support for kraft.version 1
+ context.deliverRequest(
+ context.updateVoterRequest(
+ follower,
+ new SupportedVersionRange((short) 0, (short) 1),
+ voters.listeners(follower.id())
+ )
+ );
+
+ // Leader sends API_VERSIONS to verify the voter's capabilities
+ context.pollUntilRequest();
+ RaftRequest.Outbound apiVersionRequest = context.assertSentApiVersionsRequest();
+ assertEquals(
+ new Node(follower.id(), defaultAddress.getHostString(), defaultAddress.getPort()),
+ apiVersionRequest.destination()
+ );
+
+ // Reply with API_VERSIONS showing only kraft.version 0 support
+ context.deliverResponse(
+ apiVersionRequest.correlationId(),
+ apiVersionRequest.destination(),
+ apiVersionsResponse(Errors.NONE, new SupportedVersionRange((short) 0))
+ );
+ context.pollUntilResponse();
+ context.assertSentUpdateVoterResponse(
+ Errors.INVALID_REQUEST,
+ OptionalInt.of(local.id()),
+ epoch
+ );
+ }
+
@Test
void testLeaderUpdatesVoter() throws Exception {
ReplicaKey local = replicaKey(randomReplicaId(), true);
@@ -1737,7 +1983,7 @@ public class KafkaRaftClientReconfigTest {
VoterSet.VoterNode.of(
local,
localListeners,
- Feature.KRAFT_VERSION.supportedVersionRange()
+ Feature.KRAFT_VERSION.supportedVersionRange(true)
)
);
assertEquals(updatedVoterSet, context.listener.lastCommittedVoterSet());
@@ -1765,7 +2011,7 @@ public class KafkaRaftClientReconfigTest {
"",
follower,
epoch,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
Endpoints.empty()
)
);
@@ -1782,7 +2028,7 @@ public class KafkaRaftClientReconfigTest {
"invalid-uuid",
follower,
epoch,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
Endpoints.empty()
)
);
@@ -1810,11 +2056,18 @@ public class KafkaRaftClientReconfigTest {
context.unattachedToLeader();
int epoch = context.currentEpoch();
+ // Establish a HWM and commit the latest voter set
+ context.deliverRequest(
+ context.fetchRequest(epoch, follower, context.log.endOffset().offset(), epoch, 0)
+ );
+ context.pollUntilResponse();
+ context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(local.id()));
+
// missing directory id
context.deliverRequest(
context.updateVoterRequest(
ReplicaKey.of(follower.id(), Uuid.ZERO_UUID),
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
Endpoints.empty()
)
);
@@ -1829,7 +2082,7 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
follower,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
Endpoints.empty()
)
);
@@ -1862,7 +2115,7 @@ public class KafkaRaftClientReconfigTest {
context.clusterId,
follower,
epoch - 1,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
Endpoints.empty()
)
);
@@ -1895,7 +2148,7 @@ public class KafkaRaftClientReconfigTest {
context.clusterId,
follower,
epoch + 1,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
Endpoints.empty()
)
);
@@ -1924,7 +2177,7 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
follower,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
Endpoints.empty()
)
);
@@ -1968,7 +2221,7 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
follower,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
newListeners
)
);
@@ -2019,10 +2272,13 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
follower,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
newListeners
)
);
+
+ completeApiVersions(context, follower, defaultAddress);
+
context.pollUntilResponse();
context.assertSentUpdateVoterResponse(
Errors.NONE,
@@ -2067,13 +2323,17 @@ public class KafkaRaftClientReconfigTest {
listenersMap.put(context.channel.listenerName(), defaultAddress);
listenersMap.put(ListenerName.normalised("ANOTHER_LISTENER"), newAddress);
Endpoints newListeners = Endpoints.fromInetSocketAddresses(listenersMap);
+ var notVoter = replicaKey(follower.id(), true);
context.deliverRequest(
context.updateVoterRequest(
- replicaKey(follower.id(), true),
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ notVoter,
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
newListeners
)
);
+
+ completeApiVersions(context, notVoter, defaultAddress);
+
context.pollUntilResponse();
context.assertSentUpdateVoterResponse(
Errors.VOTER_NOT_FOUND,
@@ -2118,13 +2378,17 @@ public class KafkaRaftClientReconfigTest {
listenersMap.put(context.channel.listenerName(), defaultAddress);
listenersMap.put(ListenerName.normalised("ANOTHER_LISTENER"), newAddress);
Endpoints newListeners = Endpoints.fromInetSocketAddresses(listenersMap);
+ var notVoter = ReplicaKey.of(follower.id() + 1, follower.directoryId().get());
context.deliverRequest(
context.updateVoterRequest(
- ReplicaKey.of(follower.id() + 1, follower.directoryId().get()),
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ notVoter,
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
newListeners
)
);
+
+ completeApiVersions(context, notVoter, defaultAddress);
+
context.pollUntilResponse();
context.assertSentUpdateVoterResponse(
Errors.VOTER_NOT_FOUND,
@@ -2191,7 +2455,7 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
follower,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
newListeners
)
);
@@ -2237,7 +2501,7 @@ public class KafkaRaftClientReconfigTest {
RaftRequest.Outbound updateRequest = context.assertSentUpdateVoterRequest(
local,
epoch,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
localListeners
);
context.deliverResponse(
@@ -2290,7 +2554,7 @@ public class KafkaRaftClientReconfigTest {
RaftRequest.Outbound updateRequest = context.assertSentUpdateVoterRequest(
local,
epoch,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
localListeners
);
context.deliverResponse(
@@ -2361,7 +2625,7 @@ public class KafkaRaftClientReconfigTest {
RaftRequest.Outbound updateRequest = context.assertSentUpdateVoterRequest(
local,
epoch,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
localListeners
);
context.deliverResponse(
@@ -2389,7 +2653,7 @@ public class KafkaRaftClientReconfigTest {
updateRequest = context.assertSentUpdateVoterRequest(
local,
newEpoch,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
localListeners
);
context.deliverResponse(
@@ -2438,10 +2702,17 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
voter,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
startingVoters.listeners(voter.id())
)
);
+
+ completeApiVersions(
+ context,
+ voter,
+ startingVoters.listeners(voter.id()).address(context.channel.listenerName()).get()
+ );
+
context.pollUntilResponse();
context.assertSentUpdateVoterResponse(
Errors.NONE,
@@ -2509,10 +2780,17 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
voter,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
startingVoters.listeners(voter.id())
)
);
+
+ completeApiVersions(
+ context,
+ voter,
+ startingVoters.listeners(voter.id()).address(context.channel.listenerName()).get()
+ );
+
context.pollUntilResponse();
context.assertSentUpdateVoterResponse(
Errors.NONE,
@@ -2555,10 +2833,17 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
voter1,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
newVoter1Listeners
)
);
+
+ completeApiVersions(
+ context,
+ voter1,
+ newVoter1Listeners.address(context.channel.listenerName()).get()
+ );
+
context.pollUntilResponse();
context.assertSentUpdateVoterResponse(
Errors.NONE,
@@ -2622,10 +2907,17 @@ public class KafkaRaftClientReconfigTest {
context.deliverRequest(
context.updateVoterRequest(
voter1,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
startingVoters.listeners(voter1.id())
)
);
+
+ completeApiVersions(
+ context,
+ voter1,
+ startingVoters.listeners(voter1.id()).address(context.channel.listenerName()).get()
+ );
+
context.pollUntilResponse();
context.assertSentUpdateVoterResponse(
Errors.NONE,
@@ -2726,7 +3018,7 @@ public class KafkaRaftClientReconfigTest {
context.assertSentUpdateVoterRequest(
local,
epoch,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
localListeners
);
}
@@ -2765,7 +3057,7 @@ public class KafkaRaftClientReconfigTest {
RaftRequest.Outbound updateRequest = context.assertSentUpdateVoterRequest(
local,
epoch,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
localListeners
);
context.deliverResponse(
@@ -2953,7 +3245,7 @@ public class KafkaRaftClientReconfigTest {
}
private static ApiVersionsResponseData apiVersionsResponse(Errors error) {
- return apiVersionsResponse(error, Feature.KRAFT_VERSION.supportedVersionRange());
+ return apiVersionsResponse(error, Feature.KRAFT_VERSION.supportedVersionRange(true));
}
private static ApiVersionsResponseData apiVersionsResponse(Errors error, SupportedVersionRange supportedVersions) {
diff --git a/raft/src/test/java/org/apache/kafka/raft/QuorumStateTest.java b/raft/src/test/java/org/apache/kafka/raft/QuorumStateTest.java
index 6c3418d64e8..04c53a6a4d6 100644
--- a/raft/src/test/java/org/apache/kafka/raft/QuorumStateTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/QuorumStateTest.java
@@ -83,7 +83,7 @@ public class QuorumStateTest {
localDirectoryId,
mockPartitionState,
localId.isPresent() ? voterSet.listeners(localId.getAsInt()) : Endpoints.empty(),
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
electionTimeoutMs,
fetchTimeoutMs,
store,
diff --git a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java
index 3852921f7e1..66297bfa9a6 100644
--- a/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/RaftEventSimulationTest.java
@@ -20,14 +20,19 @@ import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.feature.Features;
+import org.apache.kafka.common.feature.SupportedVersionRange;
import org.apache.kafka.common.memory.MemoryPool;
import org.apache.kafka.common.message.ApiMessageType;
+import org.apache.kafka.common.message.ApiVersionsResponseData;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.ObjectSerializationCache;
import org.apache.kafka.common.protocol.Readable;
import org.apache.kafka.common.protocol.Writable;
import org.apache.kafka.common.protocol.types.Type;
+import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.internals.BufferSupplier;
@@ -36,6 +41,7 @@ import org.apache.kafka.raft.MockLog.LogBatch;
import org.apache.kafka.raft.MockLog.LogEntry;
import org.apache.kafka.raft.internals.BatchMemoryPool;
import org.apache.kafka.server.common.Feature;
+import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.server.common.serialization.RecordSerde;
import org.apache.kafka.snapshot.RecordsSnapshotReader;
import org.apache.kafka.snapshot.SnapshotReader;
@@ -59,6 +65,8 @@ import java.util.OptionalLong;
import java.util.PriorityQueue;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
@@ -846,7 +854,7 @@ public class RaftEventSimulationTest {
clusterId,
List.of(),
endpointsFromId(nodeId, channel.listenerName()),
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
logContext,
random,
quorumConfig
@@ -919,6 +927,40 @@ public class RaftEventSimulationTest {
}
}
+ CompletionStage<RaftResponse.Outbound> handle(RaftRequest.Inbound request) {
+ var apiKey = ApiKeys.forId(request.data().apiKey());
+ switch (apiKey) {
+ case API_VERSIONS:
+ // KRaft doesn't handle ApiVersions request. These are handle by the "node".
+ // This creates a simple implementation that simply replies with the
+ // supported kraft versions.
+ var apiVersionsResponse = new ApiVersionsResponse.Builder()
+ .setSupportedFeatures(
+ Features.supportedFeatures(
+ Map.of(
+ KRaftVersion.FEATURE_NAME,
+ new SupportedVersionRange(
+ KRaftVersion.KRAFT_VERSION_0.featureLevel(),
+ KRaftVersion.LATEST_PRODUCTION.featureLevel()
+ )
+ )
+ )
+ )
+ .setApiVersions(new ApiVersionsResponseData.ApiVersionCollection())
+ .setFinalizedFeatures(Map.of())
+ .build()
+ .data();
+ var apiVersions = new RaftResponse.Outbound(
+ request.correlationId(),
+ apiVersionsResponse
+ );
+ return CompletableFuture.completedFuture(apiVersions);
+
+ default:
+ return client.handle(request);
+ }
+ }
+
long highWatermark() {
return client.quorum().highWatermark()
.map(LogOffsetMetadata::offset)
@@ -1349,7 +1391,7 @@ public class RaftEventSimulationTest {
cluster.nodeIfRunning(destination.id()).ifPresent(node -> {
inflight.put(correlationId, new InflightRequest(senderId, destination));
- node.client.handle(inbound).whenComplete((response, exception) -> {
+ node.handle(inbound).whenComplete((response, exception) -> {
if (response != null && filters.get(destination.id()).acceptOutbound(response)) {
deliver(response);
}
diff --git a/raft/src/test/java/org/apache/kafka/raft/internals/AddVoterHandlerStateTest.java b/raft/src/test/java/org/apache/kafka/raft/internals/AddVoterHandlerStateTest.java
new file mode 100644
index 00000000000..ddfc6c2139f
--- /dev/null
+++ b/raft/src/test/java/org/apache/kafka/raft/internals/AddVoterHandlerStateTest.java
@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.message.AddRaftVoterResponseData;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.raft.Endpoints;
+import org.apache.kafka.raft.ReplicaKey;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.OptionalLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class AddVoterHandlerStateTest {
+
+ @Test
+ public void testSetLastOffsetOnce() {
+ var time = new MockTime();
+ var state = new AddVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ true,
+ time.timer(1000)
+ );
+
+ assertTrue(state.lastOffset().isEmpty());
+ state.setLastOffset(100L);
+ assertEquals(OptionalLong.of(100L), state.lastOffset());
+ }
+
+ @Test
+ public void testCannotOverrideLastOffset() {
+ var time = new MockTime();
+ var state = new AddVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ true,
+ time.timer(1000)
+ );
+
+ state.setLastOffset(100L);
+ assertThrows(IllegalStateException.class, () -> state.setLastOffset(200L));
+ }
+
+ @Test
+ public void testExpectingApiResponseBeforeLastOffset() {
+ var time = new MockTime();
+ var state = new AddVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ true,
+ time.timer(1000)
+ );
+
+ assertTrue(state.expectingApiResponse(1));
+ assertFalse(state.expectingApiResponse(2));
+
+ state.setLastOffset(100L);
+ assertFalse(state.expectingApiResponse(1));
+ }
+
+ @Test
+ public void testTimeUntilExpiration() {
+ var time = new MockTime();
+ var state = new AddVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ true,
+ time.timer(1000)
+ );
+
+ assertEquals(1000, state.timeUntilOperationExpiration(time.milliseconds()));
+ time.sleep(500);
+ assertEquals(500, state.timeUntilOperationExpiration(time.milliseconds()));
+ time.sleep(500);
+ assertEquals(0, state.timeUntilOperationExpiration(time.milliseconds()));
+ }
+
+ @Test
+ public void testCompleteFuture() {
+ var time = new MockTime();
+ var state = new AddVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ true,
+ time.timer(1000)
+ );
+
+ var future = state.future().toCompletableFuture();
+ assertFalse(future.isDone());
+
+ var response = new AddRaftVoterResponseData().setErrorCode((short) 0);
+ state.completeFuture(response);
+
+ assertTrue(future.isDone());
+ assertEquals(response, future.join());
+ }
+
+ @Test
+ public void testGetters() {
+ var time = new MockTime();
+ var voterKey = ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID);
+ var endpoints = Endpoints.empty();
+
+ var state = new AddVoterHandlerState(voterKey, endpoints, true, time.timer(1000));
+
+ assertEquals(voterKey, state.voterKey());
+ assertEquals(endpoints, state.voterEndpoints());
+ assertTrue(state.ackWhenCommitted());
+
+ var stateNoAck = new AddVoterHandlerState(voterKey, endpoints, false, time.timer(1000));
+ assertFalse(stateNoAck.ackWhenCommitted());
+ }
+}
diff --git a/raft/src/test/java/org/apache/kafka/raft/internals/ChangeVoterHandlerStateTest.java b/raft/src/test/java/org/apache/kafka/raft/internals/ChangeVoterHandlerStateTest.java
index 850f6a8b714..1b03a88aac1 100644
--- a/raft/src/test/java/org/apache/kafka/raft/internals/ChangeVoterHandlerStateTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/internals/ChangeVoterHandlerStateTest.java
@@ -17,13 +17,17 @@
package org.apache.kafka.raft.internals;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.feature.SupportedVersionRange;
import org.apache.kafka.common.message.AddRaftVoterResponseData;
import org.apache.kafka.common.message.RemoveRaftVoterResponseData;
+import org.apache.kafka.common.message.UpdateRaftVoterResponseData;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.utils.MockTime;
import org.apache.kafka.raft.Endpoints;
+import org.apache.kafka.raft.LeaderAndEpoch;
import org.apache.kafka.raft.ReplicaKey;
import org.junit.jupiter.api.Test;
@@ -62,7 +66,7 @@ public class ChangeVoterHandlerStateTest {
false,
time.timer(1000)
);
- CompletableFuture<AddRaftVoterResponseData> future = addVoterState.future();
+ CompletableFuture<AddRaftVoterResponseData> future = addVoterState.future().toCompletableFuture();
state.resetAddVoterHandlerState(Errors.NONE, null, Optional.of(addVoterState));
assertTrue(state.addVoterHandlerState().isPresent());
@@ -100,7 +104,7 @@ public class ChangeVoterHandlerStateTest {
100L,
time.timer(1000)
);
- CompletableFuture<RemoveRaftVoterResponseData> future = removeVoterState.future();
+ CompletableFuture<RemoveRaftVoterResponseData> future = removeVoterState.future().toCompletableFuture();
state.resetRemoveVoterHandlerState(Errors.NONE, null, Optional.of(removeVoterState));
assertTrue(state.removeVoterHandlerState().isPresent());
@@ -131,7 +135,7 @@ public class ChangeVoterHandlerStateTest {
try {
ChangeVoterHandlerState state = new ChangeVoterHandlerState(raftMetrics);
- assertEquals(Long.MAX_VALUE, state.maybeExpirePendingOperation(time.milliseconds()));
+ assertEquals(Long.MAX_VALUE, state.maybeExpirePendingOperation(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
AddVoterHandlerState addVoterState = new AddVoterHandlerState(
ReplicaKey.of(1, Uuid.randomUuid()),
@@ -142,16 +146,17 @@ public class ChangeVoterHandlerStateTest {
state.resetAddVoterHandlerState(Errors.NONE, null, Optional.of(addVoterState));
time.sleep(500);
- assertEquals(500, state.maybeExpirePendingOperation(time.milliseconds()));
+ assertEquals(500, state.maybeExpirePendingOperation(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertTrue(state.addVoterHandlerState().isPresent());
assertEquals(1, getMetric(metrics, "uncommitted-voter-change").metricValue());
time.sleep(500);
- assertEquals(Long.MAX_VALUE, state.maybeExpirePendingOperation(time.milliseconds()));
+ assertEquals(Long.MAX_VALUE, state.maybeExpirePendingOperation(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertFalse(state.addVoterHandlerState().isPresent());
assertEquals(0, getMetric(metrics, "uncommitted-voter-change").metricValue());
- assertTrue(addVoterState.future().isDone());
- assertEquals(Errors.REQUEST_TIMED_OUT, Errors.forCode(addVoterState.future().join().errorCode()));
+ var future = addVoterState.future().toCompletableFuture();
+ assertTrue(future.isDone());
+ assertEquals(Errors.REQUEST_TIMED_OUT, Errors.forCode(future.join().errorCode()));
} finally {
raftMetrics.close();
metrics.close();
@@ -175,16 +180,17 @@ public class ChangeVoterHandlerStateTest {
state.resetRemoveVoterHandlerState(Errors.NONE, null, Optional.of(removeVoterState));
time.sleep(500);
- assertEquals(500, state.maybeExpirePendingOperation(time.milliseconds()));
+ assertEquals(500, state.maybeExpirePendingOperation(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertTrue(state.removeVoterHandlerState().isPresent());
assertEquals(1, getMetric(metrics, "uncommitted-voter-change").metricValue());
time.sleep(500);
- assertEquals(Long.MAX_VALUE, state.maybeExpirePendingOperation(time.milliseconds()));
+ assertEquals(Long.MAX_VALUE, state.maybeExpirePendingOperation(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertFalse(state.removeVoterHandlerState().isPresent());
assertEquals(0, getMetric(metrics, "uncommitted-voter-change").metricValue());
- assertTrue(removeVoterState.future().isDone());
- assertEquals(Errors.REQUEST_TIMED_OUT, Errors.forCode(removeVoterState.future().join().errorCode()));
+ var future = removeVoterState.future().toCompletableFuture();
+ assertTrue(future.isDone());
+ assertEquals(Errors.REQUEST_TIMED_OUT, Errors.forCode(future.join().errorCode()));
} finally {
raftMetrics.close();
metrics.close();
@@ -201,7 +207,7 @@ public class ChangeVoterHandlerStateTest {
try {
ChangeVoterHandlerState state = new ChangeVoterHandlerState(raftMetrics);
- assertFalse(state.isOperationPending(time.milliseconds()));
+ assertFalse(state.isOperationPending(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertEquals(0, getMetric(metrics, "uncommitted-voter-change").metricValue());
AddVoterHandlerState addVoterState = new AddVoterHandlerState(
@@ -212,11 +218,11 @@ public class ChangeVoterHandlerStateTest {
);
state.resetAddVoterHandlerState(Errors.NONE, null, Optional.of(addVoterState));
- assertTrue(state.isOperationPending(time.milliseconds()));
+ assertTrue(state.isOperationPending(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertEquals(1, getMetric(metrics, "uncommitted-voter-change").metricValue());
time.sleep(1000);
- assertFalse(state.isOperationPending(time.milliseconds()));
+ assertFalse(state.isOperationPending(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertFalse(state.addVoterHandlerState().isPresent());
assertEquals(0, getMetric(metrics, "uncommitted-voter-change").metricValue());
} finally {
@@ -235,7 +241,7 @@ public class ChangeVoterHandlerStateTest {
try {
ChangeVoterHandlerState state = new ChangeVoterHandlerState(raftMetrics);
- assertFalse(state.isOperationPending(time.milliseconds()));
+ assertFalse(state.isOperationPending(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
RemoveVoterHandlerState removeVoterState = new RemoveVoterHandlerState(
100L,
@@ -243,11 +249,11 @@ public class ChangeVoterHandlerStateTest {
);
state.resetRemoveVoterHandlerState(Errors.NONE, null, Optional.of(removeVoterState));
- assertTrue(state.isOperationPending(time.milliseconds()));
+ assertTrue(state.isOperationPending(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertEquals(1, getMetric(metrics, "uncommitted-voter-change").metricValue());
time.sleep(1000);
- assertFalse(state.isOperationPending(time.milliseconds()));
+ assertFalse(state.isOperationPending(LeaderAndEpoch.UNKNOWN, Endpoints.empty(), time.milliseconds()));
assertFalse(state.removeVoterHandlerState().isPresent());
assertEquals(0, getMetric(metrics, "uncommitted-voter-change").metricValue());
} finally {
@@ -274,12 +280,16 @@ public class ChangeVoterHandlerStateTest {
);
state.resetAddVoterHandlerState(Errors.NONE, null, Optional.of(addVoterState));
- CompletableFuture<AddRaftVoterResponseData> addFuture = addVoterState.future();
+ CompletableFuture<AddRaftVoterResponseData> addFuture = addVoterState.future().toCompletableFuture();
assertFalse(addFuture.isDone());
assertEquals(1, getMetric(metrics, "uncommitted-voter-change").metricValue());
- state.maybeResetPendingVoterHandlerState(Errors.NOT_LEADER_OR_FOLLOWER);
+ state.maybeResetPendingVoterHandlerState(
+ Errors.NOT_LEADER_OR_FOLLOWER,
+ LeaderAndEpoch.UNKNOWN,
+ Endpoints.empty()
+ );
assertFalse(state.addVoterHandlerState().isPresent());
assertEquals(0, getMetric(metrics, "uncommitted-voter-change").metricValue());
@@ -379,7 +389,7 @@ public class ChangeVoterHandlerStateTest {
false,
time.timer(1000)
);
- CompletableFuture<AddRaftVoterResponseData> firstFuture = firstAddVoter.future();
+ CompletableFuture<AddRaftVoterResponseData> firstFuture = firstAddVoter.future().toCompletableFuture();
state.resetAddVoterHandlerState(Errors.NONE, null, Optional.of(firstAddVoter));
AddVoterHandlerState secondAddVoter = new AddVoterHandlerState(
@@ -415,7 +425,7 @@ public class ChangeVoterHandlerStateTest {
100L,
time.timer(1000)
);
- CompletableFuture<RemoveRaftVoterResponseData> firstFuture = firstRemoveVoter.future();
+ CompletableFuture<RemoveRaftVoterResponseData> firstFuture = firstRemoveVoter.future().toCompletableFuture();
state.resetRemoveVoterHandlerState(Errors.NONE, null, Optional.of(firstRemoveVoter));
RemoveVoterHandlerState secondRemoveVoter = new RemoveVoterHandlerState(
@@ -434,4 +444,227 @@ public class ChangeVoterHandlerStateTest {
metrics.close();
}
}
+
+ @Test
+ public void testCannotSetAddVoterWhenUpdateVoterPresent() {
+ MockTime time = new MockTime();
+ Metrics metrics = new Metrics(time);
+ KafkaRaftMetrics raftMetrics = new KafkaRaftMetrics(metrics, "raft");
+ raftMetrics.addLeaderMetrics();
+
+ try {
+ ChangeVoterHandlerState state = new ChangeVoterHandlerState(raftMetrics);
+
+ UpdateVoterHandlerState updateVoterState = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, Uuid.randomUuid()),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+ state.resetUpdateVoterHandlerState(
+ Errors.NONE,
+ LeaderAndEpoch.UNKNOWN,
+ Endpoints.empty(),
+ Optional.of(updateVoterState)
+ );
+ assertTrue(state.updateVoterHandlerState().isPresent());
+
+ AddVoterHandlerState addVoterState = new AddVoterHandlerState(
+ ReplicaKey.of(2, Uuid.randomUuid()),
+ Endpoints.empty(),
+ false,
+ time.timer(1000)
+ );
+
+ assertThrows(IllegalStateException.class, () -> {
+ state.resetAddVoterHandlerState(Errors.NONE, null, Optional.of(addVoterState));
+ });
+
+ assertTrue(state.updateVoterHandlerState().isPresent());
+ assertFalse(state.addVoterHandlerState().isPresent());
+ } finally {
+ raftMetrics.close();
+ metrics.close();
+ }
+ }
+
+ @Test
+ public void testCannotSetRemoveVoterWhenUpdateVoterPresent() {
+ MockTime time = new MockTime();
+ Metrics metrics = new Metrics(time);
+ KafkaRaftMetrics raftMetrics = new KafkaRaftMetrics(metrics, "raft");
+ raftMetrics.addLeaderMetrics();
+
+ try {
+ ChangeVoterHandlerState state = new ChangeVoterHandlerState(raftMetrics);
+
+ UpdateVoterHandlerState updateVoterState = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, Uuid.randomUuid()),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+ state.resetUpdateVoterHandlerState(
+ Errors.NONE,
+ LeaderAndEpoch.UNKNOWN,
+ Endpoints.empty(),
+ Optional.of(updateVoterState)
+ );
+ assertTrue(state.updateVoterHandlerState().isPresent());
+
+ RemoveVoterHandlerState removeVoterState = new RemoveVoterHandlerState(
+ 100L,
+ time.timer(1000)
+ );
+
+ assertThrows(IllegalStateException.class, () -> {
+ state.resetRemoveVoterHandlerState(Errors.NONE, null, Optional.of(removeVoterState));
+ });
+
+ assertTrue(state.updateVoterHandlerState().isPresent());
+ assertFalse(state.removeVoterHandlerState().isPresent());
+ } finally {
+ raftMetrics.close();
+ metrics.close();
+ }
+ }
+
+ @Test
+ public void testCannotSetUpdateVoterWhenAddVoterPresent() {
+ MockTime time = new MockTime();
+ Metrics metrics = new Metrics(time);
+ KafkaRaftMetrics raftMetrics = new KafkaRaftMetrics(metrics, "raft");
+ raftMetrics.addLeaderMetrics();
+
+ try {
+ ChangeVoterHandlerState state = new ChangeVoterHandlerState(raftMetrics);
+
+ AddVoterHandlerState addVoterState = new AddVoterHandlerState(
+ ReplicaKey.of(1, Uuid.randomUuid()),
+ Endpoints.empty(),
+ false,
+ time.timer(1000)
+ );
+ state.resetAddVoterHandlerState(Errors.NONE, null, Optional.of(addVoterState));
+ assertTrue(state.addVoterHandlerState().isPresent());
+
+ UpdateVoterHandlerState updateVoterState = new UpdateVoterHandlerState(
+ ReplicaKey.of(2, Uuid.randomUuid()),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+
+ assertThrows(IllegalStateException.class, () -> {
+ state.resetUpdateVoterHandlerState(
+ Errors.NONE,
+ LeaderAndEpoch.UNKNOWN,
+ Endpoints.empty(),
+ Optional.of(updateVoterState)
+ );
+ });
+
+ assertTrue(state.addVoterHandlerState().isPresent());
+ assertFalse(state.updateVoterHandlerState().isPresent());
+ } finally {
+ raftMetrics.close();
+ metrics.close();
+ }
+ }
+
+ @Test
+ public void testCannotSetUpdateVoterWhenRemoveVoterPresent() {
+ MockTime time = new MockTime();
+ Metrics metrics = new Metrics(time);
+ KafkaRaftMetrics raftMetrics = new KafkaRaftMetrics(metrics, "raft");
+ raftMetrics.addLeaderMetrics();
+
+ try {
+ ChangeVoterHandlerState state = new ChangeVoterHandlerState(raftMetrics);
+
+ RemoveVoterHandlerState removeVoterState = new RemoveVoterHandlerState(
+ 100L,
+ time.timer(1000)
+ );
+ state.resetRemoveVoterHandlerState(Errors.NONE, null, Optional.of(removeVoterState));
+ assertTrue(state.removeVoterHandlerState().isPresent());
+
+ UpdateVoterHandlerState updateVoterState = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, Uuid.randomUuid()),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+
+ assertThrows(IllegalStateException.class, () -> {
+ state.resetUpdateVoterHandlerState(
+ Errors.NONE,
+ LeaderAndEpoch.UNKNOWN,
+ Endpoints.empty(),
+ Optional.of(updateVoterState)
+ );
+ });
+
+ assertTrue(state.removeVoterHandlerState().isPresent());
+ assertFalse(state.updateVoterHandlerState().isPresent());
+ } finally {
+ raftMetrics.close();
+ metrics.close();
+ }
+ }
+
+ @Test
+ public void testCanReplaceUpdateVoterWithAnotherUpdateVoter() {
+ MockTime time = new MockTime();
+ Metrics metrics = new Metrics(time);
+ KafkaRaftMetrics raftMetrics = new KafkaRaftMetrics(metrics, "raft");
+ raftMetrics.addLeaderMetrics();
+
+ try {
+ ChangeVoterHandlerState state = new ChangeVoterHandlerState(raftMetrics);
+
+ UpdateVoterHandlerState firstUpdateVoter = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, Uuid.randomUuid()),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+ CompletableFuture<UpdateRaftVoterResponseData> firstFuture = firstUpdateVoter.future().toCompletableFuture();
+ state.resetUpdateVoterHandlerState(
+ Errors.NONE,
+ LeaderAndEpoch.UNKNOWN,
+ Endpoints.empty(),
+ Optional.of(firstUpdateVoter)
+ );
+
+ UpdateVoterHandlerState secondUpdateVoter = new UpdateVoterHandlerState(
+ ReplicaKey.of(2, Uuid.randomUuid()),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+ state.resetUpdateVoterHandlerState(
+ Errors.OPERATION_NOT_ATTEMPTED,
+ LeaderAndEpoch.UNKNOWN,
+ Endpoints.empty(),
+ Optional.of(secondUpdateVoter)
+ );
+
+ assertTrue(state.updateVoterHandlerState().isPresent());
+ assertEquals(secondUpdateVoter, state.updateVoterHandlerState().get());
+ assertFalse(state.addVoterHandlerState().isPresent());
+ assertFalse(state.removeVoterHandlerState().isPresent());
+ assertTrue(firstFuture.isDone());
+ assertEquals(Errors.OPERATION_NOT_ATTEMPTED, Errors.forCode(firstFuture.join().errorCode()));
+ } finally {
+ raftMetrics.close();
+ metrics.close();
+ }
+ }
}
diff --git a/raft/src/test/java/org/apache/kafka/raft/internals/KafkaRaftMetricsTest.java b/raft/src/test/java/org/apache/kafka/raft/internals/KafkaRaftMetricsTest.java
index 620e08a04e0..1ecee25c21b 100644
--- a/raft/src/test/java/org/apache/kafka/raft/internals/KafkaRaftMetricsTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/internals/KafkaRaftMetricsTest.java
@@ -86,7 +86,7 @@ public class KafkaRaftMetricsTest {
localDirectoryId,
mockPartitionState,
voterSet.listeners(localId),
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
electionTimeoutMs,
fetchTimeoutMs,
new MockQuorumStateStore(),
diff --git a/raft/src/test/java/org/apache/kafka/raft/internals/RemoveVoterHandlerStateTest.java b/raft/src/test/java/org/apache/kafka/raft/internals/RemoveVoterHandlerStateTest.java
new file mode 100644
index 00000000000..7ff5baa90cb
--- /dev/null
+++ b/raft/src/test/java/org/apache/kafka/raft/internals/RemoveVoterHandlerStateTest.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.message.RemoveRaftVoterResponseData;
+import org.apache.kafka.common.utils.MockTime;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class RemoveVoterHandlerStateTest {
+
+ @Test
+ public void testLastOffset() {
+ var time = new MockTime();
+ var state = new RemoveVoterHandlerState(100L, time.timer(1000));
+
+ assertEquals(100L, state.lastOffset());
+ }
+
+ @Test
+ public void testTimeUntilExpiration() {
+ var time = new MockTime();
+ var state = new RemoveVoterHandlerState(100L, time.timer(1000));
+
+ assertEquals(1000, state.timeUntilOperationExpiration(time.milliseconds()));
+ time.sleep(500);
+ assertEquals(500, state.timeUntilOperationExpiration(time.milliseconds()));
+ time.sleep(500);
+ assertEquals(0, state.timeUntilOperationExpiration(time.milliseconds()));
+ }
+
+ @Test
+ public void testCompleteFuture() {
+ var time = new MockTime();
+ var state = new RemoveVoterHandlerState(100L, time.timer(1000));
+
+ var future = state.future().toCompletableFuture();
+ assertFalse(future.isDone());
+
+ var response = new RemoveRaftVoterResponseData().setErrorCode((short) 0);
+ state.completeFuture(response);
+
+ assertTrue(future.isDone());
+ assertEquals(response, future.join());
+ }
+}
diff --git a/raft/src/test/java/org/apache/kafka/raft/internals/UpdateVoterHandlerStateTest.java b/raft/src/test/java/org/apache/kafka/raft/internals/UpdateVoterHandlerStateTest.java
new file mode 100644
index 00000000000..8b3e2dddba3
--- /dev/null
+++ b/raft/src/test/java/org/apache/kafka/raft/internals/UpdateVoterHandlerStateTest.java
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.raft.internals;
+
+import org.apache.kafka.common.feature.SupportedVersionRange;
+import org.apache.kafka.common.message.UpdateRaftVoterResponseData;
+import org.apache.kafka.common.network.ListenerName;
+import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.raft.Endpoints;
+import org.apache.kafka.raft.ReplicaKey;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.OptionalLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class UpdateVoterHandlerStateTest {
+
+ @Test
+ public void testSetLastOffsetOnce() {
+ var time = new MockTime();
+ var state = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+
+ assertTrue(state.lastOffset().isEmpty());
+ state.setLastOffset(100L);
+ assertEquals(OptionalLong.of(100L), state.lastOffset());
+ }
+
+ @Test
+ public void testCannotOverrideLastOffset() {
+ var time = new MockTime();
+ var state = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+
+ state.setLastOffset(100L);
+ assertThrows(IllegalStateException.class, () -> state.setLastOffset(200L));
+ }
+
+ @Test
+ public void testExpectingApiResponseBeforeLastOffset() {
+ var time = new MockTime();
+ var state = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+
+ assertTrue(state.expectingApiResponse(1));
+ assertFalse(state.expectingApiResponse(2));
+
+ state.setLastOffset(100L);
+ assertFalse(state.expectingApiResponse(1));
+ }
+
+ @Test
+ public void testTimeUntilExpiration() {
+ var time = new MockTime();
+ var state = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+
+ assertEquals(1000, state.timeUntilOperationExpiration(time.milliseconds()));
+ time.sleep(500);
+ assertEquals(500, state.timeUntilOperationExpiration(time.milliseconds()));
+ time.sleep(500);
+ assertEquals(0, state.timeUntilOperationExpiration(time.milliseconds()));
+ }
+
+ @Test
+ public void testCompleteFuture() {
+ var time = new MockTime();
+ var state = new UpdateVoterHandlerState(
+ ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID),
+ Endpoints.empty(),
+ new ListenerName("PLAINTEXT"),
+ new SupportedVersionRange((short) 0, (short) 1),
+ time.timer(1000)
+ );
+
+ var future = state.future().toCompletableFuture();
+ assertFalse(future.isDone());
+
+ var response = new UpdateRaftVoterResponseData().setErrorCode((short) 0);
+ state.completeFuture(response);
+
+ assertTrue(future.isDone());
+ assertEquals(response, future.join());
+ }
+
+ @Test
+ public void testGetters() {
+ var time = new MockTime();
+ var voterKey = ReplicaKey.of(1, ReplicaKey.NO_DIRECTORY_ID);
+ var endpoints = Endpoints.empty();
+ var listenerName = new ListenerName("PLAINTEXT");
+ var versions = new SupportedVersionRange((short) 0, (short) 1);
+
+ var state = new UpdateVoterHandlerState(
+ voterKey,
+ endpoints,
+ listenerName,
+ versions,
+ time.timer(1000)
+ );
+
+ assertEquals(voterKey, state.voterKey());
+ assertEquals(endpoints, state.voterEndpoints());
+ assertEquals(listenerName, state.requestListenerName());
+ assertEquals(versions, state.supportedKraftVersions());
+ }
+}
diff --git a/raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientTestContext.java b/raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientTestContext.java
index ad47c8d9581..c2e32b767cb 100644
--- a/raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientTestContext.java
+++ b/raft/src/testFixtures/java/org/apache/kafka/raft/RaftClientTestContext.java
@@ -150,7 +150,7 @@ public final class RaftClientTestContext {
private final List<RaftResponse.Outbound> sentResponses = new ArrayList<>();
private final List<Throwable> uncaughtExceptions = new ArrayList<>();
- private static final int NUMBER_FETCH_TIMEOUTS_IN_UPDATE_VOTER_SET_PERIOD = 1;
+ private static final int NUMBER_FETCH_TIMEOUTS_IN_UPDATE_VOTER_SET_PERIOD = 2;
public static final class Builder {
static final int DEFAULT_ELECTION_TIMEOUT_MS = 10000;
@@ -473,7 +473,7 @@ public final class RaftClientTestContext {
clusterId,
computedBootstrapServers,
localListeners,
- Feature.KRAFT_VERSION.supportedVersionRange(),
+ Feature.KRAFT_VERSION.supportedVersionRange(true),
logContext,
random,
quorumConfig
diff --git a/raft/src/testFixtures/java/org/apache/kafka/raft/VoterSetTestUtil.java b/raft/src/testFixtures/java/org/apache/kafka/raft/VoterSetTestUtil.java
index 1324bb19738..0af56331f65 100644
--- a/raft/src/testFixtures/java/org/apache/kafka/raft/VoterSetTestUtil.java
+++ b/raft/src/testFixtures/java/org/apache/kafka/raft/VoterSetTestUtil.java
@@ -77,7 +77,7 @@ public final class VoterSetTestUtil {
public static VoterSet.VoterNode voterNode(ReplicaKey replicaKey, Endpoints endpoints) {
var supportedVersionRange = replicaKey.directoryId().isEmpty() ?
new SupportedVersionRange((short) 0) :
- Feature.KRAFT_VERSION.supportedVersionRange();
+ Feature.KRAFT_VERSION.supportedVersionRange(true);
return new VoterSet.VoterNode(replicaKey, endpoints, supportedVersionRange);
}
diff --git a/server-common/src/main/java/org/apache/kafka/server/common/Feature.java b/server-common/src/main/java/org/apache/kafka/server/common/Feature.java
index 9744ff7f9df..9a386725d4c 100644
--- a/server-common/src/main/java/org/apache/kafka/server/common/Feature.java
+++ b/server-common/src/main/java/org/apache/kafka/server/common/Feature.java
@@ -127,10 +127,10 @@ public enum Feature {
return featureVersions[featureVersions.length - 1].featureLevel();
}
- public SupportedVersionRange supportedVersionRange() {
+ public SupportedVersionRange supportedVersionRange(boolean includeUnstable) {
return new SupportedVersionRange(
minimumProduction(),
- latestTesting()
+ includeUnstable ? latestTesting() : latestProduction()
);
}
diff --git a/server/src/main/java/org/apache/kafka/server/BrokerFeatures.java b/server/src/main/java/org/apache/kafka/server/BrokerFeatures.java
index f611411f9b8..994f98e15c7 100644
--- a/server/src/main/java/org/apache/kafka/server/BrokerFeatures.java
+++ b/server/src/main/java/org/apache/kafka/server/BrokerFeatures.java
@@ -64,17 +64,17 @@ public class BrokerFeatures {
.collect(Collectors.toMap(Map.Entry::getKey, e -> VersionRange.of(e.getValue().min(), e.getValue().max())));
}
- public static Features<SupportedVersionRange> defaultSupportedFeatures(boolean unstableFeatureVersionsEnabled) {
+ public static Features<SupportedVersionRange> defaultSupportedFeatures(boolean includeUnstable) {
Map<String, SupportedVersionRange> features = new HashMap<>();
features.put(MetadataVersion.FEATURE_NAME,
new SupportedVersionRange(
MetadataVersion.MINIMUM_VERSION.featureLevel(),
- unstableFeatureVersionsEnabled ? MetadataVersion.latestTesting().featureLevel()
+ includeUnstable ? MetadataVersion.latestTesting().featureLevel()
: MetadataVersion.latestProduction().featureLevel()));
PRODUCTION_FEATURES.forEach(feature -> {
- int maxVersion = unstableFeatureVersionsEnabled ? feature.latestTesting() : feature.latestProduction();
- if (maxVersion > 0) {
- features.put(feature.featureName(), new SupportedVersionRange(feature.minimumProduction(), (short) maxVersion));
+ var range = feature.supportedVersionRange(includeUnstable);
+ if (range.max() > 0) {
+ features.put(feature.featureName(), range);
}
});
return Features.supportedFeatures(features);