kevin-wu24 commented on code in PR #22531:
URL: https://github.com/apache/kafka/pull/22531#discussion_r3482865982
##########
core/src/main/scala/kafka/server/ControllerServer.scala:
##########
@@ -206,6 +207,16 @@ class ControllerServer(
sharedServer.startForController(listenerInfo)
Review Comment:
This line needs to execute after we set the `NodeEndpointProvider` if we
keep the current approach, not before. When returning from this line, raft
network client and io thread have already started doing work.
##########
server/src/test/java/org/apache/kafka/server/BootstrapControllersIntegrationTest.java:
##########
@@ -205,6 +208,51 @@ private void testDescribeMetadataQuorum(ClusterInstance
clusterInstance, boolean
}
}
+ @ClusterTest(controllers = 3, standalone = true)
+ public void testAddRemoveRaftVoterByControllers(ClusterInstance
clusterInstance) throws Exception {
+ testAddRemoveRaftVoter(clusterInstance, true);
+ }
+
+ @ClusterTest(controllers = 3, standalone = true)
+ public void testAddRemoveRaftVoter(ClusterInstance clusterInstance) throws
Exception {
+ testAddRemoveRaftVoter(clusterInstance, false);
+ }
Review Comment:
We can parametrize this test based on the boolean
`usingBootstrapControllers` right?
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java:
##########
@@ -364,6 +364,170 @@ public void testAddVoter() throws Exception {
context.assertSentAddVoterResponse(Errors.NONE);
}
+ @Test
+ public void testAddVoterDerivesDirectoryIdAndEndpoints() 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_1141_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ ReplicaKey newVoter = replicaKey(local.id() + 2, true);
+ InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
+ "localhost",
+ 9990 + newVoter.id()
+ );
+ Endpoints newListeners = Endpoints.fromInetSocketAddresses(
+ Map.of(context.channel.listenerName(), newAddress)
+ );
+ context.client.setNodeEndpointProvider(nodeId ->
+ nodeId == newVoter.id() ? newListeners : Endpoints.empty()
+ );
+
+ prepareLeaderToReceiveAddVoter(context, epoch, local, follower,
newVoter);
+
+ context.deliverRequest(
+ context.addVoterRequest(
+ Integer.MAX_VALUE,
+ ReplicaKey.of(newVoter.id(), Uuid.ZERO_UUID),
+ Endpoints.empty()
+ )
+ );
+
+ completeApiVersionsForAddVoter(context, newVoter, newAddress);
+
+ // Handle the API_VERSIONS response
+ context.client.poll();
+ // Append new VotersRecord to log
+ context.client.poll();
+
+ commitNewVoterSetForAddVoter(context, local, follower, newVoter,
epoch);
+
+ // Expect reply for AddVoter request
+ context.pollUntilResponse();
+ context.assertSentAddVoterResponse(Errors.NONE);
+ }
+
+ @Test
+ public void testAddVoterFailsToDeriveDefaultListenerEndpoint() 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_1141_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+ ReplicaKey newVoter = replicaKey(local.id() + 2, true);
+
+ prepareLeaderToReceiveAddVoter(context, epoch, local, follower,
newVoter);
+
+ context.deliverRequest(
+ context.addVoterRequest(
+ Integer.MAX_VALUE,
+ ReplicaKey.of(newVoter.id(), Uuid.ZERO_UUID),
+ Endpoints.empty()
+ )
+ );
+
+ context.pollUntilResponse();
+ context.assertSentAddVoterResponse(Errors.INVALID_REQUEST);
+ }
+
+ @Test
+ public void testAddVoterDoesNotReplaceExplicitDirectoryId() 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_1141_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+ ReplicaKey observer = replicaKey(local.id() + 2, true);
+ ReplicaKey requestedVoter = replicaKey(observer.id(), true);
+ InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
+ "localhost",
+ 9990 + observer.id()
+ );
+ Endpoints newListeners = Endpoints.fromInetSocketAddresses(
+ Map.of(context.channel.listenerName(), newAddress)
+ );
+
+ prepareLeaderToReceiveAddVoter(context, epoch, local, follower,
observer);
+ context.deliverRequest(
+ context.addVoterRequest(Integer.MAX_VALUE, requestedVoter,
newListeners)
+ );
+ completeApiVersionsForAddVoter(context, requestedVoter, newAddress);
+
+ context.pollUntilResponse();
+ context.assertSentAddVoterResponse(Errors.REQUEST_TIMED_OUT);
Review Comment:
I think you're hitting this code in
`AddVoterHandler#handleApiVersionsResponse`:
```
if (!leaderState.isReplicaCaughtUp(current.voterKey(), currentTimeMs)) {
logger.info(
"Aborting add voter operation for {} at {} since it is
lagging behind: {}",
current.voterKey(),
current.voterEndpoints(),
leaderState.getReplicaState(current.voterKey())
);
```
We do a fetch via the `observer` replica key to catch it up, but not the
`requestedVoter` replica key.
Ideally, putting the wrong directory id in the `ADD_RAFT_VOTER` request
should not even get to this point. We should return `INVALID_REQUEST` before
even entering `AddVoterHandler`. We should be returning a similar error to the
test below.
Because you're running `add-voter` from the node that is being added (in
both manual + auto join case), the directory id would be correct or
`ZERO_UUID`, but never incorrect.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java:
##########
@@ -364,6 +364,170 @@ public void testAddVoter() throws Exception {
context.assertSentAddVoterResponse(Errors.NONE);
}
+ @Test
+ public void testAddVoterDerivesDirectoryIdAndEndpoints() 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_1141_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ ReplicaKey newVoter = replicaKey(local.id() + 2, true);
+ InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
+ "localhost",
+ 9990 + newVoter.id()
+ );
+ Endpoints newListeners = Endpoints.fromInetSocketAddresses(
+ Map.of(context.channel.listenerName(), newAddress)
+ );
+ context.client.setNodeEndpointProvider(nodeId ->
+ nodeId == newVoter.id() ? newListeners : Endpoints.empty()
+ );
+
+ prepareLeaderToReceiveAddVoter(context, epoch, local, follower,
newVoter);
+
+ context.deliverRequest(
+ context.addVoterRequest(
+ Integer.MAX_VALUE,
+ ReplicaKey.of(newVoter.id(), Uuid.ZERO_UUID),
+ Endpoints.empty()
+ )
+ );
+
+ completeApiVersionsForAddVoter(context, newVoter, newAddress);
+
+ // Handle the API_VERSIONS response
+ context.client.poll();
+ // Append new VotersRecord to log
+ context.client.poll();
+
+ commitNewVoterSetForAddVoter(context, local, follower, newVoter,
epoch);
+
+ // Expect reply for AddVoter request
+ context.pollUntilResponse();
+ context.assertSentAddVoterResponse(Errors.NONE);
+ }
+
+ @Test
+ public void testAddVoterFailsToDeriveDefaultListenerEndpoint() 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_1141_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+ ReplicaKey newVoter = replicaKey(local.id() + 2, true);
+
+ prepareLeaderToReceiveAddVoter(context, epoch, local, follower,
newVoter);
+
+ context.deliverRequest(
+ context.addVoterRequest(
+ Integer.MAX_VALUE,
+ ReplicaKey.of(newVoter.id(), Uuid.ZERO_UUID),
+ Endpoints.empty()
+ )
+ );
+
+ context.pollUntilResponse();
+ context.assertSentAddVoterResponse(Errors.INVALID_REQUEST);
+ }
+
+ @Test
+ public void testAddVoterDoesNotReplaceExplicitDirectoryId() throws
Exception {
Review Comment:
I am kind of confused about the purpose of this test. We have an observer
replica key that the leader should add to `observerStates` because we send a
fetch from that key in `prepareLeaderToReceiveAddVoter`. Then we send
`ADD_RAFT_VOTER` with the same observer id but a different directory ID, is
that correct?
It still seems to me that even after this KIP, the malformed
`ADD_RAFT_VOTER` with the wrong directory UUID delivered on L474 is not handled
by the existing code.
##########
raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientReconfigTest.java:
##########
@@ -364,6 +364,170 @@ public void testAddVoter() throws Exception {
context.assertSentAddVoterResponse(Errors.NONE);
}
+ @Test
+ public void testAddVoterDerivesDirectoryIdAndEndpoints() 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_1141_PROTOCOL)
+ .withBootstrapSnapshot(Optional.of(voters))
+ .withUnknownLeader(3)
+ .build();
+
+ context.unattachedToLeader();
+ int epoch = context.currentEpoch();
+
+ ReplicaKey newVoter = replicaKey(local.id() + 2, true);
+ InetSocketAddress newAddress = InetSocketAddress.createUnresolved(
+ "localhost",
+ 9990 + newVoter.id()
+ );
+ Endpoints newListeners = Endpoints.fromInetSocketAddresses(
+ Map.of(context.channel.listenerName(), newAddress)
+ );
+ context.client.setNodeEndpointProvider(nodeId ->
+ nodeId == newVoter.id() ? newListeners : Endpoints.empty()
+ );
Review Comment:
We should add a `withNodeEndpointProvider` method to the
`RaftClientTestContext.Builder` class instead.
##########
server/src/test/java/org/apache/kafka/server/BootstrapControllersIntegrationTest.java:
##########
@@ -205,6 +208,51 @@ private void testDescribeMetadataQuorum(ClusterInstance
clusterInstance, boolean
}
}
+ @ClusterTest(controllers = 3, standalone = true)
+ public void testAddRemoveRaftVoterByControllers(ClusterInstance
clusterInstance) throws Exception {
+ testAddRemoveRaftVoter(clusterInstance, true);
+ }
+
+ @ClusterTest(controllers = 3, standalone = true)
+ public void testAddRemoveRaftVoter(ClusterInstance clusterInstance) throws
Exception {
+ testAddRemoveRaftVoter(clusterInstance, false);
+ }
+
+ private void testAddRemoveRaftVoter(ClusterInstance clusterInstance,
boolean usingBootstrapControllers) throws Exception {
+ try (Admin admin = Admin.create(adminConfig(clusterInstance,
usingBootstrapControllers))) {
+ Set<Integer> initialVoters =
voterIds(admin.describeMetadataQuorum().quorumInfo().get());
+ AtomicInteger voterId = new AtomicInteger();
+ TestUtils.retryOnExceptionWithTimeout(30_000, () -> {
+ QuorumInfo quorumInfo =
admin.describeMetadataQuorum().quorumInfo().get();
+ voterId.set(quorumInfo.observers().stream()
Review Comment:
How did you set up an observer controller in the integration tests? This is
using the `KafkaClusterTestKit` under the hood right? Mostly curious, since I
don't remember that code being able to set up kraft observers that were not
brokers.
##########
core/src/main/scala/kafka/server/ControllerServer.scala:
##########
@@ -206,6 +207,16 @@ class ControllerServer(
sharedServer.startForController(listenerInfo)
+ raftManager.client.setNodeEndpointProvider { nodeId =>
+ Option(registrationsPublisher.controllers().get(nodeId)).map {
registration =>
+ val endpoints = registration.listeners().asScala.map { case
(listenerName, endpoint) =>
+ ListenerName.normalised(listenerName) ->
+ InetSocketAddress.createUnresolved(endpoint.host(),
endpoint.port())
+ }.asJava
+ Endpoints.fromInetSocketAddresses(endpoints)
+ }.getOrElse(Endpoints.empty())
+ }
+
Review Comment:
It would be nice if we did not introduce `public` methods to
`KafkaRaftClient` that are not part of the `RaftClient` API. This method seems
very implementation-specific. I think the issue is that it is not easy to get
controller server state into the raft layer.
Is it possible to have the raft client constructor take a
`NodeEndpointProvider` instead as a parameter?
In `ControllerServer`, I see:
```
registrationsPublisher = new ControllerRegistrationsPublisher()
```
We could initialize the publisher in `SharedServer` if we are a controller,
define the `NodeEndpointProvider` there for based on whether we are broker or
controller, and then pass it to `KafkaRaftManager`. What do you think?
--
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]
