KAFKA-16525; Dynamic KRaft network manager and channel #15986

jsancio · 2024-05-17T16:53:02Z

Allow KRaft replicas to send requests to any node (Node) not just the nodes configured in the controller.quorum.voters property. This flexibility is needed so KRaft can implement the controller.quorum.voters configuration, send request to the dynamically changing set of voters and send request to the leader endpoint (Node) discovered through the KRaft RPCs (specially BeginQuorumEpoch request and Fetch response).

This was achieved by changing the RequestManager API to accept Node instead of just the replica ID. Internally, the request manager tracks connection state using the Node.idString method to match the connection management used by NetworkClient.

The API for RequestManager is also changed so that the ConnectState class is not exposed in the API. This allows the request manager to reclaim heap memory for any connection that is ready.

The NetworkChannel was updated to receive the endpoint information (Node) through the outbound raft request (RaftRequent.Outbound). This makes the network channel more flexible as it doesn't need to be configured with the list of all possible endpoints. RaftRequest.Outbound and RaftResponse.Inbound were updated to include the remote node instead of just the remote id.

The follower state tracked by KRaft replicas was updated to include both the leader id and the leader's endpoint (Node). In this comment the node value is computed from the set of voters. In future commit this will be updated so that it is sent through KRaft RPCs. For example BeginQuorumEpoch request and Fetch response.

Support for configuring controller.quorum.bootstrap.servers was added. This includes changes to KafkaConfig, QuorumConfig, etc. All of the tests using QuorumTestHarness were changed to use the controller.quorum.bootstrap.servers instead of the controller.quorum.voters for the broker configuration. Finally, the node id for the bootstrap server will be decreasing negative numbers starting with -2.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

…c-raft-channel

hachikuji · 2024-05-22T17:36:36Z

core/src/test/resources/log4j.properties

@@ -20,6 +20,8 @@ log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n

 log4j.logger.kafka=WARN
 log4j.logger.org.apache.kafka=WARN
+# TODO; remove this line


Reminder about TODO

hachikuji · 2024-05-22T17:38:22Z

core/src/test/scala/unit/kafka/raft/RaftManagerTest.scala

