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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
