You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kevin-wu24 (via GitHub)" <gi...@apache.org> on 2026/10/23 17:34:11 UTC
Re: [PR] KAFKA-18775: Simplify controller quorum changes through the Admin API [kafka]
kevin-wu24 commented on code in PR #22531:
URL: https://github.com/apache/kafka/pull/22531#discussion_r4361088980
##########
clients/src/main/java/org/apache/kafka/common/requests/AddRaftVoterRequest.java:
##########
@@ -34,6 +36,12 @@ public Builder(AddRaftVoterRequestData data) {
@Override
public AddRaftVoterRequest build(short version) {
+ if (version < 2 && data.voterDirectoryId().equals(Uuid.ZERO_UUID)) {
+ throw new UnsupportedVersionException("Version 2 or later is required to omit voterDirectoryId");
+ }
+ if (version < 2 && data.listeners().isEmpty()) {
+ throw new UnsupportedVersionException("Version 2 or later is required to omit listeners");
+ }
Review Comment:
These exception messages are a bit confusing. Especially since it does not appear it is required to omit these fields from the code in `KafkaRaftClient`, and the KIP. How about doing something similar to how `CreateTopicsRequest#build` does validation:
```
public CreateTopicsRequest build(short version) {
if (data.validateOnly() && version == 0)
throw new UnsupportedProtocolFieldException("validateOnly", apiKey().name(), version, 1);
```
So for this code something like:
```suggestion
if (version < 2 && data.voterDirectoryId().equals(Uuid.ZERO_UUID)) {
throw new UnsupportedProtocolFieldException("A non-zero voterDirectoryId is required in AddRaftVoter version 0 or 1. Upgrade the cluster to support AddRaftVoter version 2 to support a zero voterDirectoryId.");
}
if (version < 2 && data.listeners().isEmpty()) {
throw new UnsupportedProtocolFieldException("A non-empty listeners required in AddRaftVoter version 0 or 1. Upgrade the cluster to support AddRaftVoter version 2 to support a zero voterDirectoryId.");
}
```
I see that we do this for `AddRaftVoterRequest#build`, but not for `RemoveRaftVoterRequest#build`, can we be consistent? I see that other requests like create topics and fetch have these build time validations, so I am not opposed to add a validation here for v0 and v1.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2278,39 +2283,84 @@ private CompletionStage<AddRaftVoterResponseData> handleAddVoterRequest(
);
}
- Optional<ReplicaKey> newVoter = RaftUtil.addVoterRequestVoterKey(data);
- if (newVoter.isEmpty() || newVoter.get().directoryId().isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage("Add voter request didn't include a valid voter")
- );
- }
-
- Endpoints newVoterEndpoints = Endpoints.fromAddVoterRequest(data.listeners());
- if (newVoterEndpoints.address(channel.listenerName()).isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage(
- String.format(
- "Add voter request didn't include the endpoint (%s) for the default listener %s",
- newVoterEndpoints,
- channel.listenerName()
- )
- )
- );
+ final ReplicaKey newVoter;
+ final Endpoints newVoterEndpoints;
+ try {
+ newVoter = resolveAddVoterKey(data, requestMetadata.apiVersion(), currentTimeMs);
+ newVoterEndpoints = resolveAddVoterEndpoints(data, requestMetadata.apiVersion());
+ } catch (IllegalArgumentException e) {
+ return completedFuture(RaftUtil.addVoterResponse(Errors.INVALID_REQUEST, e.getMessage()));
}
return addVoterHandler.handleAddVoterRequest(
quorum.leaderStateOrThrow(),
- newVoter.get(),
+ newVoter,
newVoterEndpoints,
data.ackWhenCommitted(),
currentTimeMs
);
}
+ private ReplicaKey resolveAddVoterKey(
Review Comment:
It would be good to document this behavior for the below cases in the PR description. I see it in the KIP: https://cwiki.apache.org/confluence/spaces/KAFKA/pages/340037949/KIP-1141+Simplifying+Add+Remove+Voter+in+MetadataQuorumCommand#KIP1141%3ASimplifyingAdd%2FRemoveVoterinMetadataQuorumCommand-Serversidechanges.
- AddRaftVoter v2 and the `voterDirectoryId` is specified, but the `endpoints` is not.
- AddRaftVoter v2 and the `endpoints` is specified, but the `voterDirectoryId` is not.
- AddRaftVoter v2 and both `endpoints` and `voterDirectoryId` are specified.
- RemoveRaftVoter v1 and `voterDirectoryId` is specified.
From reading the code here, it seems like if these fields are specified in the request, that takes precedence in AddRaftVoter v2 and RemoveRaftVoter v1, and only if they are empty/not present, do we default on what the `nodeProvider` says. This behavior seems reasonable to me, but we should document it because it is not clear without reading the code.
##########
clients/src/testFixtures/java/org/apache/kafka/clients/admin/MockAdminClient.java:
##########
@@ -1451,11 +1451,25 @@ public ListClientMetricsResourcesResult listClientMetricsResources(ListClientMet
}
@Override
+ public AddRaftVoterResult addRaftVoter(int voterId, AddRaftVoterOptions options) {
+ throw new UnsupportedOperationException("Not implemented yet");
+ }
+
+ @Override
+ @SuppressWarnings("removal")
+ @Deprecated(since = "4.4", forRemoval = true)
Review Comment:
I think this needs to be updated to 4.5.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2278,39 +2283,84 @@ private CompletionStage<AddRaftVoterResponseData> handleAddVoterRequest(
);
}
- Optional<ReplicaKey> newVoter = RaftUtil.addVoterRequestVoterKey(data);
- if (newVoter.isEmpty() || newVoter.get().directoryId().isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage("Add voter request didn't include a valid voter")
- );
- }
-
- Endpoints newVoterEndpoints = Endpoints.fromAddVoterRequest(data.listeners());
- if (newVoterEndpoints.address(channel.listenerName()).isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage(
- String.format(
- "Add voter request didn't include the endpoint (%s) for the default listener %s",
- newVoterEndpoints,
- channel.listenerName()
- )
- )
- );
+ final ReplicaKey newVoter;
+ final Endpoints newVoterEndpoints;
+ try {
+ newVoter = resolveAddVoterKey(data, requestMetadata.apiVersion(), currentTimeMs);
+ newVoterEndpoints = resolveAddVoterEndpoints(data, requestMetadata.apiVersion());
+ } catch (IllegalArgumentException e) {
+ return completedFuture(RaftUtil.addVoterResponse(Errors.INVALID_REQUEST, e.getMessage()));
}
return addVoterHandler.handleAddVoterRequest(
quorum.leaderStateOrThrow(),
- newVoter.get(),
+ newVoter,
newVoterEndpoints,
data.ackWhenCommitted(),
currentTimeMs
);
}
+ private ReplicaKey resolveAddVoterKey(
+ AddRaftVoterRequestData data,
+ short apiVersion,
+ long currentTimeMs
+ ) {
+ if (apiVersion >= 2 && data.voterDirectoryId().equals(Uuid.ZERO_UUID)) {
+ List<ReplicaKey> matchingObservers = quorum.leaderStateOrThrow()
+ .observerStates(currentTimeMs)
+ .keySet()
+ .stream()
+ .filter(key -> key.id() == data.voterId())
+ .toList();
+ if (matchingObservers.size() > 1) {
+ throw new IllegalArgumentException(String.format(
+ "Multiple observers with node ID %d were found: %s. " +
+ "This can happen if the node restarted with a new log directory while a previous " +
+ "observer remained inactive. Remove stale or inactive observer entries so that " +
Review Comment:
Also, it would be nice if we could add a unit test case for this scenario, mainly to ensure the active controller does not remain in a state where it cannot add this node's voter id when there are multiple matching observers.
Here is what the test could look like at a high level:
- The test would set up the local node as the leader.
- Deliver a fetch from the "old" node incarnation.
- Deliver a fetch from the "new" node incarnation. This represents the old node getting its log replaced and re-fetching.
- Try `AddRaftVoter` v2. This should fail with this exception.
- Advance time to expire the observer session of the old node.
- Try `AddRaftVoter` v2. Now the request succeeds.
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java:
##########
@@ -4213,4 +4214,42 @@ public void testStreamsGroupDescribeResponseV0RejectsTopologyDescriptionFields()
StreamsGroupDescribeResponse response = new StreamsGroupDescribeResponse(data);
assertThrows(UnsupportedVersionException.class, () -> response.serialize((short) 0));
}
+
+ @Test
+ public void testAddRaftVoterRequestBeforeV2RejectsZeroUUID() {
+ AddRaftVoterRequest.Builder builder = new AddRaftVoterRequest.Builder(
+ new AddRaftVoterRequestData()
+ .setClusterId("cluster-id")
+ .setTimeoutMs(1000)
+ .setVoterId(1)
+ .setVoterDirectoryId(Uuid.ZERO_UUID)
+ .setListeners(new AddRaftVoterRequestData.ListenerCollection())
+ );
+
+ String errorMsg = "Version 2 or later is required to omit voterDirectoryId";
Review Comment:
This test needs to be updated after my other suggestion. It may be easier to just check the exception class type.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2278,39 +2283,84 @@ private CompletionStage<AddRaftVoterResponseData> handleAddVoterRequest(
);
}
- Optional<ReplicaKey> newVoter = RaftUtil.addVoterRequestVoterKey(data);
- if (newVoter.isEmpty() || newVoter.get().directoryId().isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage("Add voter request didn't include a valid voter")
- );
- }
-
- Endpoints newVoterEndpoints = Endpoints.fromAddVoterRequest(data.listeners());
- if (newVoterEndpoints.address(channel.listenerName()).isEmpty()) {
- return completedFuture(
- new AddRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage(
- String.format(
- "Add voter request didn't include the endpoint (%s) for the default listener %s",
- newVoterEndpoints,
- channel.listenerName()
- )
- )
- );
+ final ReplicaKey newVoter;
+ final Endpoints newVoterEndpoints;
+ try {
+ newVoter = resolveAddVoterKey(data, requestMetadata.apiVersion(), currentTimeMs);
+ newVoterEndpoints = resolveAddVoterEndpoints(data, requestMetadata.apiVersion());
+ } catch (IllegalArgumentException e) {
+ return completedFuture(RaftUtil.addVoterResponse(Errors.INVALID_REQUEST, e.getMessage()));
}
return addVoterHandler.handleAddVoterRequest(
quorum.leaderStateOrThrow(),
- newVoter.get(),
+ newVoter,
newVoterEndpoints,
data.ackWhenCommitted(),
currentTimeMs
);
}
+ private ReplicaKey resolveAddVoterKey(
+ AddRaftVoterRequestData data,
+ short apiVersion,
+ long currentTimeMs
+ ) {
+ if (apiVersion >= 2 && data.voterDirectoryId().equals(Uuid.ZERO_UUID)) {
+ List<ReplicaKey> matchingObservers = quorum.leaderStateOrThrow()
+ .observerStates(currentTimeMs)
+ .keySet()
+ .stream()
+ .filter(key -> key.id() == data.voterId())
+ .toList();
+ if (matchingObservers.size() > 1) {
+ throw new IllegalArgumentException(String.format(
+ "Multiple observers with node ID %d were found: %s. " +
+ "This can happen if the node restarted with a new log directory while a previous " +
+ "observer remained inactive. Remove stale or inactive observer entries so that " +
Review Comment:
```suggestion
"observer remained active. Remove stale or inactive observer entries so that " +
```
Maybe it is worth printing for how long the old incarnation of the node remains an active observer by printing:
```
OBSERVER_SESSION_TIMEOUT_MS - (currentTimeMs - integerReplicaStateEntry.getValue().lastFetchTimestamp)
```
Also, there is no way to remove the observer entry in the leader state. The only way to clear this is to wait for the observer session to expire or the leader to change and the new node to fetch from it (assuming the old incarnation does not fetch to the new leader too).
##########
server/src/test/java/org/apache/kafka/server/BootstrapControllersIntegrationTest.java:
##########
@@ -205,6 +212,119 @@ private void testDescribeMetadataQuorum(ClusterInstance clusterInstance, boolean
}
}
+ @ClusterTest(controllers = 3, standalone = true)
Review Comment:
More of a high level question, but why not put these tests in `ReconfigurableQuorumIntegrationTest`? That file seems more appropriate to me. It seems the purpose of these tests are to test that the `--bootstrap-controllers` flags works for CLIs. Meaning, that when clients send requests to the brokers, forwarding them to the controller vs sending the request to the controller directly should not change the behavior of the request.
I think the `kafka-metadata-quorum` CLI is already covered by the `testDescribeMetadataQuorum` test above right?
##########
server/src/test/java/org/apache/kafka/server/ReconfigurableQuorumIntegrationTest.java:
##########
@@ -109,6 +109,7 @@ private static Map<Integer, Uuid> findVoterDirs(Admin admin) throws Exception {
}
@Test
+ @SuppressWarnings("removal")
Review Comment:
Why do we need this annotation? Is it because it is using a deprecated method? Can we instead make the call to `admin.remove/addRaftVoter` use the updated methods?
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -2435,22 +2485,38 @@ private CompletionStage<RemoveRaftVoterResponseData> handleRemoveVoterRequest(
);
}
- Optional<ReplicaKey> oldVoter = RaftUtil.removeVoterRequestVoterKey(data);
- if (oldVoter.isEmpty() || oldVoter.get().directoryId().isEmpty()) {
- return completedFuture(
- new RemoveRaftVoterResponseData()
- .setErrorCode(Errors.INVALID_REQUEST.code())
- .setErrorMessage("Remove voter request didn't include a valid voter")
- );
+ final ReplicaKey oldVoter;
+ try {
+ oldVoter = resolveRemoveVoterKey(data, requestMetadata.apiVersion());
+ } catch (IllegalArgumentException e) {
+ return completedFuture(RaftUtil.removeVoterResponse(Errors.INVALID_REQUEST, e.getMessage()));
}
return removeVoterHandler.handleRemoveVoterRequest(
quorum.leaderStateOrThrow(),
- oldVoter.get(),
+ oldVoter,
currentTimeMs
);
}
+ private ReplicaKey resolveRemoveVoterKey(RemoveRaftVoterRequestData data, short apiVersion) {
+ if (apiVersion >= 1 && data.voterDirectoryId().equals(Uuid.ZERO_UUID)) {
+ return partitionState.lastVoterSet()
+ .voterKey(data.voterId())
+ .filter(voterKey -> voterKey.directoryId().isPresent())
+ .orElseThrow(() -> new IllegalArgumentException(
+ String.format(
+ "Could not find voter with node ID %d in the current voter set to derive its directory ID",
Review Comment:
There is a convention in the `raft` module to keep line lengths less than 100 characters. Here is what claude says with line numbers which lines violate this convention in your changes:
```
KafkaRaftClient.java: 9 lines, all in the new add/remove voter resolve code
- :2292 (102): return completedFuture(RaftUtil.addVoterResponse(Errors.INVALID_REQUEST, e.getMessage()));
- :2319 (108) and :2320 (106): the "Multiple observers with node ID…" message strings
- :2325 (105): } else if (matchingObservers.isEmpty() || matchingObservers.get(0).directoryId().isEmpty()) {
- :2328 (105): the "Could not find a unique observer with node ID…" message string
- :2340 (111): .orElseThrow(() -> new IllegalArgumentException("Add voter request didn't include a valid voter"));
- :2492 (105): return completedFuture(RaftUtil.removeVoterResponse(Errors.INVALID_REQUEST, e.getMessage()));
- :2509 (115): the "Could not find voter with node ID %d in the current voter set…" message string
- :2517 (114): .orElseThrow(() -> new IllegalArgumentException("Remove voter request didn't include a valid voter"));
KafkaRaftClientReconfigTest.java: 12 lines over 100, plus 1 at exactly 100
- :375 (132): the testAddVoterDerivesMissingDirectoryIdAndEndpoints(boolean omitDirectoryId, boolean omitEndpoints) throws Exception { signature
- :388, :428, :477, :1318, :1370 (112 each): RaftClientTestContext context = new RaftClientTestContext.Builder(local.id(), local.directoryId().get())
- :391 (107): .withNodeEndpointProvider(nodeId -> nodeId == newVoter.id() ? newListeners : Endpoints.empty())
- :489 (102): the "Add two caught-up observer replica keys…" comment
- :493 (102): context.fetchRequest(epoch, duplicateObserver, context.log.endOffset().offset(), epoch, 0)
- :507 (exactly 100): String errorMsg = context.assertSentAddVoterResponse(Errors.INVALID_REQUEST).errorMessage();
- :508 (120): assertTrue(errorMsg.startsWith(String.format("Multiple observers with node ID %d were found:", …)));
- :510 (123) and :1394 (107): expected error-message strings
```
##########
tools/src/main/java/org/apache/kafka/tools/MetadataQuorumCommand.java:
##########
@@ -145,13 +148,19 @@ static void execute(String... args) throws Exception {
}
}
case "add-controller" -> {
- if (optionalCommandConfig == null) {
- throw new TerseException("You must supply the configuration file of the controller you are " +
- "adding when using add-controller.");
+ Integer controllerId = namespace.getInt("controller_id");
+ if (controllerId != null) {
+ handleAddControllerById(admin,
+ namespace.getBoolean("dry_run"),
+ controllerId);
+ } else if (optionalCommandConfig != null) {
Review Comment:
Should these flags be mutually exclusive? Meaning, is the below a valid invocation:
```
kafka-metadata-quorum add-controller --bootstrap-server <> --controller-id 1 --command-config <>
```
In the current code, you will run `handleAddControllerById` and ignore the second flag.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org