You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2026/08/21 15:42:16 UTC
(kafka) branch trunk updated: KAFKA-20865: Do not unregister KRaft voter set members (#23152)
This is an automated email from the ASF dual-hosted git repository.
junrao 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 05f61b9ff5e KAFKA-20865: Do not unregister KRaft voter set members (#23152)
05f61b9ff5e is described below
commit 05f61b9ff5ed72f32ef8371f6ae338ab9b8f7ce5
Author: Kevin Wu <kw...@confluent.io>
AuthorDate: Fri Aug 21 08:42:06 2026 -0700
KAFKA-20865: Do not unregister KRaft voter set members (#23152)
Follow up PR to
https://github.com/apache/kafka/commit/c274a7348fb66d1bde99566bdb072fbb52b03dfd,
which implements controller unregistration. This PR adds a check so that
KRaft voter set members cannot be unregistered, which can limit the
severity of weird edge cases around races between unregistration
requests and feature upgrades. Voters who are unregistered and then
re-register with a cluster who completed an unsupported feature upgrade
thanks to the unregistration would experience unavailability, which
affects quorum health.
The KRaft voter set is internal to the `/raft` module, so this PR adds
the `RaftClientVotersSupplier` alongside some new public `RaftClient`
API to expose the latest voter set. Only the latest is necessary,
because with `kraft.version=1` enabled, the controller being
unregistered depends only on the uncommitted voter set when it is
present, described below.
When dynamic quorum is enabled, the unregistration voter check only
needs to check the uncommitted voter set, if one exists. This is because
the unregistration record will be written, and thus committed, after the
uncommitted `VotersRecord` if one exists. The logic is as follows for
unregistering controller X, with the latest CVS (committed voter set)
and UVS (uncommitted voter set):
- `CVS.contains(X) && UVS.contains(X)`: cannot unregister, because X is
definitely a voter
- `CVS.contains(X) && !UVS.contains(X)`: can unregister, because the
offset of X's unregistration's record will come after its removal from
the voter set. X's unregistration cannot be committed without the new
voter set also being committed first.
- `!CVS.contains(X) && UVS.contains(X)`: cannot unregister, because
committing both the new voter set and X's unregistration results in
unregistering a voter
- `!CVS.contains(X) && !UVS.contains(X)`: can unregister, because X is
definitely not a voter
So long as UVS is present, X's unregistration request validity only
depends on that whether or not X is contained in UVS. This also applies
to the registration check that is needed for feature upgrades. If there
is an uncommitted voter set, all of that set's voters must be registered
to the cluster in order to upgrade a feature to a certain value, since
those `FeatureLevelRecords` will be committed after the uncommitted
voter set.
Reviewers: Jun Rao <ju...@gmail.com>
---
.../java/org/apache/kafka/clients/admin/Admin.java | 2 +-
.../main/scala/kafka/server/ControllerServer.scala | 11 +++--
.../org/apache/kafka/controller/Controller.java | 3 ++
.../kafka/controller/FeatureControlManager.java | 10 +++--
.../apache/kafka/controller/QuorumController.java | 13 +++---
.../apache/kafka/controller/QuorumFeatures.java | 29 +++++++------
.../kafka/controller/RaftClientVotersSupplier.java | 49 ++++++++++++++++++++++
.../controller/ClusterControlManagerTest.java | 26 ++++++------
.../ConfigurationControlManagerTest.java | 6 +--
.../controller/FeatureControlManagerTest.java | 6 +--
.../apache/kafka/controller/MockRaftClient.java | 15 +++++++
.../controller/ProducerIdControlManagerTest.java | 4 +-
.../kafka/controller/QuorumControllerTest.java | 30 +++++++++++++
.../kafka/controller/QuorumControllerTestEnv.java | 6 ++-
.../kafka/controller/QuorumFeaturesTest.java | 25 ++++++++---
.../controller/RaftClientVotersSupplierTest.java | 48 +++++++++++++++++++++
.../controller/ReplicationControlManagerTest.java | 2 +-
.../kafka/image/publisher/SnapshotEmitterTest.java | 6 +++
.../org/apache/kafka/raft/KafkaRaftClient.java | 9 ++++
.../java/org/apache/kafka/raft/RaftClient.java | 11 +++++
.../kafka/raft/KafkaRaftClientReconfigTest.java | 7 +++-
.../org/apache/kafka/raft/KafkaRaftClientTest.java | 22 ++++++++++
.../org/apache/kafka/server/KRaftClusterTest.java | 25 ++++++++++-
23 files changed, 301 insertions(+), 64 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
index 4cb40c66980..4b669dab9e7 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
@@ -1711,7 +1711,7 @@ public interface Admin extends AutoCloseable {
* <li>{@link org.apache.kafka.common.errors.NotControllerException}
* If the request does not arrive at the active controller.</li>
* <li>{@link org.apache.kafka.common.errors.InvalidRequestException}
- * If the request tries to unregister the current active controller id.</li>
+ * If the request tries to unregister a controller id which is part of the voter set.</li>
*
* </ul>
* <p>
diff --git a/core/src/main/scala/kafka/server/ControllerServer.scala b/core/src/main/scala/kafka/server/ControllerServer.scala
index c2906e4c674..01ca63fdc25 100644
--- a/core/src/main/scala/kafka/server/ControllerServer.scala
+++ b/core/src/main/scala/kafka/server/ControllerServer.scala
@@ -35,13 +35,12 @@ import org.apache.kafka.common.utils.Utils
import org.apache.kafka.common.utils.internals.LogContext
import org.apache.kafka.common.{ClusterResource, Endpoint, Uuid}
import org.apache.kafka.controller.metrics.{ControllerMetadataMetricsPublisher, QuorumControllerMetrics}
-import org.apache.kafka.controller.{Controller, QuorumController, QuorumFeatures}
+import org.apache.kafka.controller.{Controller, QuorumController, QuorumFeatures, RaftClientVotersSupplier}
import org.apache.kafka.image.publisher.{ControllerRegistrationsPublisher, KRaftMetadataCachePublisher, MetadataPublisher}
-import org.apache.kafka.metadata.{KafkaConfigSchema, KRaftMetadataCache, ListenerInfo}
+import org.apache.kafka.metadata.{KRaftMetadataCache, KafkaConfigSchema, ListenerInfo}
import org.apache.kafka.metadata.authorizer.ClusterMetadataAuthorizer
import org.apache.kafka.metadata.bootstrap.BootstrapMetadata
import org.apache.kafka.metadata.publisher.{AclPublisher, DelegationTokenPublisher, DynamicClientQuotaPublisher, DynamicTopicClusterQuotaPublisher, FeaturesPublisher, ScramPublisher}
-import org.apache.kafka.raft.QuorumConfig
import org.apache.kafka.security.{CredentialProvider, DelegationTokenManager}
import org.apache.kafka.server.{ProcessRole, SimpleApiVersionManager}
import org.apache.kafka.server.authorizer.Authorizer
@@ -211,14 +210,14 @@ class ControllerServer(
alterConfigPolicy = Option(config.
getConfiguredInstance(ALTER_CONFIG_POLICY_CLASS_NAME_CONFIG, classOf[AlterConfigPolicy]))
- val voterConnections = FutureUtils.waitWithLogging(logger.underlying, logIdent,
+ // Wait for the controller quorum voters to be resolvable before starting the controller.
+ FutureUtils.waitWithLogging(logger.underlying, logIdent,
"controller quorum voters future",
sharedServer.controllerQuorumVotersFuture,
startupDeadline, time)
- val controllerNodes = QuorumConfig.voterConnectionsToNodes(voterConnections)
val quorumFeatures = new QuorumFeatures(config.nodeId,
QuorumFeatures.defaultSupportedFeatureMap(config.unstableFeatureVersionsEnabled),
- controllerNodes.asScala.map(node => Integer.valueOf(node.id())).asJava)
+ new RaftClientVotersSupplier(raftManager.client))
val delegationTokenManagerConfigs = new DelegationTokenManagerConfigs(config)
val delegationTokenKeyString = {
diff --git a/metadata/src/main/java/org/apache/kafka/controller/Controller.java b/metadata/src/main/java/org/apache/kafka/controller/Controller.java
index c02e357ddf7..f1320ef60f5 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/Controller.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/Controller.java
@@ -415,6 +415,9 @@ public interface Controller extends AclMutator, AutoCloseable {
/**
* Attempt to unregister the given controller.
*
+ * The request fails with an {@code InvalidRequestException} if the given controller
+ * is part of the voter set.
+ *
* @param context The controller request context.
* @param controllerId The controller id to unregister.
*
diff --git a/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java b/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java
index 01188bdd6d7..d4fe17334d5 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/FeatureControlManager.java
@@ -46,6 +46,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
+import java.util.Set;
import java.util.function.Consumer;
import static org.apache.kafka.common.metadata.MetadataRecordType.FEATURE_LEVEL_RECORD;
@@ -97,6 +98,7 @@ public class FeatureControlManager {
}
public FeatureControlManager build() {
+ // set reasonable defaults for fields when this object is built in testing
if (logContext == null) logContext = new LogContext();
if (snapshotRegistry == null) snapshotRegistry = new SnapshotRegistry(logContext);
if (quorumFeatures == null) {
@@ -104,7 +106,7 @@ public class FeatureControlManager {
localSupportedFeatures.put(MetadataVersion.FEATURE_NAME, VersionRange.of(
MetadataVersion.MINIMUM_VERSION.featureLevel(),
MetadataVersion.latestProduction().featureLevel()));
- quorumFeatures = new QuorumFeatures(0, localSupportedFeatures, List.of(0));
+ quorumFeatures = new QuorumFeatures(0, localSupportedFeatures, () -> Set.of(0));
}
if (kraftVersionAccessor == null) {
kraftVersionAccessor = new KRaftVersionAccessor() {
@@ -357,7 +359,7 @@ public class FeatureControlManager {
foundControllers.add(entry.getKey());
numControllersChecked++;
}
- for (int id : quorumFeatures.quorumNodeIds()) {
+ for (int id : quorumFeatures.voterIds()) {
if (!foundControllers.contains(id)) {
return Optional.of("controller " + id + " has not registered, and may not " +
"support this feature");
@@ -470,8 +472,8 @@ public class FeatureControlManager {
}
}
- boolean isControllerId(int nodeId) {
- return quorumFeatures.isControllerId(nodeId);
+ boolean isVoterId(int nodeId) {
+ return quorumFeatures.isVoterId(nodeId);
}
boolean isElrFeatureEnabled() {
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
index 128bab401fb..7e103dd42f9 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumController.java
@@ -489,7 +489,7 @@ public final class QuorumController implements Controller {
throw new InvalidRequestException("Invalid broker name " +
configResource.name());
}
- if (!isNodeIdRegistered(nodeId)) {
+ if (!isNodeIdKnown(nodeId)) {
throw new BrokerIdNotRegisteredException("No node with id " + nodeId + " found.");
}
break;
@@ -505,13 +505,13 @@ public final class QuorumController implements Controller {
}
/**
- * Checks if a node id is registered as a broker, controller in static/dynamic quorum.
+ * Checks if a node id is known to the controller, from registrations or the voter set.
*/
- private boolean isNodeIdRegistered(int nodeId) {
+ private boolean isNodeIdKnown(int nodeId) {
if (clusterControl.brokerRegistrations().containsKey(nodeId)) {
return true;
}
- if (featureControl.isControllerId(nodeId)) {
+ if (featureControl.isVoterId(nodeId)) {
return true;
}
return clusterControl.controllerRegistrations().containsKey(nodeId);
@@ -2164,8 +2164,9 @@ public final class QuorumController implements Controller {
) {
return appendWriteEvent("unregisterController", context.deadlineNs(),
() -> {
- if (nodeId == controllerId) {
- throw new InvalidRequestException("Controller cannot unregister itself while it is active.");
+ if (featureControl.isVoterId(controllerId)) {
+ throw new InvalidRequestException("Cannot unregister controller " + controllerId +
+ " because it is part of the voter set.");
}
return clusterControl.unregisterController(controllerId);
},
diff --git a/metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java b/metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java
index e54f0604242..c43c96acdd8 100644
--- a/metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java
+++ b/metadata/src/main/java/org/apache/kafka/controller/QuorumFeatures.java
@@ -28,16 +28,19 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.Set;
+import java.util.function.Supplier;
/**
- * A holder class of the local node's supported feature flags as well as the quorum node IDs.
+ * A holder class of the local node's supported feature flags as well as a supplier of the current
+ * voter IDs.
*/
public final class QuorumFeatures {
public static final VersionRange DISABLED = VersionRange.of(0, 0);
private final int nodeId;
private final Map<String, VersionRange> localSupportedFeatures;
- private final List<Integer> quorumNodeIds;
+ private final Supplier<Set<Integer>> votersSupplier;
public static Optional<String> reasonNotSupported(
short newVersion,
@@ -73,27 +76,27 @@ public final class QuorumFeatures {
public QuorumFeatures(
int nodeId,
Map<String, VersionRange> localSupportedFeatures,
- List<Integer> quorumNodeIds
+ Supplier<Set<Integer>> votersSupplier
) {
this.nodeId = nodeId;
this.localSupportedFeatures = Collections.unmodifiableMap(localSupportedFeatures);
- this.quorumNodeIds = Collections.unmodifiableList(quorumNodeIds);
+ this.votersSupplier = votersSupplier;
}
public int nodeId() {
return nodeId;
}
- public List<Integer> quorumNodeIds() {
- return quorumNodeIds;
+ public Set<Integer> voterIds() {
+ return votersSupplier.get();
}
public VersionRange localSupportedFeature(String name) {
return localSupportedFeatures.getOrDefault(name, DISABLED);
}
- public boolean isControllerId(int nodeId) {
- return quorumNodeIds.contains(nodeId);
+ public boolean isVoterId(int nodeId) {
+ return voterIds().contains(nodeId);
}
public Optional<String> reasonNotLocallySupported(
@@ -107,16 +110,16 @@ public final class QuorumFeatures {
@Override
public int hashCode() {
- return Objects.hash(nodeId, localSupportedFeatures, quorumNodeIds);
+ return Objects.hash(nodeId, localSupportedFeatures);
}
@Override
public boolean equals(Object o) {
if (o == null || !(o.getClass().equals(QuorumFeatures.class))) return false;
QuorumFeatures other = (QuorumFeatures) o;
+ // votersSupplier is deliberately excluded from this method
return nodeId == other.nodeId &&
- localSupportedFeatures.equals(other.localSupportedFeatures) &&
- quorumNodeIds.equals(other.quorumNodeIds);
+ localSupportedFeatures.equals(other.localSupportedFeatures);
}
@Override
@@ -124,13 +127,9 @@ public final class QuorumFeatures {
List<String> features = new ArrayList<>();
localSupportedFeatures.forEach((key, value) -> features.add(key + ": " + value));
features.sort(String::compareTo);
- List<String> nodeIds = new ArrayList<>();
- quorumNodeIds.forEach(id -> nodeIds.add("" + id));
- nodeIds.sort(String::compareTo);
return "QuorumFeatures" +
"(nodeId=" + nodeId +
", localSupportedFeatures={" + features + "}" +
- ", quorumNodeIds=[" + nodeIds + "]" +
")";
}
}
diff --git a/metadata/src/main/java/org/apache/kafka/controller/RaftClientVotersSupplier.java b/metadata/src/main/java/org/apache/kafka/controller/RaftClientVotersSupplier.java
new file mode 100644
index 00000000000..f70bae8e590
--- /dev/null
+++ b/metadata/src/main/java/org/apache/kafka/controller/RaftClientVotersSupplier.java
@@ -0,0 +1,49 @@
+/*
+ * 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.controller;
+
+import org.apache.kafka.raft.RaftClient;
+
+import java.util.Set;
+import java.util.function.Supplier;
+
+/**
+ * Provides the latest set of voters according to the raft client.
+ *
+ * When the kraft.version == 0, the voter set comes from the {@code controller.quorum.voters}
+ * configuration and never changes.
+ *
+ * When the kraft.version > 0, the voter set comes from the {@code VotersRecord}s in the metadata
+ * log. The latest voter set may not have been committed yet, but if it is present, we should base
+ * voter set membership based only on that. This is because any records whose writing is conditioned
+ * on voter set membership, such as feature upgrades or unregistrations, will be written at a later
+ * offset than the uncommitted voter set. Committing these records assumes the voter set record
+ * also is committed.
+ */
+public final class RaftClientVotersSupplier implements Supplier<Set<Integer>> {
+ private final RaftClient<?> raftClient;
+
+ public RaftClientVotersSupplier(RaftClient<?> raftClient) {
+ this.raftClient = raftClient;
+ }
+
+ @Override
+ public Set<Integer> get() {
+ return raftClient.latestVoterSet().voterIds();
+ }
+}
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
index 5f8b07527ba..f6717319907 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ClusterControlManagerTest.java
@@ -95,7 +95,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
ClusterControlManager clusterControl = new ClusterControlManager.Builder().
setTime(time).
@@ -144,7 +144,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
ClusterControlManager clusterControl = new ClusterControlManager.Builder().
setClusterId("fPZv1VBsRFmnlRvmGcOW9w").
@@ -197,7 +197,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
ClusterControlManager clusterControl = new ClusterControlManager.Builder().
setClusterId("fPZv1VBsRFmnlRvmGcOW9w").
@@ -286,7 +286,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
ClusterControlManager clusterControl = new ClusterControlManager.Builder().
setClusterId("fPZv1VBsRFmnlRvmGcOW9w").
@@ -324,7 +324,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
featureControl.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
@@ -395,7 +395,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
ClusterControlManager clusterControl = new ClusterControlManager.Builder().
setTime(new MockTime(0, 0, 0)).
@@ -434,7 +434,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
ClusterControlManager clusterControl = new ClusterControlManager.Builder().
setTime(time).
@@ -497,7 +497,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
featureControl.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
@@ -580,7 +580,7 @@ public class ClusterControlManagerTest {
TestFeatureVersion.TEST_1.featureLevel()));
FeatureControlManager featureControl = new FeatureControlManager.Builder().
setSnapshotRegistry(snapshotRegistry).
- setQuorumFeatures(new QuorumFeatures(0, supportedFeatures, List.of(0))).
+ setQuorumFeatures(new QuorumFeatures(0, supportedFeatures, () -> Set.of(0))).
build();
featureControl.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
@@ -631,7 +631,7 @@ public class ClusterControlManagerTest {
KRaftVersion.KRAFT_VERSION_1.featureLevel()));
FeatureControlManager featureControl = new FeatureControlManager.Builder().
setSnapshotRegistry(snapshotRegistry).
- setQuorumFeatures(new QuorumFeatures(0, supportedFeatures, List.of(0))).
+ setQuorumFeatures(new QuorumFeatures(0, supportedFeatures, () -> Set.of(0))).
build();
featureControl.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
@@ -714,7 +714,7 @@ public class ClusterControlManagerTest {
Map.of(MetadataVersion.FEATURE_NAME, VersionRange.of(
MetadataVersion.IBP_3_5_IV0.featureLevel(),
MetadataVersion.IBP_3_6_IV0.featureLevel())),
- List.of(0))).
+ () -> Set.of(0))).
build();
featureControl.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
@@ -1148,7 +1148,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
featureControl.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
@@ -1182,7 +1182,7 @@ public class ClusterControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
featureControl.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java
index b1e5f4c8872..bb6c50016c1 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ConfigurationControlManagerTest.java
@@ -487,7 +487,7 @@ public class ConfigurationControlManagerTest {
FeatureControlManager featureManager = new FeatureControlManager.Builder().
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of())).
+ () -> Set.of())).
build();
featureManager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
@@ -538,7 +538,7 @@ public class ConfigurationControlManagerTest {
FeatureControlManager featureManager = new FeatureControlManager.Builder().
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of())).
+ () -> Set.of())).
build();
featureManager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
@@ -573,7 +573,7 @@ public class ConfigurationControlManagerTest {
FeatureControlManager featureManager = new FeatureControlManager.Builder().
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of())).
+ () -> Set.of())).
build();
featureManager.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
diff --git a/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java
index 941578cedcc..8910bafc197 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/FeatureControlManagerTest.java
@@ -79,7 +79,7 @@ public class FeatureControlManagerTest {
public static QuorumFeatures features(Object... args) {
Map<String, VersionRange> features = QuorumFeatures.defaultSupportedFeatureMap(true);
features.putAll(rangeMap(args));
- return new QuorumFeatures(0, features, List.of());
+ return new QuorumFeatures(0, features, () -> Set.of());
}
private static Map<String, Short> updateMap(Object... args) {
@@ -422,7 +422,7 @@ public class FeatureControlManagerTest {
MetadataVersion.MINIMUM_VERSION.featureLevel(), MetadataVersion.latestTesting().featureLevel()));
localSupportedFeatures.put(Feature.TEST_VERSION.featureName(), VersionRange.of(0, 2));
FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(new QuorumFeatures(0, localSupportedFeatures, List.of())).
+ setQuorumFeatures(new QuorumFeatures(0, localSupportedFeatures, () -> Set.of())).
setClusterFeatureSupportDescriber(createFakeClusterFeatureSupportDescriber(
List.of(new SimpleImmutableEntry<>(1, Map.of(Feature.TEST_VERSION.featureName(), VersionRange.of(0, 3)))),
List.of())).
@@ -458,7 +458,7 @@ public class FeatureControlManagerTest {
MetadataVersion.IBP_4_0_IV1.featureLevel(), MetadataVersion.latestTesting().featureLevel()));
localSupportedFeatures.put(Feature.ELIGIBLE_LEADER_REPLICAS_VERSION.featureName(), VersionRange.of(0, 1));
FeatureControlManager manager = new FeatureControlManager.Builder().
- setQuorumFeatures(new QuorumFeatures(0, localSupportedFeatures, List.of())).
+ setQuorumFeatures(new QuorumFeatures(0, localSupportedFeatures, () -> Set.of())).
setClusterFeatureSupportDescriber(createFakeClusterFeatureSupportDescriber(
List.of(new SimpleImmutableEntry<>(1, Map.of(Feature.ELIGIBLE_LEADER_REPLICAS_VERSION.featureName(), VersionRange.of(0, 1)))),
List.of())).
diff --git a/metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java
index f817717a2b3..2730213dc24 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/MockRaftClient.java
@@ -30,6 +30,7 @@ import org.apache.kafka.queue.KafkaEventQueue;
import org.apache.kafka.raft.Batch;
import org.apache.kafka.raft.LeaderAndEpoch;
import org.apache.kafka.raft.RaftClient;
+import org.apache.kafka.raft.VoterSet;
import org.apache.kafka.raft.errors.BufferAllocationException;
import org.apache.kafka.raft.errors.NotLeaderException;
import org.apache.kafka.raft.internals.MemoryBatchReader;
@@ -421,6 +422,11 @@ public final class MockRaftClient implements RaftClient<ApiMessageAndVersion>, A
*/
private KRaftVersion lastKRaftVersion;
+ /**
+ * The latest voter set used by this raft client.
+ */
+ private volatile VoterSet latestVoterSet = VoterSet.empty();
+
/**
* Whether this raft client has been shut down.
*/
@@ -801,6 +807,15 @@ public final class MockRaftClient implements RaftClient<ApiMessageAndVersion>, A
return lastKRaftVersion;
}
+ @Override
+ public VoterSet latestVoterSet() {
+ return latestVoterSet;
+ }
+
+ public void setLatestVoterSet(VoterSet voterSet) {
+ this.latestVoterSet = voterSet;
+ }
+
@Override
public void upgradeKRaftVersion(int epoch, KRaftVersion version, boolean validateOnly) {
if (!validateOnly) {
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ProducerIdControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ProducerIdControlManagerTest.java
index 5f524e5e3fb..afb70a083d2 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ProducerIdControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ProducerIdControlManagerTest.java
@@ -30,7 +30,7 @@ import org.apache.kafka.timeline.SnapshotRegistry;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import java.util.List;
+import java.util.Set;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -48,7 +48,7 @@ public class ProducerIdControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
ClusterControlManager clusterControl = new ClusterControlManager.Builder().
setTime(time).
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
index b1ca959b4bf..ef7047d2f52 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
@@ -1404,6 +1404,36 @@ public class QuorumControllerTest {
}
}
+ @Test
+ public void testUnregisterControllerInVoterSet() throws Throwable {
+ try (
+ MockRaftClientTestEnv clientEnv = new MockRaftClientTestEnv.Builder(1).build();
+ QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(clientEnv).
+ setControllerBuilderInitializer(builder ->
+ builder.setQuorumFeatures(new QuorumFeatures(
+ 0,
+ QuorumFeatures.defaultSupportedFeatureMap(true),
+ () -> Set.of(0, 1)))).
+ build()
+ ) {
+ QuorumController active = controlEnv.activeController();
+
+ // Controller 0 and 1 are part of the voter set
+ assertEquals("Cannot unregister controller 0 because it is part of the voter set.",
+ assertThrows(ExecutionException.class,
+ () -> active.unregisterController(ANONYMOUS_CONTEXT, 0).get()).getCause().getMessage());
+
+ assertEquals("Cannot unregister controller 1 because it is part of the voter set.",
+ assertThrows(ExecutionException.class,
+ () -> active.unregisterController(ANONYMOUS_CONTEXT, 1).get()).getCause().getMessage());
+
+ // Controller 2 is not part of the voter set, so the voter set check passes
+ assertEquals("Controller ID 2 is not currently registered.",
+ assertThrows(ExecutionException.class,
+ () -> active.unregisterController(ANONYMOUS_CONTEXT, 2).get()).getCause().getMessage());
+ }
+ }
+
@Test
public void testIsNodeIdRegisteredWithDynamicQuorum() throws Throwable {
try (
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
index e6ececd2ba1..6b6d9ab62b7 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTestEnv.java
@@ -31,9 +31,11 @@ import java.util.List;
import java.util.Map;
import java.util.OptionalInt;
import java.util.OptionalLong;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
+import java.util.stream.Collectors;
import java.util.stream.IntStream;
import static java.util.concurrent.TimeUnit.NANOSECONDS;
@@ -99,7 +101,7 @@ public class QuorumControllerTestEnv implements AutoCloseable {
int numControllers = clientEnv.raftClients().size();
this.controllers = new ArrayList<>(numControllers);
try {
- List<Integer> nodeIds = IntStream.range(0, numControllers).boxed().toList();
+ Set<Integer> nodeIds = IntStream.range(0, numControllers).boxed().collect(Collectors.toSet());
for (int nodeId = 0; nodeId < numControllers; nodeId++) {
QuorumController.Builder builder = new QuorumController.Builder(nodeId, clientEnv.clusterId());
builder.setRaftClient(clientEnv.raftClients().get(nodeId));
@@ -111,7 +113,7 @@ public class QuorumControllerTestEnv implements AutoCloseable {
}
builder.setBootstrapMetadata(bootstrapMetadata);
builder.setLeaderImbalanceCheckIntervalNs(leaderImbalanceCheckIntervalNs);
- builder.setQuorumFeatures(new QuorumFeatures(nodeId, QuorumFeatures.defaultSupportedFeatureMap(true), nodeIds));
+ builder.setQuorumFeatures(new QuorumFeatures(nodeId, QuorumFeatures.defaultSupportedFeatureMap(true), () -> nodeIds));
sessionTimeoutMillis.ifPresent(timeout ->
builder.setSessionTimeoutNs(NANOSECONDS.convert(timeout, TimeUnit.MILLISECONDS))
);
diff --git a/metadata/src/test/java/org/apache/kafka/controller/QuorumFeaturesTest.java b/metadata/src/test/java/org/apache/kafka/controller/QuorumFeaturesTest.java
index 19fdd87b918..03e1525f42c 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/QuorumFeaturesTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/QuorumFeaturesTest.java
@@ -29,6 +29,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -43,7 +45,7 @@ public class QuorumFeaturesTest {
);
private static final QuorumFeatures QUORUM_FEATURES = new QuorumFeatures(0, LOCAL,
- List.of(0, 1, 2));
+ () -> Set.of(0, 1, 2));
@Test
public void testDefaultFeatureMap() {
@@ -109,10 +111,21 @@ public class QuorumFeaturesTest {
}
@Test
- public void testIsControllerId() {
- assertTrue(QUORUM_FEATURES.isControllerId(0));
- assertTrue(QUORUM_FEATURES.isControllerId(1));
- assertTrue(QUORUM_FEATURES.isControllerId(2));
- assertFalse(QUORUM_FEATURES.isControllerId(3));
+ public void testIsVoterId() {
+ assertTrue(QUORUM_FEATURES.isVoterId(0));
+ assertTrue(QUORUM_FEATURES.isVoterId(1));
+ assertTrue(QUORUM_FEATURES.isVoterId(2));
+ assertFalse(QUORUM_FEATURES.isVoterId(3));
+ }
+
+ @Test
+ public void testVoterIdsAreSuppliedOnEveryCall() {
+ List<Set<Integer>> voters = List.of(Set.of(0, 1, 2), Set.of(0, 1, 3));
+ AtomicInteger calls = new AtomicInteger(0);
+ QuorumFeatures quorumFeatures = new QuorumFeatures(0, LOCAL,
+ () -> voters.get(calls.getAndIncrement()));
+
+ assertEquals(Set.of(0, 1, 2), quorumFeatures.voterIds());
+ assertEquals(Set.of(0, 1, 3), quorumFeatures.voterIds());
}
}
diff --git a/metadata/src/test/java/org/apache/kafka/controller/RaftClientVotersSupplierTest.java b/metadata/src/test/java/org/apache/kafka/controller/RaftClientVotersSupplierTest.java
new file mode 100644
index 00000000000..790934a1121
--- /dev/null
+++ b/metadata/src/test/java/org/apache/kafka/controller/RaftClientVotersSupplierTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.controller;
+
+import org.apache.kafka.raft.RaftClient;
+import org.apache.kafka.raft.VoterSet;
+import org.apache.kafka.raft.VoterSetTestUtil;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Set;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class RaftClientVotersSupplierTest {
+ private static VoterSet voterSet(int... ids) {
+ return VoterSetTestUtil.voterSet(VoterSetTestUtil.voterMap(IntStream.of(ids), true));
+ }
+
+ @Test
+ public void testVoterSupplier() {
+ RaftClient<?> raftClient = Mockito.mock(RaftClient.class);
+ RaftClientVotersSupplier votersSupplier = new RaftClientVotersSupplier(raftClient);
+
+ Mockito.when(raftClient.latestVoterSet()).thenReturn(voterSet(0, 1, 2));
+ assertEquals(Set.of(0, 1, 2), votersSupplier.get());
+
+ Mockito.when(raftClient.latestVoterSet()).thenReturn(voterSet(0, 1));
+ assertEquals(Set.of(0, 1), votersSupplier.get());
+ }
+}
diff --git a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
index 71dc3c00cfe..7ecb66dad79 100644
--- a/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
+++ b/metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java
@@ -238,7 +238,7 @@ public class ReplicationControlManagerTest {
setSnapshotRegistry(snapshotRegistry).
setQuorumFeatures(new QuorumFeatures(0,
QuorumFeatures.defaultSupportedFeatureMap(true),
- List.of(0))).
+ () -> Set.of(0))).
build();
this.featureControl.replay(new FeatureLevelRecord().
setName(MetadataVersion.FEATURE_NAME).
diff --git a/metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java b/metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java
index 57fe845c98f..760706fafea 100644
--- a/metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java
+++ b/metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java
@@ -24,6 +24,7 @@ import org.apache.kafka.image.FakeSnapshotWriter;
import org.apache.kafka.image.MetadataImageTest;
import org.apache.kafka.raft.LeaderAndEpoch;
import org.apache.kafka.raft.RaftClient;
+import org.apache.kafka.raft.VoterSet;
import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.apache.kafka.server.common.KRaftVersion;
import org.apache.kafka.server.common.OffsetAndEpoch;
@@ -133,6 +134,11 @@ public class SnapshotEmitterTest {
return KRaftVersion.KRAFT_VERSION_0;
}
+ @Override
+ public VoterSet latestVoterSet() {
+ return VoterSet.empty();
+ }
+
@Override
public void upgradeKRaftVersion(int epoch, KRaftVersion version, boolean validateOnly) {
// nothing to do
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 a672106a588..91aa1ac7d90 100644
--- a/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
+++ b/raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java
@@ -3869,6 +3869,15 @@ public final class KafkaRaftClient<T> implements RaftClient<T> {
.orElseGet(partitionState::lastKraftVersion);
}
+ @Override
+ public VoterSet latestVoterSet() {
+ if (!isInitialized()) {
+ throw new IllegalStateException("Cannot read the voter set before the replica has been initialized");
+ }
+
+ return partitionState.lastVoterSet();
+ }
+
@Override
public void upgradeKRaftVersion(int epoch, KRaftVersion version, boolean validateOnly) {
if (!isInitialized()) {
diff --git a/raft/src/main/java/org/apache/kafka/raft/RaftClient.java b/raft/src/main/java/org/apache/kafka/raft/RaftClient.java
index d028fe848a3..c8ea5da31fe 100644
--- a/raft/src/main/java/org/apache/kafka/raft/RaftClient.java
+++ b/raft/src/main/java/org/apache/kafka/raft/RaftClient.java
@@ -264,6 +264,17 @@ public interface RaftClient<T> extends AutoCloseable {
*/
KRaftVersion kraftVersion();
+ /**
+ * Returns the latest set of voters, even if it hasn't been committed durably to Raft.
+ *
+ * When the kraft.version is 0, the set of voters is the one configured in
+ * {@code controller.quorum.voters} and it never changes. When the kraft.version is 1, the set
+ * of voters is the one from the latest {@code VotersRecord} in the log or snapshot.
+ *
+ * @return the latest set of voters
+ */
+ VoterSet latestVoterSet();
+
/**
* Request that the leader to upgrade the kraft version.
*
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 9c20a09b051..badf78e46c1 100644
--- a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java
@@ -346,10 +346,10 @@ public class KafkaRaftClientReconfigTest {
);
prepareLeaderToReceiveAddVoter(context, epoch, local, follower, newVoter);
+ assertEquals(Set.of(local.id(), follower.id()), context.client.latestVoterSet().voterIds());
// Attempt to add new voter to the quorum
context.deliverRequest(context.addVoterRequest(Integer.MAX_VALUE, newVoter, newListeners));
-
completeApiVersions(context, newVoter, newAddress);
// Handle the API_VERSIONS response
@@ -357,6 +357,11 @@ public class KafkaRaftClientReconfigTest {
// Append new VotersRecord to log
context.poll();
+ assertEquals(
+ Set.of(local.id(), follower.id(), newVoter.id()),
+ context.client.latestVoterSet().voterIds()
+ );
+
commitNewVoterSetForAddVoter(context, local, follower, newVoter, epoch);
// Expect reply for AddVoter request
diff --git a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
index 4a191784ac6..69869311f60 100644
--- a/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
+++ b/raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java
@@ -96,6 +96,28 @@ class KafkaRaftClientTest {
assertEquals(context.log.endOffset().offset(), context.client.logEndOffset());
}
+ @Test
+ public void testVoterSetsWithStaticVoters() throws Exception {
+ int localId = randomReplicaId();
+ ReplicaKey otherNode = replicaKey(localId + 1, false);
+ Set<Integer> voters = Set.of(localId, otherNode.id());
+ RaftClientTestContext context = new RaftClientTestContext.Builder(localId, voters).build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ // Establish the high watermark
+ context.deliverRequest(
+ context.fetchRequest(epoch, otherNode, context.log.endOffset().offset(), epoch, 0)
+ );
+ context.pollUntilResponse();
+ context.assertSentFetchPartitionResponse(Errors.NONE, epoch, OptionalInt.of(localId));
+ assertTrue(context.client.highWatermark().isPresent());
+
+ assertEquals(KRaftVersion.KRAFT_VERSION_0, context.client.kraftVersion());
+ assertEquals(voters, context.client.latestVoterSet().voterIds());
+ }
+
@ParameterizedTest
@ValueSource(booleans = { true, false })
public void testInitializeAsLeaderFromStateStoreSingleMemberQuorum(boolean withKip853Rpc) throws Exception {
diff --git a/server/src/test/java/org/apache/kafka/server/KRaftClusterTest.java b/server/src/test/java/org/apache/kafka/server/KRaftClusterTest.java
index 52ef977c634..cc94a712623 100644
--- a/server/src/test/java/org/apache/kafka/server/KRaftClusterTest.java
+++ b/server/src/test/java/org/apache/kafka/server/KRaftClusterTest.java
@@ -705,6 +705,19 @@ public class KRaftClusterTest {
cluster.waitForActiveController();
try (Admin admin = createAdminClient(cluster, usingBootstrapControllers)) {
+ // The controller is still part of the voter set, so it can't be unregistered yet
+ assertFutureThrows(
+ InvalidRequestException.class,
+ admin.unregisterController(controllerIdToUnregister).all(),
+ "Cannot unregister controller " + controllerIdToUnregister +
+ " because it is part of the voter set."
+ );
+
+ admin.removeRaftVoter(
+ controllerIdToUnregister,
+ initialVoters.get(controllerIdToUnregister)
+ ).all().get();
+
assertDoesNotThrow(() -> admin.unregisterController(controllerIdToUnregister).all().get());
}
@@ -730,6 +743,11 @@ public class KRaftClusterTest {
.map(Map.Entry::getKey)
.findFirst()
.orElseThrow();
+ // The voter set is static in this cluster, so every controller is a voter
+ int inactiveId = cluster.controllers().keySet().stream()
+ .filter(id -> id != activeId)
+ .findFirst()
+ .orElseThrow();
try (Admin admin = createAdminClient(cluster, usingBootstrapControllers)) {
assertFutureThrows(
@@ -740,7 +758,12 @@ public class KRaftClusterTest {
assertFutureThrows(
InvalidRequestException.class,
admin.unregisterController(activeId).all(),
- "Controller cannot unregister itself while it is active."
+ "Cannot unregister controller " + activeId + " because it is part of the voter set."
+ );
+ assertFutureThrows(
+ InvalidRequestException.class,
+ admin.unregisterController(inactiveId).all(),
+ "Cannot unregister controller " + inactiveId + " because it is part of the voter set."
);
}
}