You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jsancio (via GitHub)" <gi...@apache.org> on 2024/05/17 16:53:05 UTC
[PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
jsancio opened a new pull request, #15986:
URL: https://github.com/apache/kafka/pull/15986
DRAFT
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619506439
##########
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],
Review Comment:
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.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #15986:
URL: https://github.com/apache/kafka/pull/15986#issuecomment-2134135719
@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.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619518236
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,296 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
+import java.util.Iterator;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
+/**
+ * The request manager keeps tracks of the connection with remote replicas.
+ *
+ * When sending a request update this type by calling {@code onRequestSent(Node, long, long)}. When
+ * the RPC returns a response, update this manager with {@code onResponseResult(Node, long, boolean, long)}.
+ *
+ * Connections start in the ready state ({@code isReady(Node, long)} returns true).
+ *
+ * When a request times out or completes successfully the collection will transition back to the
+ * ready state.
+ *
+ * When a request completes with an error it still transition to the backoff state until
+ * {@code retryBackoffMs}.
+ */
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<>();
Review Comment:
I don't understand why you are changing this to treat node ID as a string. An Int seems better.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619507872
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -156,22 +156,24 @@ public boolean canGrantVote(ReplicaKey candidateKey, boolean isLogUpToDate) {
log.debug(
"Rejecting vote request from candidate ({}) since we already have a leader {} in epoch {}",
candidateKey,
- leaderId(),
+ leader,
epoch
);
return false;
}
@Override
public String toString() {
- return "FollowerState(" +
- "fetchTimeoutMs=" + fetchTimeoutMs +
- ", epoch=" + epoch +
- ", leaderId=" + leaderId +
- ", voters=" + voters +
- ", highWatermark=" + highWatermark +
- ", fetchingSnapshot=" + fetchingSnapshot +
- ')';
+ return String.format(
Review Comment:
Hmm, I'm not sure I see a lot of benefit to this change.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1614958193
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,196 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
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<>();
+ private final ArrayList<Node> bootstrapServers;
private final int retryBackoffMs;
private final int requestTimeoutMs;
private final Random random;
- public RequestManager(Set<Integer> voterIds,
- int retryBackoffMs,
- int requestTimeoutMs,
- Random random) {
-
+ public RequestManager(
+ Collection<Node> bootstrapServers,
+ int retryBackoffMs,
+ int requestTimeoutMs,
+ Random random
+ ) {
+ this.bootstrapServers = new ArrayList<>(bootstrapServers);
this.retryBackoffMs = retryBackoffMs;
this.requestTimeoutMs = requestTimeoutMs;
- this.voters.addAll(voterIds);
this.random = random;
-
- for (Integer voterId: voterIds) {
- ConnectionState connection = new ConnectionState(voterId);
- connections.put(voterId, connection);
- }
- }
-
- public ConnectionState getOrCreate(int id) {
- return connections.computeIfAbsent(id, key -> new ConnectionState(id));
}
- public OptionalInt findReadyVoter(long currentTimeMs) {
- int startIndex = random.nextInt(voters.size());
- OptionalInt res = OptionalInt.empty();
- for (int i = 0; i < voters.size(); i++) {
- int index = (startIndex + i) % voters.size();
- Integer voterId = voters.get(index);
- ConnectionState connection = connections.get(voterId);
- boolean isReady = connection.isReady(currentTimeMs);
+ public Optional<Node> findReadyBootstrapServer(long currentTimeMs) {
+ int startIndex = random.nextInt(bootstrapServers.size());
+ Optional<Node> res = Optional.empty();
+ for (int i = 0; i < bootstrapServers.size(); i++) {
+ int index = (startIndex + i) % bootstrapServers.size();
+ Node node = bootstrapServers.get(index);
- if (isReady) {
- res = OptionalInt.of(voterId);
- } else if (connection.inFlightCorrelationId.isPresent()) {
- res = OptionalInt.empty();
+ if (isReady(node, currentTimeMs)) {
+ res = Optional.of(node);
+ } else if (hasInflightRequest(node, currentTimeMs)) {
+ res = Optional.empty();
break;
}
}
+
return res;
}
- public long backoffBeforeAvailableVoter(long currentTimeMs) {
- long minBackoffMs = Long.MAX_VALUE;
- for (Integer voterId : voters) {
- ConnectionState connection = connections.get(voterId);
- if (connection.isReady(currentTimeMs)) {
+ public long backoffBeforeAvailableBootstrapServer(long currentTimeMs) {
+ long minBackoffMs = Math.max(retryBackoffMs, requestTimeoutMs);
+ for (Node node : bootstrapServers) {
+ if (isReady(node, currentTimeMs)) {
return 0L;
- } else if (connection.isBackingOff(currentTimeMs)) {
- minBackoffMs = Math.min(minBackoffMs, connection.remainingBackoffMs(currentTimeMs));
+ } else if (isBackingOff(node, currentTimeMs)) {
+ minBackoffMs = Math.min(minBackoffMs, remainingBackoffMs(node, currentTimeMs));
} else {
- minBackoffMs = Math.min(minBackoffMs, connection.remainingRequestTimeMs(currentTimeMs));
+ minBackoffMs = Math.min(minBackoffMs, remainingRequestTimeMs(node, currentTimeMs));
}
}
+
return minBackoffMs;
}
+ public boolean hasRequestTimedOut(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.hasRequestTimedOut(timeMs);
+ }
+
+ public boolean isReady(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return true;
+ }
+
+ boolean ready = state.isReady(timeMs);
+ if (ready) {
+ reset(node);
+ }
+
+ return ready;
+ }
+
+ public boolean isBackingOff(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.isBackingOff(timeMs);
+ }
+
+ public long remainingRequestTimeMs(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return 0;
+ }
+
+ return state.remainingRequestTimeMs(timeMs);
+ }
+
+ public long remainingBackoffMs(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return 0;
+ }
+
+ return state.remainingBackoffMs(timeMs);
+ }
+
+ public boolean isResponseExpected(Node node, long correlationId) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.isResponseExpected(correlationId);
+ }
+
+ public void onResponseReceived(Node node, long correlationId) {
Review Comment:
Sounds good. Went with `onResponseResult(Node, long, boolean, long)`.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624866274
##########
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
-
Review Comment:
ok
##########
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}
Review Comment:
ok
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624701021
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -156,22 +156,24 @@ public boolean canGrantVote(ReplicaKey candidateKey, boolean isLogUpToDate) {
log.debug(
"Rejecting vote request from candidate ({}) since we already have a leader {} in epoch {}",
candidateKey,
- leaderId(),
+ leader,
epoch
);
return false;
}
@Override
public String toString() {
- return "FollowerState(" +
- "fetchTimeoutMs=" + fetchTimeoutMs +
- ", epoch=" + epoch +
- ", leaderId=" + leaderId +
- ", voters=" + voters +
- ", highWatermark=" + highWatermark +
- ", fetchingSnapshot=" + fetchingSnapshot +
- ')';
+ return String.format(
Review Comment:
`String.format` are easier to update and read.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619506788
##########
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}
Review Comment:
It would be better to do this in a separate cleanup PR.
##########
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
-
Review Comment:
It would be better to do this in a separate cleanup PR.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619514531
##########
raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java:
##########
@@ -199,6 +206,34 @@ private static Map<Integer, InetSocketAddress> parseVoterConnections(
return voterMap;
}
+ public static InetSocketAddress parseBootstrapServer(String bootstrapServer) {
+ String host = Utils.getHost(bootstrapServer);
+ if (host == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host name from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ Integer port = Utils.getPort(bootstrapServer);
+ if (port == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host port from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ return InetSocketAddress.createUnresolved(host, port);
Review Comment:
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.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619510911
##########
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();
Review Comment:
This would be more clearly expressed as something like
```
Optional<Node> leader = leaderId.isPresent() ?
partitionState.lastVoterSet().voterNode(leaderId.getAsInt(), channel.listenerName()) :
Optional.empty();
```
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624709723
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,296 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
+import java.util.Iterator;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
+/**
+ * The request manager keeps tracks of the connection with remote replicas.
+ *
+ * When sending a request update this type by calling {@code onRequestSent(Node, long, long)}. When
+ * the RPC returns a response, update this manager with {@code onResponseResult(Node, long, boolean, long)}.
+ *
+ * Connections start in the ready state ({@code isReady(Node, long)} returns true).
+ *
+ * When a request times out or completes successfully the collection will transition back to the
+ * ready state.
+ *
+ * When a request completes with an error it still transition to the backoff state until
+ * {@code retryBackoffMs}.
+ */
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<>();
Review Comment:
Mainly to match `NetworkClient`. `NetworkClient` uses `Node::idString` as the key instead of `Node::id`.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "hachikuji (via GitHub)" <gi...@apache.org>.
hachikuji commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1610398594
##########
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
Review Comment:
nit: we seem to have this same little snippet in a few places. Is there somewhere we could add a helper?
##########
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
Review Comment:
Reminder about TODO
##########
raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java:
##########
@@ -231,4 +266,26 @@ public String toString() {
return "non-empty list";
}
}
+
+ public static class ControllerQuorumBootstrapServersValidator implements ConfigDef.Validator {
+ @Override
+ public void ensureValid(String name, Object value) {
+ if (value == null) {
+ throw new ConfigException(name, null);
+ }
+
+ @SuppressWarnings("unchecked")
+ List<String> entries = (List<String>) value;
+
+ // Attempt to parse the connect strings
+ for (String entry : entries) {
+ QuorumConfig.parseBootstrapServer(entry);
Review Comment:
nit: drop unnecessary `QuorumConfig` prefix?
##########
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala:
##########
@@ -1366,6 +1368,26 @@ class KafkaConfigTest {
assertEquals(expectedVoters, addresses)
}
+ @Test
+ def testParseQuorumBootstrapServers(): Unit = {
Review Comment:
Perhaps we should have some invalid test cases as well?
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,196 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
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<>();
+ private final ArrayList<Node> bootstrapServers;
private final int retryBackoffMs;
private final int requestTimeoutMs;
private final Random random;
- public RequestManager(Set<Integer> voterIds,
- int retryBackoffMs,
- int requestTimeoutMs,
- Random random) {
-
+ public RequestManager(
+ Collection<Node> bootstrapServers,
+ int retryBackoffMs,
+ int requestTimeoutMs,
+ Random random
+ ) {
+ this.bootstrapServers = new ArrayList<>(bootstrapServers);
this.retryBackoffMs = retryBackoffMs;
this.requestTimeoutMs = requestTimeoutMs;
- this.voters.addAll(voterIds);
this.random = random;
-
- for (Integer voterId: voterIds) {
- ConnectionState connection = new ConnectionState(voterId);
- connections.put(voterId, connection);
- }
- }
-
- public ConnectionState getOrCreate(int id) {
- return connections.computeIfAbsent(id, key -> new ConnectionState(id));
}
- public OptionalInt findReadyVoter(long currentTimeMs) {
- int startIndex = random.nextInt(voters.size());
- OptionalInt res = OptionalInt.empty();
- for (int i = 0; i < voters.size(); i++) {
- int index = (startIndex + i) % voters.size();
- Integer voterId = voters.get(index);
- ConnectionState connection = connections.get(voterId);
- boolean isReady = connection.isReady(currentTimeMs);
+ public Optional<Node> findReadyBootstrapServer(long currentTimeMs) {
+ int startIndex = random.nextInt(bootstrapServers.size());
+ Optional<Node> res = Optional.empty();
+ for (int i = 0; i < bootstrapServers.size(); i++) {
+ int index = (startIndex + i) % bootstrapServers.size();
+ Node node = bootstrapServers.get(index);
- if (isReady) {
- res = OptionalInt.of(voterId);
- } else if (connection.inFlightCorrelationId.isPresent()) {
- res = OptionalInt.empty();
+ if (isReady(node, currentTimeMs)) {
+ res = Optional.of(node);
+ } else if (hasInflightRequest(node, currentTimeMs)) {
+ res = Optional.empty();
break;
}
}
+
return res;
}
- public long backoffBeforeAvailableVoter(long currentTimeMs) {
- long minBackoffMs = Long.MAX_VALUE;
- for (Integer voterId : voters) {
- ConnectionState connection = connections.get(voterId);
- if (connection.isReady(currentTimeMs)) {
+ public long backoffBeforeAvailableBootstrapServer(long currentTimeMs) {
+ long minBackoffMs = Math.max(retryBackoffMs, requestTimeoutMs);
+ for (Node node : bootstrapServers) {
+ if (isReady(node, currentTimeMs)) {
return 0L;
- } else if (connection.isBackingOff(currentTimeMs)) {
- minBackoffMs = Math.min(minBackoffMs, connection.remainingBackoffMs(currentTimeMs));
+ } else if (isBackingOff(node, currentTimeMs)) {
+ minBackoffMs = Math.min(minBackoffMs, remainingBackoffMs(node, currentTimeMs));
} else {
- minBackoffMs = Math.min(minBackoffMs, connection.remainingRequestTimeMs(currentTimeMs));
+ minBackoffMs = Math.min(minBackoffMs, remainingRequestTimeMs(node, currentTimeMs));
}
}
+
return minBackoffMs;
}
+ public boolean hasRequestTimedOut(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.hasRequestTimedOut(timeMs);
+ }
+
+ public boolean isReady(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return true;
+ }
+
+ boolean ready = state.isReady(timeMs);
+ if (ready) {
+ reset(node);
+ }
+
+ return ready;
+ }
+
+ public boolean isBackingOff(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.isBackingOff(timeMs);
+ }
+
+ public long remainingRequestTimeMs(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return 0;
+ }
+
+ return state.remainingRequestTimeMs(timeMs);
+ }
+
+ public long remainingBackoffMs(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return 0;
+ }
+
+ return state.remainingBackoffMs(timeMs);
+ }
+
+ public boolean isResponseExpected(Node node, long correlationId) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.isResponseExpected(correlationId);
+ }
+
+ public void onResponseReceived(Node node, long correlationId) {
Review Comment:
I wonder if we could consolidate `onResponseReceived` and `onResponseError`. Perhaps something like `notifyResponseResult` or something like that.
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -393,17 +422,35 @@ public void initialize(
logger.info("Reading KRaft snapshot and log as part of the initialization");
partitionState.updateState();
- VoterSet lastVoterSet = partitionState.lastVoterSet();
- requestManager = new RequestManager(
- lastVoterSet.voterIds(),
- quorumConfig.retryBackoffMs(),
- quorumConfig.requestTimeoutMs(),
- random
- );
+ if (requestManager == null) {
+ // The request manager wasn't created using the bootstrap servers
+ // create it using the voters static configuration
+ List<Node> bootstrapNodes = voterAddresses
+ .entrySet()
+ .stream()
+ .map(entry ->
+ new Node(
+ entry.getKey(),
+ entry.getValue().getHostString(),
+ entry.getValue().getPort()
+ )
+ )
+ .collect(Collectors.toList());
+
+ logger.info("Starting request manager with bootstrap servers: {}", bootstrapNodes);
Review Comment:
Perhaps we change this message to something like "Staring request manager with static voter set: {}"
##########
raft/src/main/java/org/apache/kafka/raft/internals/VoterSet.java:
##########
@@ -53,15 +56,41 @@ final public class VoterSet {
}
/**
- * Returns the socket address for a given voter at a given listener.
+ * Returns the node information for all the given voter ids and listener.
*
- * @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
Review Comment:
nit: ids of the voters
##########
raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java:
##########
@@ -199,6 +206,34 @@ private static Map<Integer, InetSocketAddress> parseVoterConnections(
return voterMap;
}
+ public static InetSocketAddress parseBootstrapServer(String bootstrapServer) {
+ String host = Utils.getHost(bootstrapServer);
+ if (host == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host name from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ Integer port = Utils.getPort(bootstrapServer);
+ if (port == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host port from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ return InetSocketAddress.createUnresolved(host, port);
Review Comment:
Is this going to give us a nice exception message if the host is invalid?
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -29,40 +30,42 @@
public class FollowerState implements EpochState {
private final int fetchTimeoutMs;
private final int epoch;
- private final int leaderId;
+ private final Node leader;
private final Set<Integer> voters;
// Used for tracking the expiration of both the Fetch and FetchSnapshot requests
private final Timer fetchTimer;
private Optional<LogOffsetMetadata> highWatermark;
/* Used to track the currently fetching snapshot. When fetching snapshot regular
* Fetch request are paused
*/
- private Optional<RawSnapshotWriter> fetchingSnapshot;
+ private Optional<RawSnapshotWriter> fetchingSnapshot = Optional.empty();
+
+ // TODO: remove this when done debuggin CI failures
Review Comment:
TODO reminder
##########
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
Review Comment:
TODO reminder
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1611939749
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -393,17 +422,35 @@ public void initialize(
logger.info("Reading KRaft snapshot and log as part of the initialization");
partitionState.updateState();
- VoterSet lastVoterSet = partitionState.lastVoterSet();
- requestManager = new RequestManager(
- lastVoterSet.voterIds(),
- quorumConfig.retryBackoffMs(),
- quorumConfig.requestTimeoutMs(),
- random
- );
+ if (requestManager == null) {
+ // The request manager wasn't created using the bootstrap servers
+ // create it using the voters static configuration
+ List<Node> bootstrapNodes = voterAddresses
+ .entrySet()
+ .stream()
+ .map(entry ->
+ new Node(
+ entry.getKey(),
+ entry.getValue().getHostString(),
+ entry.getValue().getPort()
+ )
+ )
+ .collect(Collectors.toList());
+
+ logger.info("Starting request manager with bootstrap servers: {}", bootstrapNodes);
Review Comment:
Done.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619507872
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -156,22 +156,24 @@ public boolean canGrantVote(ReplicaKey candidateKey, boolean isLogUpToDate) {
log.debug(
"Rejecting vote request from candidate ({}) since we already have a leader {} in epoch {}",
candidateKey,
- leaderId(),
+ leader,
epoch
);
return false;
}
@Override
public String toString() {
- return "FollowerState(" +
- "fetchTimeoutMs=" + fetchTimeoutMs +
- ", epoch=" + epoch +
- ", leaderId=" + leaderId +
- ", voters=" + voters +
- ", highWatermark=" + highWatermark +
- ", fetchingSnapshot=" + fetchingSnapshot +
- ')';
+ return String.format(
Review Comment:
I'm not sure I see a lot of benefit to this change.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619511561
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1635,24 +1693,29 @@ private Optional<Boolean> maybeHandleCommonResponse(
}
private void maybeTransition(
- OptionalInt leaderId,
+ Optional<Node> leader,
int epoch,
long currentTimeMs
) {
+ OptionalInt leaderId = OptionalInt.empty();
Review Comment:
Again would be more clearly expressed with `?:` rather than mutating the OptionalInt reference
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624707145
##########
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();
Review Comment:
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.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1611946424
##########
raft/src/main/java/org/apache/kafka/raft/internals/VoterSet.java:
##########
@@ -53,15 +56,41 @@ final public class VoterSet {
}
/**
- * Returns the socket address for a given voter at a given listener.
+ * Returns the node information for all the given voter ids and listener.
*
- * @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
Review Comment:
Fixed.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624697438
##########
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],
Review Comment:
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
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1615318177
##########
raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java:
##########
@@ -199,6 +206,34 @@ private static Map<Integer, InetSocketAddress> parseVoterConnections(
return voterMap;
}
+ public static InetSocketAddress parseBootstrapServer(String bootstrapServer) {
+ String host = Utils.getHost(bootstrapServer);
+ if (host == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host name from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ Integer port = Utils.getPort(bootstrapServer);
+ if (port == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host port from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ return InetSocketAddress.createUnresolved(host, port);
Review Comment:
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.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624745219
##########
raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java:
##########
@@ -1635,24 +1693,29 @@ private Optional<Boolean> maybeHandleCommonResponse(
}
private void maybeTransition(
- OptionalInt leaderId,
+ Optional<Node> leader,
int epoch,
long currentTimeMs
) {
+ OptionalInt leaderId = OptionalInt.empty();
Review Comment:
Fixed.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1620608088
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,296 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
+import java.util.Iterator;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
+/**
+ * The request manager keeps tracks of the connection with remote replicas.
+ *
+ * When sending a request update this type by calling {@code onRequestSent(Node, long, long)}. When
+ * the RPC returns a response, update this manager with {@code onResponseResult(Node, long, boolean, long)}.
+ *
+ * Connections start in the ready state ({@code isReady(Node, long)} returns true).
+ *
+ * When a request times out or completes successfully the collection will transition back to the
+ * ready state.
+ *
+ * When a request completes with an error it still transition to the backoff state until
+ * {@code retryBackoffMs}.
+ */
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<>();
+ private final ArrayList<Node> bootstrapServers;
private final int retryBackoffMs;
private final int requestTimeoutMs;
private final Random random;
- public RequestManager(Set<Integer> voterIds,
- int retryBackoffMs,
- int requestTimeoutMs,
- Random random) {
-
+ public RequestManager(
+ Collection<Node> bootstrapServers,
+ int retryBackoffMs,
+ int requestTimeoutMs,
+ Random random
+ ) {
+ this.bootstrapServers = new ArrayList<>(bootstrapServers);
this.retryBackoffMs = retryBackoffMs;
this.requestTimeoutMs = requestTimeoutMs;
- this.voters.addAll(voterIds);
this.random = random;
+ }
- for (Integer voterId: voterIds) {
- ConnectionState connection = new ConnectionState(voterId);
- connections.put(voterId, connection);
+ /**
+ * Returns true if there any connection with pending requests.
+ *
+ * This is useful for satisfying the invariant that there is only one pending Fetch request.
+ * If there are more than one pending fetch request, it is possible for the follower to write
+ * the same offset twice.
+ *
+ * @param currentTimeMs the current time
+ * @return true if the request manager is tracking at least one request
+ */
+ public boolean hasAnyInflightRequest(long currentTimeMs) {
+ boolean result = false;
+
+ Iterator<ConnectionState> iterator = connections.values().iterator();
+ while (iterator.hasNext()) {
+ ConnectionState connection = iterator.next();
+ if (connection.hasRequestTimedOut(currentTimeMs)) {
+ // Mark the node as ready after request timeout
+ iterator.remove();
+ } else if (connection.isBackoffComplete(currentTimeMs)) {
+ // Mark the node as ready after completed backoff
+ iterator.remove();
+ } else if (connection.hasInflightRequest(currentTimeMs)) {
Review Comment:
It looks like we'll also check hasRequestTimeout in hasInflightRequest. We can consider to do a small optimization here in the future.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1611947146
##########
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
Review Comment:
Fixed.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624726847
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,296 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
+import java.util.Iterator;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
+/**
+ * The request manager keeps tracks of the connection with remote replicas.
+ *
+ * When sending a request update this type by calling {@code onRequestSent(Node, long, long)}. When
+ * the RPC returns a response, update this manager with {@code onResponseResult(Node, long, boolean, long)}.
+ *
+ * Connections start in the ready state ({@code isReady(Node, long)} returns true).
+ *
+ * When a request times out or completes successfully the collection will transition back to the
+ * ready state.
+ *
+ * When a request completes with an error it still transition to the backoff state until
+ * {@code retryBackoffMs}.
+ */
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<>();
+ private final ArrayList<Node> bootstrapServers;
private final int retryBackoffMs;
private final int requestTimeoutMs;
private final Random random;
- public RequestManager(Set<Integer> voterIds,
- int retryBackoffMs,
- int requestTimeoutMs,
- Random random) {
-
+ public RequestManager(
+ Collection<Node> bootstrapServers,
+ int retryBackoffMs,
+ int requestTimeoutMs,
+ Random random
+ ) {
+ this.bootstrapServers = new ArrayList<>(bootstrapServers);
this.retryBackoffMs = retryBackoffMs;
this.requestTimeoutMs = requestTimeoutMs;
- this.voters.addAll(voterIds);
this.random = random;
+ }
- for (Integer voterId: voterIds) {
- ConnectionState connection = new ConnectionState(voterId);
- connections.put(voterId, connection);
+ /**
+ * Returns true if there any connection with pending requests.
+ *
+ * This is useful for satisfying the invariant that there is only one pending Fetch request.
+ * If there are more than one pending fetch request, it is possible for the follower to write
+ * the same offset twice.
+ *
+ * @param currentTimeMs the current time
+ * @return true if the request manager is tracking at least one request
+ */
+ public boolean hasAnyInflightRequest(long currentTimeMs) {
+ boolean result = false;
+
+ Iterator<ConnectionState> iterator = connections.values().iterator();
+ while (iterator.hasNext()) {
+ ConnectionState connection = iterator.next();
+ if (connection.hasRequestTimedOut(currentTimeMs)) {
+ // Mark the node as ready after request timeout
+ iterator.remove();
+ } else if (connection.isBackoffComplete(currentTimeMs)) {
+ // Mark the node as ready after completed backoff
+ iterator.remove();
+ } else if (connection.hasInflightRequest(currentTimeMs)) {
+ // If there is at least one inflight request, it is enough
+ // to stop checking the rest of the connections
+ result = true;
+ break;
+ }
}
- }
- public ConnectionState getOrCreate(int id) {
- return connections.computeIfAbsent(id, key -> new ConnectionState(id));
+ return result;
}
- public OptionalInt findReadyVoter(long currentTimeMs) {
- int startIndex = random.nextInt(voters.size());
- OptionalInt res = OptionalInt.empty();
- for (int i = 0; i < voters.size(); i++) {
- int index = (startIndex + i) % voters.size();
- Integer voterId = voters.get(index);
- ConnectionState connection = connections.get(voterId);
- boolean isReady = connection.isReady(currentTimeMs);
+ /**
+ * Returns a random bootstrap node that is ready to receive a request.
+ *
+ * This method doesn't return a node if there is at least one request pending. In general this
+ * method is used to send Fetch requests. Fetch requests have the invariant that there can
+ * only be one pending Fetch request for the LEO.
+ *
+ * @param currentTimeMs the current time
+ * @return a random ready bootstrap node
+ */
+ public Optional<Node> findReadyBootstrapServer(long currentTimeMs) {
+ // Check that there are no infilght requests accross any of the known nodes not just
+ // the bootstrap servers
+ if (hasAnyInflightRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
- if (isReady) {
- res = OptionalInt.of(voterId);
- } else if (connection.inFlightCorrelationId.isPresent()) {
- res = OptionalInt.empty();
+ int startIndex = random.nextInt(bootstrapServers.size());
+ Optional<Node> result = Optional.empty();
+ for (int i = 0; i < bootstrapServers.size(); i++) {
+ int index = (startIndex + i) % bootstrapServers.size();
+ Node node = bootstrapServers.get(index);
+
+ if (isReady(node, currentTimeMs)) {
+ result = Optional.of(node);
break;
}
}
- return res;
+
+ return result;
}
- public long backoffBeforeAvailableVoter(long currentTimeMs) {
- long minBackoffMs = Long.MAX_VALUE;
- for (Integer voterId : voters) {
- ConnectionState connection = connections.get(voterId);
- if (connection.isReady(currentTimeMs)) {
- return 0L;
+ /**
+ * Computes the amount of time needed to wait before a bootstrap server is ready for a Fetch
+ * request.
+ *
+ * If there is a connection with a pending request it returns the amount of time to wait until
+ * the request times out.
+ *
+ * Returns zero, if there are no pending request and at least one of the boorstrap servers is
+ * ready.
+ *
+ * If all of the bootstrap servers are backing off and there are no pending requests, return
+ * the minimum amount of time until a bootstrap server becomes ready.
+ *
+ * @param currentTimeMs the current time
+ * @return the amount of time to wait until bootstrap server can accept a Fetch request
+ */
+ public long backoffBeforeAvailableBootstrapServer(long currentTimeMs) {
+ long minBackoffMs = retryBackoffMs;
+
+ Iterator<ConnectionState> iterator = connections.values().iterator();
+ while (iterator.hasNext()) {
+ ConnectionState connection = iterator.next();
+ if (connection.hasRequestTimedOut(currentTimeMs)) {
+ // Mark the node as ready after request timeout
+ iterator.remove();
+ } else if (connection.isBackoffComplete(currentTimeMs)) {
+ // Mark the node as ready after completed backoff
+ iterator.remove();
+ } else if (connection.hasInflightRequest(currentTimeMs)) {
+ // There can be at most one inflight fetch request
+ return connection.remainingRequestTimeMs(currentTimeMs);
} else if (connection.isBackingOff(currentTimeMs)) {
minBackoffMs = Math.min(minBackoffMs, connection.remainingBackoffMs(currentTimeMs));
- } else {
- minBackoffMs = Math.min(minBackoffMs, connection.remainingRequestTimeMs(currentTimeMs));
}
}
+
+ // There are no inflight fetch requests so check if there is a ready bootstrap server
+ for (Node node : bootstrapServers) {
+ if (isReady(node, currentTimeMs)) {
+ return 0L;
+ }
+ }
+
+ // There are no ready bootstrap servers and inflight fetch requests, return the backoff
return minBackoffMs;
}
+ public boolean hasRequestTimedOut(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.hasRequestTimedOut(timeMs);
+ }
+
+ public boolean isReady(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return true;
+ }
+
+ boolean ready = state.isReady(timeMs);
+ if (ready) {
+ reset(node);
+ }
+
+ return ready;
+ }
+
+ public boolean isBackingOff(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.isBackingOff(timeMs);
+ }
+
+ public long remainingRequestTimeMs(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return 0;
+ }
+
+ return state.remainingRequestTimeMs(timeMs);
+ }
+
+ public long remainingBackoffMs(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return 0;
+ }
+
+ return state.remainingBackoffMs(timeMs);
Review Comment:
Thanks. Looks like I already removed this in another commit.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624737125
##########
raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java:
##########
@@ -199,6 +206,34 @@ private static Map<Integer, InetSocketAddress> parseVoterConnections(
return voterMap;
}
+ public static InetSocketAddress parseBootstrapServer(String bootstrapServer) {
+ String host = Utils.getHost(bootstrapServer);
+ if (host == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host name from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ Integer port = Utils.getPort(bootstrapServer);
+ if (port == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host port from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ return InetSocketAddress.createUnresolved(host, port);
Review Comment:
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.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624865998
##########
raft/src/main/java/org/apache/kafka/raft/FollowerState.java:
##########
@@ -156,22 +156,24 @@ public boolean canGrantVote(ReplicaKey candidateKey, boolean isLogUpToDate) {
log.debug(
"Rejecting vote request from candidate ({}) since we already have a leader {} in epoch {}",
candidateKey,
- leaderId(),
+ leader,
epoch
);
return false;
}
@Override
public String toString() {
- return "FollowerState(" +
- "fetchTimeoutMs=" + fetchTimeoutMs +
- ", epoch=" + epoch +
- ", leaderId=" + leaderId +
- ", voters=" + voters +
- ", highWatermark=" + highWatermark +
- ", fetchingSnapshot=" + fetchingSnapshot +
- ')';
+ return String.format(
Review Comment:
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.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "showuon (via GitHub)" <gi...@apache.org>.
showuon commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1620562119
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,296 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
+import java.util.Iterator;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
+/**
+ * The request manager keeps tracks of the connection with remote replicas.
+ *
+ * When sending a request update this type by calling {@code onRequestSent(Node, long, long)}. When
+ * the RPC returns a response, update this manager with {@code onResponseResult(Node, long, boolean, long)}.
+ *
+ * Connections start in the ready state ({@code isReady(Node, long)} returns true).
+ *
+ * When a request times out or completes successfully the collection will transition back to the
+ * ready state.
+ *
+ * When a request completes with an error it still transition to the backoff state until
+ * {@code retryBackoffMs}.
+ */
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<>();
+ private final ArrayList<Node> bootstrapServers;
private final int retryBackoffMs;
private final int requestTimeoutMs;
private final Random random;
- public RequestManager(Set<Integer> voterIds,
- int retryBackoffMs,
- int requestTimeoutMs,
- Random random) {
-
+ public RequestManager(
+ Collection<Node> bootstrapServers,
+ int retryBackoffMs,
+ int requestTimeoutMs,
+ Random random
+ ) {
+ this.bootstrapServers = new ArrayList<>(bootstrapServers);
this.retryBackoffMs = retryBackoffMs;
this.requestTimeoutMs = requestTimeoutMs;
- this.voters.addAll(voterIds);
this.random = random;
+ }
- for (Integer voterId: voterIds) {
- ConnectionState connection = new ConnectionState(voterId);
- connections.put(voterId, connection);
+ /**
+ * Returns true if there any connection with pending requests.
+ *
+ * This is useful for satisfying the invariant that there is only one pending Fetch request.
+ * If there are more than one pending fetch request, it is possible for the follower to write
+ * the same offset twice.
+ *
+ * @param currentTimeMs the current time
+ * @return true if the request manager is tracking at least one request
+ */
+ public boolean hasAnyInflightRequest(long currentTimeMs) {
+ boolean result = false;
+
+ Iterator<ConnectionState> iterator = connections.values().iterator();
+ while (iterator.hasNext()) {
+ ConnectionState connection = iterator.next();
+ if (connection.hasRequestTimedOut(currentTimeMs)) {
+ // Mark the node as ready after request timeout
+ iterator.remove();
+ } else if (connection.isBackoffComplete(currentTimeMs)) {
+ // Mark the node as ready after completed backoff
+ iterator.remove();
+ } else if (connection.hasInflightRequest(currentTimeMs)) {
+ // If there is at least one inflight request, it is enough
+ // to stop checking the rest of the connections
+ result = true;
+ break;
+ }
}
- }
- public ConnectionState getOrCreate(int id) {
- return connections.computeIfAbsent(id, key -> new ConnectionState(id));
+ return result;
}
- public OptionalInt findReadyVoter(long currentTimeMs) {
- int startIndex = random.nextInt(voters.size());
- OptionalInt res = OptionalInt.empty();
- for (int i = 0; i < voters.size(); i++) {
- int index = (startIndex + i) % voters.size();
- Integer voterId = voters.get(index);
- ConnectionState connection = connections.get(voterId);
- boolean isReady = connection.isReady(currentTimeMs);
+ /**
+ * Returns a random bootstrap node that is ready to receive a request.
+ *
+ * This method doesn't return a node if there is at least one request pending. In general this
+ * method is used to send Fetch requests. Fetch requests have the invariant that there can
+ * only be one pending Fetch request for the LEO.
+ *
+ * @param currentTimeMs the current time
+ * @return a random ready bootstrap node
+ */
+ public Optional<Node> findReadyBootstrapServer(long currentTimeMs) {
+ // Check that there are no infilght requests accross any of the known nodes not just
+ // the bootstrap servers
+ if (hasAnyInflightRequest(currentTimeMs)) {
+ return Optional.empty();
+ }
- if (isReady) {
- res = OptionalInt.of(voterId);
- } else if (connection.inFlightCorrelationId.isPresent()) {
- res = OptionalInt.empty();
+ int startIndex = random.nextInt(bootstrapServers.size());
+ Optional<Node> result = Optional.empty();
+ for (int i = 0; i < bootstrapServers.size(); i++) {
+ int index = (startIndex + i) % bootstrapServers.size();
+ Node node = bootstrapServers.get(index);
+
+ if (isReady(node, currentTimeMs)) {
+ result = Optional.of(node);
break;
}
}
- return res;
+
+ return result;
}
- public long backoffBeforeAvailableVoter(long currentTimeMs) {
- long minBackoffMs = Long.MAX_VALUE;
- for (Integer voterId : voters) {
- ConnectionState connection = connections.get(voterId);
- if (connection.isReady(currentTimeMs)) {
- return 0L;
+ /**
+ * Computes the amount of time needed to wait before a bootstrap server is ready for a Fetch
+ * request.
+ *
+ * If there is a connection with a pending request it returns the amount of time to wait until
+ * the request times out.
+ *
+ * Returns zero, if there are no pending request and at least one of the boorstrap servers is
+ * ready.
+ *
+ * If all of the bootstrap servers are backing off and there are no pending requests, return
+ * the minimum amount of time until a bootstrap server becomes ready.
+ *
+ * @param currentTimeMs the current time
+ * @return the amount of time to wait until bootstrap server can accept a Fetch request
+ */
+ public long backoffBeforeAvailableBootstrapServer(long currentTimeMs) {
+ long minBackoffMs = retryBackoffMs;
+
+ Iterator<ConnectionState> iterator = connections.values().iterator();
+ while (iterator.hasNext()) {
+ ConnectionState connection = iterator.next();
+ if (connection.hasRequestTimedOut(currentTimeMs)) {
+ // Mark the node as ready after request timeout
+ iterator.remove();
+ } else if (connection.isBackoffComplete(currentTimeMs)) {
+ // Mark the node as ready after completed backoff
+ iterator.remove();
+ } else if (connection.hasInflightRequest(currentTimeMs)) {
+ // There can be at most one inflight fetch request
+ return connection.remainingRequestTimeMs(currentTimeMs);
} else if (connection.isBackingOff(currentTimeMs)) {
minBackoffMs = Math.min(minBackoffMs, connection.remainingBackoffMs(currentTimeMs));
- } else {
- minBackoffMs = Math.min(minBackoffMs, connection.remainingRequestTimeMs(currentTimeMs));
}
}
+
+ // There are no inflight fetch requests so check if there is a ready bootstrap server
+ for (Node node : bootstrapServers) {
+ if (isReady(node, currentTimeMs)) {
+ return 0L;
+ }
+ }
+
+ // There are no ready bootstrap servers and inflight fetch requests, return the backoff
return minBackoffMs;
}
+ public boolean hasRequestTimedOut(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.hasRequestTimedOut(timeMs);
+ }
+
+ public boolean isReady(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return true;
+ }
+
+ boolean ready = state.isReady(timeMs);
+ if (ready) {
+ reset(node);
+ }
+
+ return ready;
+ }
+
+ public boolean isBackingOff(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return false;
+ }
+
+ return state.isBackingOff(timeMs);
+ }
+
+ public long remainingRequestTimeMs(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return 0;
+ }
+
+ return state.remainingRequestTimeMs(timeMs);
+ }
+
+ public long remainingBackoffMs(Node node, long timeMs) {
+ ConnectionState state = connections.get(node.idString());
+ if (state == null) {
+ return 0;
+ }
+
+ return state.remainingBackoffMs(timeMs);
Review Comment:
nit: additional space after return.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1610641974
##########
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala:
##########
@@ -1366,6 +1368,26 @@ class KafkaConfigTest {
assertEquals(expectedVoters, addresses)
}
+ @Test
+ def testParseQuorumBootstrapServers(): Unit = {
Review Comment:
Okay. It does look like there are bugs in the 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
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1611946739
##########
raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java:
##########
@@ -231,4 +266,26 @@ public String toString() {
return "non-empty list";
}
}
+
+ public static class ControllerQuorumBootstrapServersValidator implements ConfigDef.Validator {
+ @Override
+ public void ensureValid(String name, Object value) {
+ if (value == null) {
+ throw new ConfigException(name, null);
+ }
+
+ @SuppressWarnings("unchecked")
+ List<String> entries = (List<String>) value;
+
+ // Attempt to parse the connect strings
+ for (String entry : entries) {
+ QuorumConfig.parseBootstrapServer(entry);
Review Comment:
Fixed.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1610641974
##########
core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala:
##########
@@ -1366,6 +1368,26 @@ class KafkaConfigTest {
assertEquals(expectedVoters, addresses)
}
+ @Test
+ def testParseQuorumBootstrapServers(): Unit = {
Review Comment:
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
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619520419
##########
raft/src/main/java/org/apache/kafka/raft/internals/VoterSet.java:
##########
@@ -55,15 +57,41 @@ final public class VoterSet {
}
/**
- * Returns the socket address for a given voter at a given listener.
+ * Returns the node information for all the given voter ids and listener.
*
- * @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 ids of the voters
+ * @param listenerName the name of the listener
+ * @return the node information for all of the voter ids
+ * @throws IllegalArgumentException if there are missing endpoints
*/
- 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) {
Review Comment:
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?
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624716888
##########
raft/src/main/java/org/apache/kafka/raft/internals/VoterSet.java:
##########
@@ -55,15 +57,41 @@ final public class VoterSet {
}
/**
- * Returns the socket address for a given voter at a given listener.
+ * Returns the node information for all the given voter ids and listener.
*
- * @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 ids of the voters
+ * @param listenerName the name of the listener
+ * @return the node information for all of the voter ids
+ * @throws IllegalArgumentException if there are missing endpoints
*/
- 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) {
Review Comment:
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.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on PR #15986:
URL: https://github.com/apache/kafka/pull/15986#issuecomment-2145823988
> 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.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624858821
##########
raft/src/main/java/org/apache/kafka/raft/QuorumConfig.java:
##########
@@ -199,6 +206,34 @@ private static Map<Integer, InetSocketAddress> parseVoterConnections(
return voterMap;
}
+ public static InetSocketAddress parseBootstrapServer(String bootstrapServer) {
+ String host = Utils.getHost(bootstrapServer);
+ if (host == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host name from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ Integer port = Utils.getPort(bootstrapServer);
+ if (port == null) {
+ throw new ConfigException(
+ String.format(
+ "Failed to parse host port from {} for the configuration {}. Each " +
+ "entry should be in the form \"{host}:{port}\"",
+ bootstrapServer,
+ QUORUM_BOOTSTRAP_SERVERS_CONFIG
+ )
+ );
+ }
+
+ return InetSocketAddress.createUnresolved(host, port);
Review Comment:
oh, I thought `Utils.getHost` resolved the hostname. I can see now that it doesn't. Fair enough.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624871047
##########
raft/src/main/java/org/apache/kafka/raft/internals/VoterSet.java:
##########
@@ -55,15 +57,41 @@ final public class VoterSet {
}
/**
- * Returns the socket address for a given voter at a given listener.
+ * Returns the node information for all the given voter ids and listener.
*
- * @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 ids of the voters
+ * @param listenerName the name of the listener
+ * @return the node information for all of the voter ids
+ * @throws IllegalArgumentException if there are missing endpoints
*/
- 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) {
Review Comment:
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.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe merged PR #15986:
URL: https://github.com/apache/kafka/pull/15986
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624698613
##########
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}
Review Comment:
Yeah. Next time. I can't help but clean up code as I see it.
##########
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
-
Review Comment:
Yes. Next time.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on PR #15986:
URL: https://github.com/apache/kafka/pull/15986#issuecomment-2145613112
> 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`.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1619506788
##########
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}
Review Comment:
Do this in a separate cleanup PR.
##########
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
-
Review Comment:
Do this in a separate cleanup PR.
--
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
Re: [PR] KAFKA-16252; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on PR #15986:
URL: https://github.com/apache/kafka/pull/15986#issuecomment-2138338743
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. Unless 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.
- I think we should try to avoid doing too much validation in `KafkaConfig`. 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.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "jsancio (via GitHub)" <gi...@apache.org>.
jsancio commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624733666
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,296 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
+import java.util.Iterator;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
+/**
+ * The request manager keeps tracks of the connection with remote replicas.
+ *
+ * When sending a request update this type by calling {@code onRequestSent(Node, long, long)}. When
+ * the RPC returns a response, update this manager with {@code onResponseResult(Node, long, boolean, long)}.
+ *
+ * Connections start in the ready state ({@code isReady(Node, long)} returns true).
+ *
+ * When a request times out or completes successfully the collection will transition back to the
+ * ready state.
+ *
+ * When a request completes with an error it still transition to the backoff state until
+ * {@code retryBackoffMs}.
+ */
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<>();
+ private final ArrayList<Node> bootstrapServers;
private final int retryBackoffMs;
private final int requestTimeoutMs;
private final Random random;
- public RequestManager(Set<Integer> voterIds,
- int retryBackoffMs,
- int requestTimeoutMs,
- Random random) {
-
+ public RequestManager(
+ Collection<Node> bootstrapServers,
+ int retryBackoffMs,
+ int requestTimeoutMs,
+ Random random
+ ) {
+ this.bootstrapServers = new ArrayList<>(bootstrapServers);
this.retryBackoffMs = retryBackoffMs;
this.requestTimeoutMs = requestTimeoutMs;
- this.voters.addAll(voterIds);
this.random = random;
+ }
- for (Integer voterId: voterIds) {
- ConnectionState connection = new ConnectionState(voterId);
- connections.put(voterId, connection);
+ /**
+ * Returns true if there any connection with pending requests.
+ *
+ * This is useful for satisfying the invariant that there is only one pending Fetch request.
+ * If there are more than one pending fetch request, it is possible for the follower to write
+ * the same offset twice.
+ *
+ * @param currentTimeMs the current time
+ * @return true if the request manager is tracking at least one request
+ */
+ public boolean hasAnyInflightRequest(long currentTimeMs) {
+ boolean result = false;
+
+ Iterator<ConnectionState> iterator = connections.values().iterator();
+ while (iterator.hasNext()) {
+ ConnectionState connection = iterator.next();
+ if (connection.hasRequestTimedOut(currentTimeMs)) {
+ // Mark the node as ready after request timeout
+ iterator.remove();
+ } else if (connection.isBackoffComplete(currentTimeMs)) {
+ // Mark the node as ready after completed backoff
+ iterator.remove();
+ } else if (connection.hasInflightRequest(currentTimeMs)) {
Review Comment:
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.
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624867453
##########
raft/src/main/java/org/apache/kafka/raft/RequestManager.java:
##########
@@ -17,108 +17,296 @@
package org.apache.kafka.raft;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.HashMap;
-import java.util.List;
+import java.util.Iterator;
import java.util.Map;
-import java.util.OptionalInt;
+import java.util.Optional;
import java.util.OptionalLong;
import java.util.Random;
-import java.util.Set;
+import org.apache.kafka.common.Node;
+/**
+ * The request manager keeps tracks of the connection with remote replicas.
+ *
+ * When sending a request update this type by calling {@code onRequestSent(Node, long, long)}. When
+ * the RPC returns a response, update this manager with {@code onResponseResult(Node, long, boolean, long)}.
+ *
+ * Connections start in the ready state ({@code isReady(Node, long)} returns true).
+ *
+ * When a request times out or completes successfully the collection will transition back to the
+ * ready state.
+ *
+ * When a request completes with an error it still transition to the backoff state until
+ * {@code retryBackoffMs}.
+ */
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<>();
Review Comment:
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
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624863508
##########
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();
Review Comment:
thanks
--
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
Re: [PR] KAFKA-16525; Dynamic KRaft network manager and channel [kafka]
Posted by "cmccabe (via GitHub)" <gi...@apache.org>.
cmccabe commented on code in PR #15986:
URL: https://github.com/apache/kafka/pull/15986#discussion_r1624855666
##########
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],
Review Comment:
ok.
--
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