@@ -118,6 +120,10 @@ class RaftManagerTest {
      new Metrics(Time.SYSTEM),
      Option.empty,
      CompletableFuture.completedFuture(QuorumConfig.parseVoterConnections(config.quorumVoters)),
+      config.quorumBootstrapServers


nit: we seem to have this same little snippet in a few places. Is there somewhere we could add a helper?

hachikuji · 2024-05-22T17:39:05Z

core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala

@@ -1366,6 +1368,26 @@ class KafkaConfigTest {
    assertEquals(expectedVoters, addresses)
  }

+  @Test
+  def testParseQuorumBootstrapServers(): Unit = {


Perhaps we should have some invalid test cases as well?

Okay. It does look like there are bugs in the implementation of Utils.getHost and Utils.getPort. I don't really want to fix them in this PR. I created this issue: https://issues.apache.org/jira/browse/KAFKA-16824

hachikuji · 2024-05-22T17:40:09Z

raft/src/main/java/org/apache/kafka/raft/FollowerState.java

-    private Optional<RawSnapshotWriter> fetchingSnapshot;
+    private Optional<RawSnapshotWriter> fetchingSnapshot = Optional.empty();
+
+    // TODO: remove this when done debuggin CI failures


TODO reminder

hachikuji · 2024-05-22T17:45:51Z

raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java

+
+            // Attempt to parse the connect strings
+            for (String entry : entries) {
+                QuorumConfig.parseBootstrapServer(entry);


nit: drop unnecessary QuorumConfig prefix?

hachikuji · 2024-05-22T17:47:07Z

raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java

+            );
+        }
+
+        return InetSocketAddress.createUnresolved(host, port);


Is this going to give us a nice exception message if the host is invalid?

I am not sure. I know that it throws an IllegalArgumentException if host is null or port is outside the valid range. I think it is better for Utils.getHost and Utils.getPort to better catch invalid host and port values so we can throw the corresponding ConfigException. I filed https://issues.apache.org/jira/browse/KAFKA-16824 to improve the implementation of those methods.

I think it's a mistake to resolve hostnames as part of configuration validation. Just treat them as strings.

Otherwise you could end up not being able to start Kafka because 1 out of 3 hostnames in your bootstrap list could not be resolved (because of a DNS problem or whatever)

Resolve a hostname when you're actually using it, not before.

I agree. I have been fixing any code that resolves host names outside of the NetworkClient. This PR should not be resolving host names outside of the NetworkClient. Let me know if you find a case that violates this.

oh, I thought Utils.getHost resolved the hostname. I can see now that it doesn't. Fair enough.

hachikuji · 2024-05-22T17:58:42Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -1234,7 +1286,22 @@ private boolean handleFetchResponse(
            } else {
                Records records = FetchResponse.recordsOrFail(partitionResponse);
                if (records.sizeInBytes() > 0) {
-                    appendAsFollower(records);
+                    // TODO: remove all of this code when done debugging


TODO reminder

hachikuji · 2024-05-22T18:04:29Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

+                )
+                .collect(Collectors.toList());
+
+            logger.info("Starting request manager with bootstrap servers: {}", bootstrapNodes);


Perhaps we change this message to something like "Staring request manager with static voter set: {}"

hachikuji · 2024-05-22T18:20:31Z

raft/src/main/java/org/apache/kafka/raft/RequestManager.java

+        return state.isResponseExpected(correlationId);
+    }
+
+    public void onResponseReceived(Node node, long correlationId) {


I wonder if we could consolidate onResponseReceived and onResponseError. Perhaps something like notifyResponseResult or something like that.

Sounds good. Went with onResponseResult(Node, long, boolean, long).

hachikuji · 2024-05-22T18:21:00Z

raft/src/main/java/org/apache/kafka/raft/internals/VoterSet.java

-     * @param voter the id of the voter
-     * @param listener the name of the listener
-     * @return the socket address if it exists, otherwise {@code Optional.empty()}
+     * @param voterIds the id of the voters


nit: ids of the voters

…c-raft-channel

jsancio · 2024-05-27T23:24:12Z

@hachikuji, thanks for the review. The PR is ready for another round.

The core of the issue that I fixed was that KRaft was sending two Fetch requests for the same LEO. One of the Fetch request was going to the leader using the Node with an id greater than or equal to 0. The second Fetch request was going to the leader using the Node in the bootstrap server list which had a negative id.

I fixed the issue by updating the RequestManager and KafkaRaftClient to only allow one outstanding (Fetch) request.

cmccabe · 2024-05-29T21:41:21Z

core/src/main/scala/kafka/server/SharedServer.scala

@@ -94,6 +95,7 @@ class SharedServer(
  val time: Time,
  private val _metrics: Metrics,
  val controllerQuorumVotersFuture: CompletableFuture[JMap[Integer, InetSocketAddress]],
+  val bootstrapServers: JCollection[InetSocketAddress],


I don't see why this needs to be a new argument. We already have KafkaConfig passed in as an argument, and this just comes from QuorumConfig.parseBootstrapServers(config.quorumBootstrapServers).

Adding a new argument here creates a lot of churn, which is a lot of work for merges and such.

It's the same reason why the controllerQuorumVotersFuture is passed as an argument and it is not read from the KafkaConfig. Tests tend to configure the voters using the 0 port so that the OS can dynamically assign an open port. We need the same functionality for bootstrapServers in most tests the bootstrap server are the same se of endpoint as the voters but without the node id.

Having said that I have plans to remove all of these complexity after 3.8. This is mainly here before of kafka raft cluster test kit. I created this issue to track this work: https://issues.apache.org/jira/browse/KAFKA-16653

cmccabe · 2024-05-29T21:41:49Z

core/src/test/scala/integration/kafka/api/PlaintextConsumerTest.scala

@@ -21,12 +21,12 @@ import kafka.utils.{TestInfoUtils, TestUtils}
 import org.apache.kafka.clients.admin.{NewPartitions, NewTopic}
 import org.apache.kafka.clients.consumer._
 import org.apache.kafka.clients.producer.{ProducerConfig, ProducerRecord}
-import org.apache.kafka.common.{KafkaException, MetricName, TopicPartition}


It would be better to do this in a separate cleanup PR.

Yeah. Next time. I can't help but clean up code as I see it.

cmccabe · 2024-05-29T21:41:52Z

core/src/test/resources/log4j.properties

@@ -21,6 +21,5 @@ log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n
 log4j.logger.kafka=WARN
 log4j.logger.org.apache.kafka=WARN

-


It would be better to do this in a separate cleanup PR.

Yes. Next time.

cmccabe · 2024-05-29T21:43:19Z

raft/src/main/java/org/apache/kafka/raft/FollowerState.java

-            ", highWatermark=" + highWatermark +
-            ", fetchingSnapshot=" + fetchingSnapshot +
-            ')';
+        return String.format(


Hmm, I'm not sure I see a lot of benefit to this change.

String.format are easier to update and read.

Personally I find String.format actually harder to read in the "simple" cases where you're just printing x=myX, y=myY. I think where it really shines is where you want an exotic format (like x=%02x, y=%08d, etc.) or where the textual part is complex. However, I don't feel that strongly about this so you can leave it.

cmccabe · 2024-05-29T21:47:31Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

@@ -1593,6 +1644,13 @@ private Optional<Boolean> maybeHandleCommonResponse(
        int epoch,
        long currentTimeMs
    ) {
+        Optional<Node> leader = Optional.empty();


This would be more clearly expressed as something like

Optional<Node> leader = leaderId.isPresent() ? partitionState.lastVoterSet().voterNode(leaderId.getAsInt(), channel.listenerName()) : Optional.empty();

I find that suggestion much harder to read because the important characters ? and : get lost with all the other characters. I updated the code anyways.

cmccabe · 2024-05-29T21:48:17Z

raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java

        int epoch,
        long currentTimeMs
    ) {
+        OptionalInt leaderId = OptionalInt.empty();


Again would be more clearly expressed with ?: rather than mutating the OptionalInt reference

cmccabe · 2024-05-29T21:57:25Z

raft/src/main/java/org/apache/kafka/raft/RequestManager.java

 public class RequestManager {
-    private final Map<Integer, ConnectionState> connections = new HashMap<>();
-    private final List<Integer> voters = new ArrayList<>();
+    private final Map<String, ConnectionState> connections = new HashMap<>();


I don't understand why you are changing this to treat node ID as a string. An Int seems better.

Mainly to match NetworkClient. NetworkClient uses Node::idString as the key instead of Node::id.

That seems like a mistake in NetworkClient, and I don't see why we'd want to emulate it here. But it's not a very big mistake in the grand scheme of things

cmccabe · 2024-05-29T22:00:16Z

raft/src/main/java/org/apache/kafka/raft/internals/VoterSet.java

-    public Optional<InetSocketAddress> voterAddress(int voter, String listener) {
-        return Optional.ofNullable(voters.get(voter))
-            .flatMap(voterNode -> voterNode.address(listener));
+    public Set<Node> voterNodes(Stream<Integer> voterIds, ListenerName listenerName) {


This seems like the wrong behavior to me for the case where one of the voters doesn't have a specific listener, but the other ones do. Wouldn't we want to continue to use the other, working voters in that case?

Maybe. I had that as the original implementation but changed it to throw an IllegalArgumentException.

If the user configured the cluster so that the default/preferred listener of one node/voter is not supported by all of the voters, it is clearly misconfigured. Should KRaft to continue silently knowing that the voter can never become leader? Or should Kafka catch the misconfiguration early and fail?

I opted, and in KRaft we generally opt, towards fail fast.

I guess. We have to be a bit careful though, because the quorum topolgy can change over time. If we encounter an old state (that isn't the latest) and that causes us to kill the kcontroller before we can see the latest state, that could be quite bad.

cmccabe · 2024-05-29T22:07:00Z

Thanks for the PR, @jsancio . A few meta-comments:

I'm not sure I see the benefit to changing the toString functions to use String.format. It seems more brittle than the simple string concatenation approach. String.format can be helpful when you want to print something in a specific way, like String.format("%02d", myInt).... but that isn't the case here.
Changing the SharedServer constructor is a huge pain and generates a lot of churn. Since the thing you're passing comes from the static configuration anyway, let's not do that. Just read the static configuration in SharedServer (which we do for many other things)
I think we should try to avoid doing too much validation in KafkaConfig. For example, things like hostnames should be resolved when we actually need them. It would be silly for one unresolvable hostname to make configuration validation fail, and hence fail the whole kcontroller startup process.
I don't think we want to change all of the tests to use controller.quorum.bootstrap.servers. We still need to support the old configuration. Let's make it so that tests using IBP_3_8_IV0 or newer use the new thing, and tests using an older MV use the old configuration. That way we will have good coverage.

showuon · 2024-05-30T11:51:36Z

raft/src/main/java/org/apache/kafka/raft/RequestManager.java

+            return 0;
+        }
+
+        return  state.remainingBackoffMs(timeMs);


nit: additional space after return.

Thanks. Looks like I already removed this in another commit.

showuon · 2024-05-30T12:21:45Z

raft/src/main/java/org/apache/kafka/raft/RequestManager.java

+            } else if (connection.isBackoffComplete(currentTimeMs)) {
+                // Mark the node as ready after completed backoff
+                iterator.remove();
+            } else if (connection.hasInflightRequest(currentTimeMs)) {


It looks like we'll also check hasRequestTimeout in hasInflightRequest. We can consider to do a small optimization here in the future.

I agree. I think we can re-implement the RequestManager so that all operations consistently update the state as part of checking the state. I think it would be much easier to read and understand. I opted against it in this PR to minimize the diff and the possibility of introducing a breaking change.

…c-raft-channel

jsancio · 2024-06-03T16:12:58Z

I don't think we want to change all of the tests to use controller.quorum.bootstrap.servers. We still need to support the old configuration. Let's make it so that tests using IBP_3_8_IV0 or newer use the new thing, and tests using an older MV use the old configuration. That way we will have good coverage.

controller.quorum.bootstrap.servers is not gated by metadata version, kraft version or IBP. This property only affects the behavior of the local node and it doesn't affect the network protocol (RPCs) or data version stored on disk. In other words using this configuration should be safe irrespective of the software version of the remote nodes.

I decided to change all of the quorum test harness test to use controller.quorum.bootstrap.servers because this feature is a super set of the controller.quorum.voters feature since the id are negatives. In other words, if tests pass with controller.quorum.bootstrap.servers they should pass with controller.quorum.bootstrap.servers.

cmccabe

LGTM

cmccabe · 2024-06-03T18:07:47Z

controller.quorum.bootstrap.servers is not gated by metadata version, kraft version or IBP. This property only affects the behavior of the local node and it doesn't affect the network protocol (RPCs) or data version stored on disk. In other words using this configuration should be safe irrespective of the software version of the remote nodes.

I agree with this, but that was not my point. My point was that we should have some tests exercising both code paths.

I do think we need to fix this later, but I don't want to hold up the PR on it.

Allow KRaft replicas to send requests to any node (Node) not just the nodes configured in the controller.quorum.voters property. This flexibility is needed so KRaft can implement the controller.quorum.voters configuration, send request to the dynamically changing set of voters and send request to the leader endpoint (Node) discovered through the KRaft RPCs (specially BeginQuorumEpoch request and Fetch response). This was achieved by changing the RequestManager API to accept Node instead of just the replica ID. Internally, the request manager tracks connection state using the Node.idString method to match the connection management used by NetworkClient. The API for RequestManager is also changed so that the ConnectState class is not exposed in the API. This allows the request manager to reclaim heap memory for any connection that is ready. The NetworkChannel was updated to receive the endpoint information (Node) through the outbound raft request (RaftRequent.Outbound). This makes the network channel more flexible as it doesn't need to be configured with the list of all possible endpoints. RaftRequest.Outbound and RaftResponse.Inbound were updated to include the remote node instead of just the remote id. The follower state tracked by KRaft replicas was updated to include both the leader id and the leader's endpoint (Node). In this comment the node value is computed from the set of voters. In future commit this will be updated so that it is sent through KRaft RPCs. For example BeginQuorumEpoch request and Fetch response. Support for configuring controller.quorum.bootstrap.servers was added. This includes changes to KafkaConfig, QuorumConfig, etc. All of the tests using QuorumTestHarness were changed to use the controller.quorum.bootstrap.servers instead of the controller.quorum.voters for the broker configuration. Finally, the node id for the bootstrap server will be decreasing negative numbers starting with -2. Reviewers: Jason Gustafson <jason@confluent.io>, Luke Chen <showuon@gmail.com>, Colin P. McCabe <cmccabe@apache.org>

KAFKA-16252; Dynamic KRaft network manager and channel

a87ae00

jsancio added kraft KIP-853 labels May 17, 2024

jsancio added 6 commits May 17, 2024 16:15

KAFKA-16525; Add more tests for request manager

1b349c4

KAFKA-16525; Add bootstrap server configuration

5f34a22

KAFKA-16525; Use bootstrap servers in quorum test harness

9da75f6

KAFKA-16525; Make VoterSet.voterNodes stricter

1d8c8b8

KAFKA-16525; Fix Scala 2.12 compile error

5a3abe4

Merge remote-tracking branch 'upstream/trunk' into kafka-16525-dynami…

138236e

…c-raft-channel

jsancio marked this pull request as ready for review May 20, 2024 17:07

jsancio added 4 commits May 21, 2024 14:37

KAFKA-16525; Enable raft trace for integration test

e9e12db

KAFKA-16525; Let's try this configuration

33ec50a

KAFKA-16525; Enable raft trace for the core tests

0654c11

KAFKA-16525; More reliable to log the fetch responses on errors

ad6895c

hachikuji reviewed May 22, 2024

View reviewed changes

jsancio added 8 commits May 23, 2024 12:22

KAFKA-16525; Include request in response to help debugging

ccbb8f8

KAFKA-16525; Add bootstrap server retry unittest

bb4933b

KAFKA-16525; Fix retry logic in request manager

f7476e6

KAFKA-16525; Fix retry error with Fetch

ba7a2c8

Merge remote-tracking branch 'upstream/trunk' into kafka-16525-dynami…

97ee073

…c-raft-channel

KAFKA-16525; Add more request manager tests

5feb381

KAFKA-16525; Remove debugging code

a923b73

Merge remote-tracking branch 'upstream/trunk' into kafka-16525-dynami…

49330ea

…c-raft-channel

cmccabe reviewed May 29, 2024

View reviewed changes

showuon reviewed May 30, 2024

View reviewed changes

ahuang98 mentioned this pull request May 31, 2024

KAFKA-16252: Dynamic KRaft network manager and channel #16160

Closed

3 tasks

jsancio changed the title ~~KAFKA-16252; Dynamic KRaft network manager and channel~~ KAFKA-16525; Dynamic KRaft network manager and channel Jun 2, 2024

Merge remote-tracking branch 'upstream/trunk' into kafka-16525-dynami…

69aea43

…c-raft-channel

KAFKA-16525; Use Java's conditional expression

d2371b2

cmccabe approved these changes Jun 3, 2024

View reviewed changes

cmccabe merged commit 459da47 into apache:trunk Jun 3, 2024
1 check failed

		@@ -21,6 +21,5 @@ log4j.appender.stdout.layout.ConversionPattern=[%d] %p %m (%c:%L)%n
		log4j.logger.kafka=WARN
		log4j.logger.org.apache.kafka=WARN

KAFKA-16525; Dynamic KRaft network manager and channel #15986

KAFKA-16525; Dynamic KRaft network manager and channel #15986

Conversation

jsancio commented May 17, 2024 • edited

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio May 22, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio commented May 27, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe May 29, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe May 29, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe May 29, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cmccabe commented May 29, 2024 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jsancio commented Jun 3, 2024

cmccabe left a comment

Choose a reason for hiding this comment

cmccabe commented Jun 3, 2024

jsancio commented May 17, 2024 •

edited

jsancio May 22, 2024 •

edited

cmccabe May 29, 2024 •

edited

cmccabe May 29, 2024 •

edited

cmccabe May 29, 2024 •

edited

cmccabe commented May 29, 2024 •

edited