Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-5316431652 I have file a PR for the behavior change issue: #23176 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-5296549054 @lianetm @AndrewJSchofield Thanks for bringing this up. I left an idea in the ticket, and we can discuss further there. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
AndrewJSchofield commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-5296166069 I think we may be able to retain the previous behaviour by default and opt in to the new KIP-909 by setting a non-default value for the `bootstrap.resolve.timeout.ms` config. We'd have to change the default value to, say, 0 meaning use old behaviour. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-5296120482 Hey folks, we're noticing the impact this change is having on client apps as a breaking change, requiring app logic updates. E.g., app logic handling DNS failures in the pre-kip-909 way (expecting ConfigException when creating the consumer/prodiucer/admin, and handling it to re-create the instance assuming DNS errors expected), vs now getting a working client instance, but that fails fatally later on on different operations where there is nothing expecting a recreation needed. Given that 4.4 is a minor, this is probably something we need to reconsider. Thoughts? cc. @frankvicky , @chia7712 https://issues.apache.org/jira/browse/KAFKA-20939 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3694456769
##
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##
@@ -3398,4 +3406,59 @@ public static void resetCounters() {
CLOSE_COUNT.set(0);
}
}
+
+@Test
+public void testProducerBootstrapResolutionExceptionPropagated() {
+String invalidHost = "unresolvable.invalid:9092";
+Map configs = Map.of(
+ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, invalidHost,
+CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG, "3000"
+);
+
+try (KafkaProducer producer = new
KafkaProducer<>(configs)) {
+assertThrows(BootstrapResolutionException.class, () -> {
+long startTime = System.currentTimeMillis();
+long maxWaitTime = 15000;
+while (System.currentTimeMillis() - startTime < maxWaitTime) {
+producer.partitionsFor("test-topic");
+}
+fail("Expected BootstrapResolutionException to be thrown
within " + maxWaitTime + "ms");
+});
+
+// After the first failure, any further API call must also throw.
This guards against
+// accidentally clearing the bootstrap error from the metadata
layer.
+assertThrows(BootstrapResolutionException.class, () ->
producer.partitionsFor("test-topic"));
+}
+}
+
+@Test
+public void
testProducerSendOffsetsToTransactionBootstrapResolutionExceptionPropagated() {
+// sendOffsetsToTransaction became metadata-aware in KIP-1319, so it
can also block on
+// bootstrap and must surface BootstrapResolutionException.
+String invalidHost = "unresolvable.invalid:9092";
+Map configs = Map.of(
+ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, invalidHost,
+CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG, "3000",
+ProducerConfig.TRANSACTIONAL_ID_CONFIG, "test-tx-id"
+);
+
+Map offsets = Map.of(
+new TopicPartition("test-topic", 0),
+new OffsetAndMetadata(0L)
+);
+ConsumerGroupMetadata groupMetadata = new
ConsumerGroupMetadata("test-group");
Review Comment:
We should use mock instead of constructing `ConsumerGroupMetadata`.
```java
ConsumerGroupMetadata groupMetadata =
mock(ConsumerGroupMetadata.class);
when(groupMetadata.groupId()).thenReturn("test-group");
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3690784458
##
clients/src/main/java/org/apache/kafka/clients/BootstrapConfiguration.java:
##
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients;
+
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.utils.Utils;
+
+import java.util.List;
+
+public final class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
Review Comment:
Don't we need a similar "Invalid port" check at this level (early when
reading the configs, not when resolving DNS).
We have it here
https://github.com/apache/kafka/blob/971c6b413103aa0db6e629fbe2eba32bd6024f52/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java#L140
but we don't go down that path from the client anymore right?
So the concern is what happens if we configure an invalid port and start a
client? I expect it should fail fast (not attempt DNS resolution for the whole
timeout with the wrong port), and to have that I expect we need the port check
on this enabled?
btw, I think we might also need a clean up on the
`parseAndValidateAddressess`? There are so many of those, but doesn't seem we
used them all anymore after this and the ShareConsumer PR.
##
clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java:
##
@@ -60,6 +60,14 @@ public class CommonClientConfigs {
+ "resolve each
bootstrap address into a list of canonical names. After "
+ "the bootstrap phase,
this behaves the same as use_all_dns_ips.";
+public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG =
"bootstrap.resolve.timeout.ms";
+public static final long DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS = 2 * 60 *
1000L;
+public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_DOC = "Maximum
amount of time clients can spend trying to" +
+" resolve for the bootstrap server address. If the resolution cannot
be completed within this timeframe, a " +
Review Comment:
do we mean "resolve DNS for ..."?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-5111850126 Except for the follow-up tasks, I have triggered full [E2E tests](https://jenkins.opensource4you.tw/job/kafka-e2e/869/console) :) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 merged PR #21080: URL: https://github.com/apache/kafka/pull/21080 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3657655129
##
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/AdminMetadataTest.java:
##
@@ -142,19 +141,12 @@ public void testListTopicsWithOptionTimeoutMs() {
@ClusterTest
public void testListTopicsWithOptionListInternal() throws Exception {
try (Admin admin = clusterInstance.admin()) {
-String topic = "test-topic";
-admin.createTopics(List.of(new NewTopic(topic, 1, (short)
1))).all().get();
-clusterInstance.waitTopicCreation(topic, 1);
+admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1))).all().get();
+clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME,
1);
-try (Consumer consumer =
clusterInstance.consumer()) {
-consumer.subscribe(List.of(topic));
-consumer.poll(Duration.ofMillis(100));
-}
-
-TestUtils.waitForCondition(() -> {
-Set topicNames = admin.listTopics(new
ListTopicsOptions().listInternal(true)).names().get();
-return topicNames.contains(Topic.GROUP_METADATA_TOPIC_NAME);
-}, "Expected to see internal topic " +
Topic.GROUP_METADATA_TOPIC_NAME);
+Set topicNames = admin.listTopics(new
ListTopicsOptions().listInternal(true)).names().get();
+assertTrue(topicNames.contains(Topic.GROUP_METADATA_TOPIC_NAME),
+"Expected to see internal topic " +
Topic.GROUP_METADATA_TOPIC_NAME);
Review Comment:
https://github.com/apache/kafka/pull/22960
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3657541545
##
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/AdminMetadataTest.java:
##
@@ -142,19 +141,12 @@ public void testListTopicsWithOptionTimeoutMs() {
@ClusterTest
public void testListTopicsWithOptionListInternal() throws Exception {
try (Admin admin = clusterInstance.admin()) {
-String topic = "test-topic";
-admin.createTopics(List.of(new NewTopic(topic, 1, (short)
1))).all().get();
-clusterInstance.waitTopicCreation(topic, 1);
+admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1))).all().get();
+clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME,
1);
-try (Consumer consumer =
clusterInstance.consumer()) {
-consumer.subscribe(List.of(topic));
-consumer.poll(Duration.ofMillis(100));
-}
-
-TestUtils.waitForCondition(() -> {
-Set topicNames = admin.listTopics(new
ListTopicsOptions().listInternal(true)).names().get();
-return topicNames.contains(Topic.GROUP_METADATA_TOPIC_NAME);
-}, "Expected to see internal topic " +
Topic.GROUP_METADATA_TOPIC_NAME);
+Set topicNames = admin.listTopics(new
ListTopicsOptions().listInternal(true)).names().get();
+assertTrue(topicNames.contains(Topic.GROUP_METADATA_TOPIC_NAME),
+"Expected to see internal topic " +
Topic.GROUP_METADATA_TOPIC_NAME);
Review Comment:
I think this should be fixed by separate PR to slim this PR down. WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3657478146
##
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/admin/AdminMetadataTest.java:
##
@@ -142,19 +141,12 @@ public void testListTopicsWithOptionTimeoutMs() {
@ClusterTest
public void testListTopicsWithOptionListInternal() throws Exception {
try (Admin admin = clusterInstance.admin()) {
-String topic = "test-topic";
-admin.createTopics(List.of(new NewTopic(topic, 1, (short)
1))).all().get();
-clusterInstance.waitTopicCreation(topic, 1);
+admin.createTopics(List.of(new
NewTopic(Topic.GROUP_METADATA_TOPIC_NAME, 1, (short) 1))).all().get();
+clusterInstance.waitTopicCreation(Topic.GROUP_METADATA_TOPIC_NAME,
1);
-try (Consumer consumer =
clusterInstance.consumer()) {
-consumer.subscribe(List.of(topic));
-consumer.poll(Duration.ofMillis(100));
-}
-
-TestUtils.waitForCondition(() -> {
-Set topicNames = admin.listTopics(new
ListTopicsOptions().listInternal(true)).names().get();
-return topicNames.contains(Topic.GROUP_METADATA_TOPIC_NAME);
-}, "Expected to see internal topic " +
Topic.GROUP_METADATA_TOPIC_NAME);
+Set topicNames = admin.listTopics(new
ListTopicsOptions().listInternal(true)).names().get();
+assertTrue(topicNames.contains(Topic.GROUP_METADATA_TOPIC_NAME),
+"Expected to see internal topic " +
Topic.GROUP_METADATA_TOPIC_NAME);
Review Comment:
Rewrite `testListTopicsWithOptionListInternal` to explicitly create the
`__consumer_offsets`
topic via Admin rather than relying on a `consumer.poll(100ms)` to trigger
group formation.
The Scala predecessor of this test relied on `IntegrationTestHarness.setUp()`
pre-creating `__consumer_offsets` via `createOffsetsTopic()`. The Java
migration in
#22290 replaced that with a `consumer.subscribe` + `poll(100ms)` sequence,
which
races against internal group join timing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3655771060
##
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##
@@ -1086,6 +1091,7 @@ public Future send(ProducerRecord
record) {
*when send is invoked after producer has
been closed.
* @throws InterruptException If the thread is interrupted while
blocked
* @throws SerializationException If the key or value are not valid
objects given the configured serializers
+ * @throws BootstrapResolutionException if DNS resolution of the bootstrap
servers fails within {@code bootstrap.resolve.timeout.ms}
Review Comment:
What should users do if they encounter a `BootstrapResolutionException` ?
Since this exception is never cleared, they must close the producer, right? If
so, we should explicitly document that it is an unrecoverable error.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3655685820
##
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##
@@ -633,6 +636,8 @@ public synchronized void maybeThrowExceptionForTopic(String
topic) {
}
private void clearErrorsAndMaybeThrowException(Supplier
recoverableExceptionSupplier) {
+if (bootstrapFatalException != null)
Review Comment:
Could you reuse `maybeThrowBootstrapFatalException`?
##
clients/src/main/java/org/apache/kafka/clients/Metadata.java:
##
@@ -616,6 +617,8 @@ public synchronized void maybeThrowAnyException() {
* in the last metadata update.
*/
protected synchronized void maybeThrowFatalException() {
+if (bootstrapFatalException != null)
Review Comment:
Could you reuse `maybeThrowBootstrapFatalException`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3655665050
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1207,6 +1287,152 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
Review Comment:
Would you mind moving `BootstrapConfiguration` out of `NetworkClient` to
slim it down?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3655642608
##
clients/src/main/java/org/apache/kafka/clients/MetadataSnapshot.java:
##
@@ -227,8 +227,8 @@ private void computeClusterView() {
.stream()
.map(metadata -> MetadataResponse.toPartitionInfo(metadata,
nodes))
.collect(Collectors.toList());
-this.clusterInstance = new Cluster(clusterId, nodes.values(),
partitionInfos, unauthorizedTopics,
-invalidTopics, internalTopics, controller, topicIds);
+this.clusterInstance = new Cluster(clusterId, nodes.values(),
partitionInfos,
Review Comment:
unrelated 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080: URL: https://github.com/apache/kafka/pull/21080#discussion_r3655548834 ## clients/src/main/java/org/apache/kafka/clients/ClientUtils.java: ## @@ -201,6 +247,17 @@ public static NetworkClient createNetworkClient(AbstractConfig config, metricsGroupPrefix, channelBuilder, logContext); +ClientDnsLookup dnsLookup = ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG)); + +NetworkClient.BootstrapConfiguration bootstrapConfiguration = +(bootstrapServers != null && !bootstrapServers.isEmpty()) Review Comment: That branch was dead. Every `ClientUtils.createNetworkClient` caller is guaranteed non-empty: - Producer/Consumer: `bootstrap.servers` is validated with `anyNonDuplicateValues(false, false)` (empty not allowed) - Admin: `determineBootstrapType` throws `ConfigException` if neither `bootstrap.servers` nor `bootstrap.controllers` is set The `DISABLED` path is only used by broker-side code (`NodeToControllerChannelManagerImpl`, `NetworkUtils`, `ReplicaVerificationTool`, `BrokerApiVersionsCommand`, `ConnectionStressWorker`), and those construct `NetworkClient` directly — they never go through `ClientUtils`. Simplified to always use `enabled(...)`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3655475851
##
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##
@@ -538,12 +575,19 @@ static KafkaAdminClient createInternal(
try {
// Since we only request node information, it's safe to pass true
for allowAutoTopicCreation (and it
// simplifies communication with older brokers)
-AdminBootstrapAddresses adminAddresses =
AdminBootstrapAddresses.fromConfig(config);
Review Comment:
Already gone — `AdminBootstrapAddresses` (and its test) were deleted in
`e3159c8c5b` a while back.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3655161602
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -835,6 +899,21 @@ public LeastLoadedNode leastLoadedNode(long now) {
}
}
+/**
+ * Handle the case when there are no nodes available.
+ *
+ * If bootstrap is disabled or already complete, throw
IllegalStateException.
+ * If bootstrap is enabled but not yet complete, return null to allow DNS
resolution to continue.
Review Comment:
this statement is out of date.
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -201,6 +247,17 @@ public static NetworkClient
createNetworkClient(AbstractConfig config,
metricsGroupPrefix,
channelBuilder,
logContext);
+ClientDnsLookup dnsLookup =
ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG));
+
+NetworkClient.BootstrapConfiguration bootstrapConfiguration =
+(bootstrapServers != null && !bootstrapServers.isEmpty())
Review Comment:
Why do we have a path that does not require the bootstrap servers?
##
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java:
##
@@ -148,6 +149,17 @@ public void rebootstrap(long now) {
AdminMetadataManager.this.rebootstrap(now);
}
+@Override
+public boolean isBootstrapped() {
+return bootstrapCluster != null;
+}
+
+@Override
+public void bootstrap(List addresses) {
+// Called by NetworkClient.ensureBootstrapped() after DNS
resolution succeeds
+AdminMetadataManager.this.update(Cluster.bootstrap(addresses), 0);
Review Comment:
Would you mind adding comments for the magic `0`?
##
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##
@@ -538,12 +575,19 @@ static KafkaAdminClient createInternal(
try {
// Since we only request node information, it's safe to pass true
for allowAutoTopicCreation (and it
// simplifies communication with older brokers)
-AdminBootstrapAddresses adminAddresses =
AdminBootstrapAddresses.fromConfig(config);
Review Comment:
It seems `AdminBootstrapAddresses` is useless. Could you remove it as well?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
Copilot commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3636336228
##
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java:
##
@@ -1630,4 +1654,110 @@ public KafkaException getAndClearFailure() {
return failure;
}
}
+
+private void bootstrapMetadataWithNodes(Metadata metadata, List
nodes) {
+List serverAddresses = new ArrayList<>();
+nodes.forEach(node -> serverAddresses.add(new
InetSocketAddress(node.host(), node.port(;
+metadata.bootstrap(serverAddresses);
+}
+
+private void bootstrapMetadata(Metadata metadata) {
+List serverAddresses = new ArrayList<>(List.of(
+new InetSocketAddress("localhost0", 8000),
+new InetSocketAddress("localhost1", 8000)
+));
+metadata.bootstrap(serverAddresses);
Review Comment:
These helper methods use new InetSocketAddress("localhost0", ...) /
("localhost1", ...) which can trigger real DNS resolution during test
execution. Using createUnresolved avoids environmental DNS dependencies and
reduces flakiness.
This issue also appears on line 1673 of the same file.
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -835,6 +899,21 @@ public LeastLoadedNode leastLoadedNode(long now) {
}
}
+/**
+ * Handle the case when there are no nodes available.
+ *
+ * If bootstrap is disabled or already complete, throw
IllegalStateException.
+ * If bootstrap is enabled but not yet complete, return null to allow DNS
resolution to continue.
+ */
Review Comment:
The Javadoc for handleEmptyNodeList() says it will "return null" when
bootstrap is enabled but not yet complete, but the method actually returns a
non-null LeastLoadedNode with a null node. This mismatch is confusing for
callers/readers; update the comment to describe the real behavior.
##
clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java:
##
@@ -60,6 +60,12 @@ public class CommonClientConfigs {
+ "resolve each
bootstrap address into a list of canonical names. After "
+ "the bootstrap phase,
this behaves the same as use_all_dns_ips.";
+public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG =
"bootstrap.resolve.timeout.ms";
+public static final long DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS = 2 * 60 *
1000L;
+public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_DOC = "Maximum
amount of time clients can spend trying to" +
+" resolve for the bootstrap server address. If the resolution cannot
be completed within this timeframe, a " +
+"BootstrapResolutionException will be thrown.";
Review Comment:
The documentation string for bootstrap.resolve.timeout.ms has an awkward
phrase ("resolve for the bootstrap server address"). Rewording it improves
clarity.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3628846348
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1207,6 +1296,155 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Start the timer on the first poll so its budget represents "time we
spend on
+// bootstrap since polling began" — the caller may have created the
client well before
+// its first API call, and we don't want that idle gap to eat into the
budget.
+if (bootstrapTimer == null)
+bootstrapTimer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
+
+// Check if a pending resolution completed before checking the
timeout, so that a
+// result arriving at the same time as the deadline is not incorrectly
rejected.
+if (maybeProcessBootstrapResolutionResult(currentTimeMs))
Review Comment:
`checkBootstrapTimeout` doesn't actually throw — it just records the failure
on the metadata, callers surface it later via `maybeThrowFatalException`. And
on the very first poll, the timer isn't expired yet:
1. `bootstrapTimer = time.timer(1)` → `startMs = time.milliseconds() = T`,
deadline = T+1
2. `maybeProcessBootstrapResolutionResult(currentTimeMs)` — if the eager
pending future already completed (fast DNS), we bootstrap and early-return
3. Otherwise `bootstrapTimer.update(currentTimeMs)` keeps cached time at T
(Timer.update uses max)
4. `checkBootstrapTimeout` → `T >= T+1` → false, nothing recorded
So the first poll always gets a shot: either the eager resolution has
finished (bootstrap succeeds), or we exit without recording a failure. The
timer can only fire from the second poll onward, once ≥1ms of wall-clock has
elapsed. With `bootstrap.resolve.timeout.ms` now enforced to be `atLeast(1L)`,
this gives the "at least one attempt" guarantee.
Also cleaned up a stale comment that still referenced
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3628673895
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##
@@ -2204,6 +2201,13 @@ private void acquireAndEnsureOpen() {
release();
throw new IllegalStateException("This consumer has already been
closed.");
}
+
+try {
+metadata.maybeThrowBootstrapFatalException();
Review Comment:
Fair point — kept it.
Originally left off because for the classic consumer any network-hitting
call already surfaces the failure via `metadata.awaitUpdate` →
`maybeThrowFatalException` (which checks `bootstrapFatalException`), so
`testConsumerBootstrapResolutionExceptionPropagatedToPoll` passes for both
`CLASSIC` and `CONSUMER`. But
(1) it aligns Classic / Async / Share, and
(2) it also covers pure-local APIs (`subscribe`, `unsubscribe`, `assign`,
`wakeup`, query methods) that never poll, so the caller sees the error
immediately.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3628557832
##
clients/src/main/java/org/apache/kafka/common/errors/BootstrapResolutionException.java:
##
@@ -0,0 +1,37 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.common.errors;
+
+import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.annotation.InterfaceAudience;
+
+/**
+ * Indicates that the {@link org.apache.kafka.clients.NetworkClient} was
unable to resolve a DNS address within
+ * the time specified by {@link
org.apache.kafka.clients.CommonClientConfigs#BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG}
+ *
+ * @see org.apache.kafka.clients.CommonClientConfigs
+ */
[email protected]
+public class BootstrapResolutionException extends KafkaException {
+public BootstrapResolutionException(String message) {
Review Comment:
Have you considered adding it to consumer docs?
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##
@@ -2204,6 +2201,13 @@ private void acquireAndEnsureOpen() {
release();
throw new IllegalStateException("This consumer has already been
closed.");
}
+
+try {
+metadata.maybeThrowBootstrapFatalException();
Review Comment:
Why is `metadata.maybeThrowBootstrapFatalException()`; only applied to the
async consumer? Should we port it to the classic consumer?
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -750,6 +803,18 @@ private void ensureActive() {
public void close() {
state.compareAndSet(State.ACTIVE, State.CLOSING);
if (state.compareAndSet(State.CLOSING, State.CLOSED)) {
+cancelBootstrapResolution();
+if (bootstrapExecutor != null) {
+bootstrapExecutor.shutdownNow();
+// Wait briefly for the executor's worker thread to actually
terminate. Otherwise
+// a lingering thread can trip test-side thread-leak detectors
even though the
+// executor has already been shut down.
+try {
+bootstrapExecutor.awaitTermination(1, TimeUnit.SECONDS);
Review Comment:
Could you leverage `ThreadUtils.shutdownExecutorServiceQuietly`?
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1207,6 +1296,155 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3599465469
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1207,6 +1296,155 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Start the timer on the first poll so its budget represents "time we
spend on
+// bootstrap since polling began" — the caller may have created the
client well before
+// its first API call, and we don't want that idle gap to eat into the
budget.
+if (bootstrapTimer == null)
+bootstrapTimer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
Review Comment:
Cluster.bootstrap(...) uses InetSocketAddress.getHostString(), so the
resulting Nodes keep the original hostname. On rebootstrap,
metadata.rebootstrap() → metadata.bootstrap(cached addresses) →
Cluster.bootstrap(...) → Nodes with hostnames. Any subsequent connection
attempt goes through ClusterConnectionStates which performs a synchronous DNS
resolution, so DNS changes ARE picked up.
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,137 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3597807141
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1207,6 +1296,155 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Start the timer on the first poll so its budget represents "time we
spend on
+// bootstrap since polling began" — the caller may have created the
client well before
+// its first API call, and we don't want that idle gap to eat into the
budget.
+if (bootstrapTimer == null)
+bootstrapTimer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
Review Comment:
Should we re-create the bootstrapTimer if the re-bootstrap process happens?
Or does bootstrap.resolve.timeout.ms mean the total elapsed time we can use to
resolve DNS? If so, that seems to be an issue for a long-running client, since
it could time out immediately when encountering a re-bootstrap.
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,137 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080: URL: https://github.com/apache/kafka/pull/21080#discussion_r3595715920 ## clients/src/main/java/org/apache/kafka/clients/NetworkClient.java: ## @@ -391,11 +391,14 @@ public NetworkClient(MetadataUpdater metadataUpdater, this.metadataRecoveryStrategy = metadataRecoveryStrategy; this.metadataClusterCheckEnable = metadataClusterCheckEnable; this.bootstrapConfiguration = bootstrapConfiguration; -// Bootstrap timer is lazily initialized on first poll to ensure timeout starts when polling begins Review Comment: Good catch — decoupled the two. Constructor now only kicks off the pending resolution (CompletableFuture.supplyAsync(...) directly), without creating the bootstrap timer. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080: URL: https://github.com/apache/kafka/pull/21080#discussion_r3595257209 ## clients/src/main/java/org/apache/kafka/clients/NetworkClient.java: ## @@ -391,11 +391,14 @@ public NetworkClient(MetadataUpdater metadataUpdater, this.metadataRecoveryStrategy = metadataRecoveryStrategy; this.metadataClusterCheckEnable = metadataClusterCheckEnable; this.bootstrapConfiguration = bootstrapConfiguration; -// Bootstrap timer is lazily initialized on first poll to ensure timeout starts when polling begins Review Comment: don't we need to keep like behaviour? (timer starting on the first poll). If we start the timer any sooner and let's say dns fails, it won't be retried until the first poll, but we would still be consuming 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3581457215
##
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerCommitTest.java:
##
@@ -290,6 +290,12 @@ private void testAsyncCommit(GroupProtocol groupProtocol)
throws InterruptedExce
try (var consumer = createConsumer(groupProtocol, false)) {
consumer.assign(List.of(tp));
+// Warm up: force bootstrap DNS resolution and group coordinator
discovery before
+// issuing async commits. Without this, KIP-909's deferred
bootstrap can push
+// coordinator discovery past the first commitAsync attempt,
yielding a
+// RetriableCommitFailedException that this test does not tolerate.
+consumer.commitSync(Map.of(tp, new OffsetAndMetadata(0L)));
Review Comment:
Interesting. I do get the situation and the need for the pre-warm. But then
it makes me wonder, is there a reason why we have to wait for the first call to
poll to trigger the async resolution?? vs triggering it early when the client
is created? (we know the result is needed anyways and that it may take time to
resolve).
Meaning we could trigger an async resolution when the network client is
created (no time-boundary or check here, it just an early trigger to get
ahead). Then poll remains the main driver of the bootstrap process as it is
now: it is the one checking the result, enforcing the timer, propagating errors
if any, retrying etc. Only the trigger moves to an earlier point in time
(constructor)
Even with that we may still have the situation this test shows (delayed DNS
resolution), but just wondering if it's worth/doable to avoid the delay we are
introducing ourselves compared to how it was before.
Makes sense or am I missing something ? We can address as an improvement in
a follow-up if we think it makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3580357456
##
clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java:
##
@@ -64,7 +64,10 @@ public class CommonClientConfigs {
public static final long DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS = 2 * 60 *
1000L;
public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_DOC = "Maximum
amount of time clients can spend trying to" +
" resolve for the bootstrap server address. If the resolution cannot
be completed within this timeframe, a " +
-"BootstrapResolutionException will be thrown.";
+"BootstrapResolutionException will be thrown. Setting
this to 0 means the client" +
+" attempts resolution once without waiting for it to complete: the
resolution is triggered on the first" +
+" poll and its result is checked on the next; if the address is not
resolved by then, a" +
Review Comment:
the concept of poll (network poll) here may be too low level for a
user-facing config imo , wdyt?.
We could generalize maybe, loosely saying that it is expect to be resolved
immediately/right away (e.g., "..the client attempts resolution once and checks
the result right after, fails with BootstrapEx if bootstrap did not complete").
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3574183327
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,137 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Check if a pending resolution completed before checking the
timeout, so that a
+// result arriving at the same time as the deadline is not incorrectly
rejected.
+if (pendingBootstrapResolution != null &&
pendingBootstrapResolution.isDone()) {
+processBootstrapResolutionResult(currentTimeMs);
+if (metadataUpdater.isBootstrapped())
+return;
+}
+
+maybeStartBootstrapResolution(currentTimeMs);
Review Comment:
do we have a test to cover this timeout=0 behaviour? let's add it if not
present, seems something that could easily break with a simple
reorder/refactoring of this flow.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3574183327
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,137 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Check if a pending resolution completed before checking the
timeout, so that a
+// result arriving at the same time as the deadline is not incorrectly
rejected.
+if (pendingBootstrapResolution != null &&
pendingBootstrapResolution.isDone()) {
+processBootstrapResolutionResult(currentTimeMs);
+if (metadataUpdater.isBootstrapped())
+return;
+}
+
+maybeStartBootstrapResolution(currentTimeMs);
Review Comment:
do we have a test to cover this timeout=0 behaviour? let's add it if not
present, seems something that could easily break we a simple
reorder/refactoring of this flow.
##
clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java:
##
@@ -60,6 +60,12 @@ public class CommonClientConfigs {
+ "resolve each
bootstrap address into a list of canonical names. After "
+ "the bootstrap phase,
this behaves the same as use_all_dns_ips.";
+public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG =
"bootstrap.resolve.timeout.ms";
+public static final long DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS = 2 * 60 *
1000L;
+public static final String BOOTSTRAP_RESOLVE_TIMEOUT_MS_DOC = "Maximum
amount of time clients can spend trying to" +
+" resolve for the bootstrap server address. If the resolution cannot
be completed within this timeframe, a " +
+"BootstrapResolutionException will be thrown.";
Review Comment:
should we clarify the timeout=0 behaviour after the latest changes? (mention
that 0 means the client attempts resolution once and fails with
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3571811025
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,137 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Check if a pending resolution completed before checking the
timeout, so that a
+// result arriving at the same time as the deadline is not incorrectly
rejected.
+if (pendingBootstrapResolution != null &&
pendingBootstrapResolution.isDone()) {
+processBootstrapResolutionResult(currentTimeMs);
+if (metadataUpdater.isBootstrapped())
+return;
+}
+
+maybeStartBootstrapResolution(currentTimeMs);
Review Comment:
Make sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3571743450
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,137 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Check if a pending resolution completed before checking the
timeout, so that a
+// result arriving at the same time as the deadline is not incorrectly
rejected.
+if (pendingBootstrapResolution != null &&
pendingBootstrapResolution.isDone()) {
+processBootstrapResolutionResult(currentTimeMs);
+if (metadataUpdater.isBootstrapped())
+return;
+}
+
+maybeStartBootstrapResolution(currentTimeMs);
Review Comment:
how was this addressed? was it by disallowing 0 (atLeast(1))? (not sure if
I'm missing something)
If so, disallowing 0 is not really giving it a shot, what's the intention?
If we want to allow 0 with the meaning that it will attempt bootstrap but do
not wait for it, I guess it could be taken as trigger resolution and check
result on the next poll (fail if not resolved)?.
If that's the intention we probably need to re-adjust the sequence to check
the timeout before triggering the resolution?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3527912504
##
tools/src/main/java/org/apache/kafka/tools/BrokerApiVersionsCommand.java:
##
@@ -182,6 +183,11 @@ static AdminClient create(AbstractConfig config) {
"admin",
ClientUtils.createChannelBuilder(config, time, logContext),
logContext);
+NetworkClient.BootstrapConfiguration bootstrapConfiguration =
NetworkClient.BootstrapConfiguration.enabled(
Review Comment:
I see. will revert the 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3520934561
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,137 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Check if a pending resolution completed before checking the
timeout, so that a
+// result arriving at the same time as the deadline is not incorrectly
rejected.
+if (pendingBootstrapResolution != null &&
pendingBootstrapResolution.isDone()) {
Review Comment:
We could consolidate the condition into `processBootstrapResolutionResult`.
For example:
```java
/**
* Checks if there is a completed bootstrap DNS resolution and processes
the result.
* @return true if the client is now bootstrapped and the calling method
should early return.
*/
private boolean maybeProcessBootstrapResolutionResult(final long
currentTimeMs) {
if (pendingBootstrapResolution == null ||
!pendingBootstrapResolution.isDone())
return false;
List servers = List.of();
try {
servers = pendingBootstrapResolution.getNow(List.of());
} catch (CompletionException e) {
log.debug("DNS resolution failed", e);
}
pendingBootstrapResolution = null;
if (!servers.isEmpty()) {
log.debug("Bootstrap DNS resolution succeeded, {} servers
resolved", servers.size());
metadataUpdater.bootstrap(servers);
return true;
}
log.debug("Failed to resolve bootstrap servers, will retry after
{}ms. Remaining time: {}ms",
bootstrapConfiguration.retryBackoffMs,
bootstrapTimer.remainingMs());
bootstrapResolutionRetryMs = currentTimeMs +
bootstrapConfiguration.retryBackoffMs;
return false;
}
```
##
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3501084974
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,139 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Check if a pending resolution completed before checking the
timeout, so that a
+// result arriving at the same time as the deadline is not incorrectly
rejected.
+if (pendingBootstrapResolution != null &&
pendingBootstrapResolution.isDone()) {
+processBootstrapResolutionResult(currentTimeMs);
+if (metadataUpdater.isBootstrapped())
+return;
+}
+
+if (bootstrapTimer != null) {
+bootstrapTimer.update(currentTimeMs);
+checkBootstrapTimeout();
+}
+maybeStartBootstrapResolution(currentTimeMs);
+}
+
+/**
+ * Record a permanent bootstrap failure on the metadata if the timeout has
expired.
+ * The error is not thrown here; callers observe it through their metadata
layer
+ * (e.g. {@link Metadata#maybeThrowFatalException()} for
Producer/Consumer, or
+ * {@code AdminMetadataManager#bootstrapFatalException()} for AdminClient).
+ */
+private void checkBootstrapTimeout() {
+if (bootstrapTimer.isExpired() && bootstrapException == null) {
+cancelBootstrapResolution();
+bootstrapException = new BootstrapResolutionException("Failed to
resolve bootstrap servers after " +
+bootstrapConfiguration.bootstrapResolveTimeoutMs + "ms. " +
+"Please check your bootstrap.servers configuration and DNS
settings.");
+metadataUpdater.bootstrapFailed(bootstrapException);
Review Comment:
Make sense
--
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:
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3500538515
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1280,139 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+for (String url : bootstrapServers) {
+if (Utils.getHost(url) == null || Utils.getPort(url) == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
+}
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (bootstrapException != null)
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Check if a pending resolution completed before checking the
timeout, so that a
+// result arriving at the same time as the deadline is not incorrectly
rejected.
+if (pendingBootstrapResolution != null &&
pendingBootstrapResolution.isDone()) {
+processBootstrapResolutionResult(currentTimeMs);
+if (metadataUpdater.isBootstrapped())
+return;
+}
+
+if (bootstrapTimer != null) {
+bootstrapTimer.update(currentTimeMs);
+checkBootstrapTimeout();
+}
+maybeStartBootstrapResolution(currentTimeMs);
+}
+
+/**
+ * Record a permanent bootstrap failure on the metadata if the timeout has
expired.
+ * The error is not thrown here; callers observe it through their metadata
layer
+ * (e.g. {@link Metadata#maybeThrowFatalException()} for
Producer/Consumer, or
+ * {@code AdminMetadataManager#bootstrapFatalException()} for AdminClient).
+ */
+private void checkBootstrapTimeout() {
+if (bootstrapTimer.isExpired() && bootstrapException == null) {
+cancelBootstrapResolution();
+bootstrapException = new BootstrapResolutionException("Failed to
resolve bootstrap servers after " +
+bootstrapConfiguration.bootstrapResolveTimeoutMs + "ms. " +
+"Please check your bootstrap.servers configuration and DNS
settings.");
+metadataUpdater.bootstrapFailed(bootstrapException);
Review Comment:
since the throw was removed from here, this will now continue execution and
run the `maybeStartBootstrapResolution` that was not reached before. Will that
start a new async resolution after the failure?
Taking a step back,
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3499736486
##
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##
@@ -1544,7 +1591,13 @@ private void processRequests() {
// Wait for network responses.
log.trace("Entering KafkaClient#poll(timeout={})",
pollTimeout);
-List responses = client.poll(Math.max(0L,
pollTimeout), now);
+List responses;
+try {
+responses = client.poll(Math.max(0L, pollTimeout), now);
+} catch (BootstrapResolutionException e) {
+failAllPendingCalls(e, now);
+break;
Review Comment:
In the latest commit, `AdminMetadataManager` now persists a
`bootstrapFatalException` (set by `AdminMetadataUpdater.bootstrapFailed`), and
`AdminClientRunnable.enqueue` checks it before queueing, failing new calls
immediately with `BootstrapResolutionException`. So even though the I/O thread
still exits after break, every subsequent API call surfaces the bootstrap error
— covered by the second `admin.listTopics().names().get()` in
`testAdminBootstrapResolutionExceptionPropagated`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3499492050
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -53,6 +53,74 @@ public final class ClientUtils {
private ClientUtils() {
}
+/**
+ * Resolves a single URL to one or more InetSocketAddress based on the DNS
lookup strategy.
+ *
+ * @param url the original URL string (for logging)
+ * @param host the hostname extracted from the URL
+ * @param port the port extracted from the URL
+ * @param clientDnsLookup the DNS lookup strategy
+ * @return list of resolved addresses (may be empty if addresses are
unresolved)
+ * @throws UnknownHostException if DNS resolution fails
+ */
+private static List resolveAddress(
+String url,
+String host,
+Integer port,
+ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+List addresses = new ArrayList<>();
+
+if (clientDnsLookup ==
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+InetAddress[] inetAddresses = InetAddress.getAllByName(host);
+for (InetAddress inetAddress : inetAddresses) {
+String resolvedCanonicalName =
inetAddress.getCanonicalHostName();
+InetSocketAddress address = new
InetSocketAddress(resolvedCanonicalName, port);
+if (address.isUnresolved()) {
+log.warn("Couldn't resolve server {} from {} as DNS
resolution of the canonical hostname {} failed for {}",
+url, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
resolvedCanonicalName, host);
+} else {
+addresses.add(address);
+}
+}
+} else {
+InetSocketAddress address = new InetSocketAddress(host, port);
+if (address.isUnresolved()) {
+log.warn("Couldn't resolve server {} from {} as DNS resolution
failed for {}",
+url, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
host);
+} else {
+addresses.add(address);
+}
+}
+
+return addresses;
+}
+
+public static List parseAddresses(List urls,
ClientDnsLookup clientDnsLookup) {
+List addresses = new ArrayList<>();
+if (urls == null) {
+return addresses;
+}
+for (String url : urls) {
+if (Thread.currentThread().isInterrupted()) {
+break;
+}
+
+final String host = getHost(url);
+final Integer port = getPort(url);
+
+if (host == null || port == null)
+throw new ConfigException("Invalid url in " +
CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG + ": " + url);
Review Comment:
Ah, I miss 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3462956747
##
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##
@@ -4316,6 +4319,46 @@ public void
testConstructorFailsOnNetworkClientConstructorFailure(GroupProtocol
assertEquals("No LoginModule found for
org.example.InvalidLoginModule", cause.getMessage());
}
+@ParameterizedTest
+@EnumSource(value = GroupProtocol.class)
+public void
testConsumerBootstrapResolutionExceptionPropagatedToPoll(GroupProtocol
protocol) throws InterruptedException {
+// Use an invalid hostname that will fail DNS resolution (using RFC
6761 reserved .invalid TLD)
+String invalidHost = "unresolvable.invalid:9092";
+
+Map configs = Map.of(
+ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName(),
+ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName(),
+CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, invalidHost,
+// Set a short bootstrap timeout so the test doesn't take too long
+CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG, "3000",
+ConsumerConfig.GROUP_PROTOCOL_CONFIG, protocol.name(),
+ConsumerConfig.GROUP_ID_CONFIG, "test-group"
+);
+
+try (KafkaConsumer consumer = new
KafkaConsumer<>(configs)) {
+// Subscribe to a topic to trigger metadata fetch
+consumer.subscribe(Set.of("test-topic"));
+
+// Poll continuously until we get the BootstrapResolutionException
+// The exception should be thrown after
bootstrap.resolve.timeout.ms expires
+BootstrapResolutionException exception =
+assertThrows(BootstrapResolutionException.class, () -> {
Review Comment:
same here, let's assert we get the error in other API calls after this first
one that detects it
##
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##
@@ -3416,4 +3424,26 @@ public static void resetCounters() {
CLOSE_COUNT.set(0);
}
}
+
+@Test
+public void testProducerBootstrapResolutionExceptionPropagated() {
Review Comment:
let's add another call to the producer API after the assertThrows, that will
ensure that all calls throw, not only the first one after the error (this is
what I expect should fail if we're not persisting the BootstrapException in the
metadata layer, comments above)
##
clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java:
##
@@ -309,6 +309,14 @@ public void run() {
*
*/
void runOnce() {
+try {
+runOnceInternal();
+} catch (BootstrapResolutionException e) {
+metadata.fatalErrorAndNotify(e);
Review Comment:
with this we end up reusing the existing fatalError for the Bootstrap
exception, but that one gets cleared, which is not what we want (to ensure that
once Bootstrap fails, all API calls get the error, not just the first call).
What about the idea from before of having a separate bootstrapFatalError in
the metadata? that wouldn't get cleared, and we should set it at the network
client level where we get it probably (to ensure that it applies to all
consumer, producer, admin equally)
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -53,6 +53,74 @@ public final class ClientUtils {
private ClientUtils() {
}
+/**
+ * Resolves a single URL to one or more InetSocketAddress based on the DNS
lookup strategy.
+ *
+ * @param url the original URL string (for logging)
+ * @param host the hostname extracted from the URL
+ * @param port the port extracted from the URL
+ * @param clientDnsLookup the DNS lookup strategy
+ * @return list of resolved addresses (may be empty if addresses are
unresolved)
+ * @throws UnknownHostException if DNS resolution fails
+ */
+private static List resolveAddress(
+String url,
+String host,
+Integer port,
+ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+List addresses = new ArrayList<>();
+
+if (clientDnsLookup ==
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+InetAddress[] inetAddresses = InetAddress.getAllByName(host);
+for (InetAddress inetAddress : inetAddresses) {
+String resolvedCanonicalName =
inetAddress.getCanonicalHostName();
+InetSocketAddress address = new
InetSocketAddress(resolvedCanonicalName, port);
+if (address.isUnresolved()) {
+log.warn("Couldn't resolve server {} from {} as DNS
resolution of the canonical hostname {} failed for {}",
+url,
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3422182381
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -629,6 +643,7 @@ private void doSend(ClientRequest clientRequest, boolean
isInternalRequest, long
@Override
public List poll(long timeout, long now) {
ensureActive();
+ensureBootstrapped(now);
Review Comment:
Make sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3397481639
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1277,128 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+if (bootstrapTimer != null) {
+bootstrapTimer.update(currentTimeMs);
+checkBootstrapTimeout();
Review Comment:
uhm we're checking expiration here before checking if the result is ready
(pendingBootstrapResolution.isDone in hidden is
`handleAsyncBootstrapResolution`), so even if we got to resolve DNS async we
could fail with timeout just because of this order, right?
If so, we can maybe consider splitting the "maybe trigger resolution" from
the "check result" (both together now in the handleAsyncBootstrapResolution),
so we can check the result before timing out?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3397306885
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -629,6 +643,7 @@ private void doSend(ClientRequest clientRequest, boolean
isInternalRequest, long
@Override
public List poll(long timeout, long now) {
ensureActive();
+ensureBootstrapped(now);
Review Comment:
> What happens if the resolution keeps failing? Would it lead to a busy-loop?
sorry to insist here, but I still don't get how we are addressing this?
From what I see, if the resolution keeps failing like @chia7712 mentioned
(e.g, bootstrap timed out), the background threads (sender and async consumer)
will continue to run in a busy loop just burning CPU (just throw here every
time and loop again, never make it to the selector.poll which is the blocking
call that would pace things down). Is that what happens?
If so, one option is to have this `ensureBootstrap` fail once only, and from
that moment on just keep the "bootstrapFatalError" at the metadata level?
- when the timer expires ensureBootstrap throws the BootstrapException (like
it does now) + sets the metadata.bootstrapFatalError (similar the
metadata.fatalError, with the main diff that we don't want to clear the
bootstrap one, it's permanent)
- following iterations of this background loop see a flag "bootstrapFailed"
or similar, so bypass/early-return the ensureBootstrap and make it to the
selector.poll -> no busy loop anymore
- all API calls still see the BootrsapException, retrieved from the
metadata.bootstrapFatalError
Thoughts? not sure if I'm missing something in the chain here
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1210,6 +1277,128 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+if (bootstrapTimer != null) {
+bootstrapTimer.update(currentTimeMs);
+checkBootstrapTimeout();
Review Comment:
uhm we're checking expiration here before checking if the result is ready
(pendingBootstrapResolution.isDone in hidden in
`handleAsyncBootstrapResolution`), so even if we got to resolve DNS async we
could fail with timeout just because of this order, right?
If so, we can maybe
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080: URL: https://github.com/apache/kafka/pull/21080#discussion_r3396658965 ## clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java: ## Review Comment: this is where we can be throwing the new BootstrapExc too I expect (KIP-1319 got in and made this API metadata-aware). Could we add a test like we did for send/partitionsFor (to ensure we do surface the error), and update the java doc and KIP too? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3381552415
##
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##
@@ -426,6 +431,12 @@ public class ConsumerConfig extends AbstractConfig {
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
Importance.MEDIUM,
CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
+.define(BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG,
+Type.LONG,
+
CommonClientConfigs.DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS,
+atLeast(0L),
+Importance.MEDIUM,
Review Comment:
misconfig here, I will correct 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3375185604
##
clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java:
##
@@ -1056,6 +1055,7 @@ public Future send(ProducerRecord
record) {
*when send is invoked after producer has
been closed.
* @throws InterruptException If the thread is interrupted while
blocked
* @throws SerializationException If the key or value are not valid
objects given the configured serializers
+ * @throws BootstrapResolutionException if DNS resolution of the bootstrap
servers fails within {@code bootstrap.resolve.timeout.ms}
Review Comment:
the `sendOffsetsToTx` API is getting metadata-aware with KIP-1319 btw (this
PR https://github.com/apache/kafka/pull/22460). That means that it could start
throwing this BootstrapException once that PR and this KIP-909 PR go in I
expect, right?
We can update in either PR I expect, whichever goes in last so we have all
the parts (update just to mention the new exception in the java doc + update
the KIP). I'm reviewing both so it's covered, just sharing for awareness for
now. cc. @dajac
##
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##
@@ -426,6 +431,12 @@ public class ConsumerConfig extends AbstractConfig {
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
Importance.MEDIUM,
CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
+.define(BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG,
+Type.LONG,
+
CommonClientConfigs.DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS,
+atLeast(0L),
+Importance.MEDIUM,
Review Comment:
the KIP says HIGH for the config, what's the intention?
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##
@@ -157,13 +158,22 @@ public void poll(final long timeoutMs, final long
currentTimeMs) {
* @param onClose True when the network thread is closing.
*/
public void poll(final long timeoutMs, final long currentTimeMs, boolean
onClose) {
-trySend(currentTimeMs);
+try {
+trySend(currentTimeMs);
+} catch (BootstrapResolutionException e) {
+// Bootstrap DNS resolution timeout - propagate to app thread
+backgroundEventHandler.add(new ErrorEvent(e));
Review Comment:
in the case where the boostrap timed out, and mainly thinking about the
async consumer, wouldn't we end up flooding the queue here adding ErrorEvents
in the background?
Wonder if we should propagate this via `metadata.fatalError` instead. The
async consumer was initially propagating the metadata errors with events like
this but changed to avoid this flooding issue and to ensure we could cancel
in-flight futures in the background before failing in the app thread API call
(IIRC)
This
https://github.com/apache/kafka/blob/5111ad253a7ac12061c73ce4ab62a19ca7dc3d4e/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java#L172
and maybeFailOnMetadataError on the ConsumerNetworkThread. Makes sense?
--update
for the ShareConsumer follow-up we need to think more about this, that one
uses events IIRC, so we need to think about the flooding there
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3363838119
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -629,6 +643,7 @@ private void doSend(ClientRequest clientRequest, boolean
isInternalRequest, long
@Override
public List poll(long timeout, long now) {
ensureActive();
+ensureBootstrapped(now);
Review Comment:
how did we address this? (got to the same question after the recent fixes to
propagate the BootstrapException to the app thread, what happens here in the
background network poll loop?
##
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##
@@ -3262,4 +3263,26 @@ public static void resetCounters() {
CLOSE_COUNT.set(0);
}
}
+
+@Test
+public void testProducerBootstrapResolutionExceptionPropagated() {
+String invalidHost = "unresolvable.invalid:9092";
+Map configs = Map.of(
+ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName(),
+CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, invalidHost,
+CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG, "3000"
+);
+
+try (KafkaProducer producer = new
KafkaProducer<>(configs)) {
+assertThrows(BootstrapResolutionException.class, () -> {
+long startTime = System.currentTimeMillis();
+long maxWaitTime = 15000;
+while (System.currentTimeMillis() - startTime < maxWaitTime) {
+producer.partitionsFor("test-topic");
Review Comment:
could you update the java doc with the new exception? same for send()
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3358944813
##
tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java:
##
@@ -664,14 +665,19 @@ private static class ReplicaFetcherBlockingSend {
metrics,
time,
"replica-fetcher",
-new HashMap() {{
+new HashMap<>() {{
put("broker-id", sourceNode.idString());
put("fetcher-id", String.valueOf(fetcherId));
}},
false,
channelBuilder,
logContext
);
+NetworkClient.BootstrapConfiguration bootstrapConfiguration =
NetworkClient.BootstrapConfiguration.enabled(
Review Comment:
good catch!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3358903352
##
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java:
##
@@ -148,6 +149,17 @@ public void rebootstrap(long now) {
AdminMetadataManager.this.rebootstrap(now);
}
+@Override
+public boolean isBootstrapped() {
+return bootstrapCluster != null;
+}
+
+@Override
+public void bootstrap(List addresses) {
+// Called by NetworkClient.ensureBootstrapped() after DNS
resolution succeeds
+AdminMetadataManager.this.update(Cluster.bootstrap(addresses), 0);
Review Comment:
Yes. In `update()`, the now parameter is only used in the else branch to
update `lastMetadataUpdateMs`, but `Cluster.bootstrap(addresses)` produces a
cluster where `isBootstrapConfigured()` returns true, so we always take the if
branch and now is never 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3358152911
##
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##
@@ -426,6 +431,11 @@ public class ConsumerConfig extends AbstractConfig {
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY.toString()),
Importance.MEDIUM,
CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
+.define(BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG,
Review Comment:
should we validate "atLeast(0L)," (like in producer and admin) also they
don't have the same "importance", should we align?
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java:
##
@@ -157,7 +158,12 @@ public void poll(final long timeoutMs, final long
currentTimeMs) {
* @param onClose True when the network thread is closing.
*/
public void poll(final long timeoutMs, final long currentTimeMs, boolean
onClose) {
-trySend(currentTimeMs);
+try {
+trySend(currentTimeMs);
+} catch (BootstrapResolutionException e) {
+// Bootstrap DNS resolution timeout - propagate to app thread
Review Comment:
this made me notice we don't seem to propagate it for the Producer? or am I
missing it?
We want producer.send/partitionsFor to throw the Bootstrap exception, but
it's the same challenge we solved here: the exception is on the network layer,
Sender thread for producer, we need to propagate to the producer thread.
I imagine the Sender.runOnce call to the client.poll needs to catch the
Bootstrap error and propagate it via the producer metadata fatalError? (so that
the Producer API calls throw it on "metadata.awaitUpdate")
##
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##
@@ -4197,6 +4200,46 @@ public void
testConstructorFailsOnNetworkClientConstructorFailure(GroupProtocol
assertEquals("No LoginModule found for
org.example.InvalidLoginModule", cause.getMessage());
}
+@ParameterizedTest
+@EnumSource(value = GroupProtocol.class)
+public void
testConsumerBootstrapResolutionExceptionPropagatedToPoll(GroupProtocol
protocol) throws InterruptedException {
Review Comment:
could we add similar test for the producer.send/partitionsFor? (to be sure
that we're propagating the bootstrap exception, related to comment above). Same
for admin btw, can't find tests to ensure we do throw Bootstrap at the API
level and not swallow it along the way)
##
trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressWorker.java:
##
@@ -164,6 +166,11 @@ public boolean tryConnect() {
try (Metrics metrics = new Metrics()) {
try (Selector selector = new
Selector(conf.getLong(AdminClientConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
metrics, Time.SYSTEM, "", channelBuilder, logContext))
{
+NetworkClient.BootstrapConfiguration
bootstrapConfiguration = NetworkClient.BootstrapConfiguration.enabled(
Review Comment:
ditto
##
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java:
##
@@ -148,6 +149,17 @@ public void rebootstrap(long now) {
AdminMetadataManager.this.rebootstrap(now);
}
+@Override
+public boolean isBootstrapped() {
+return bootstrapCluster != null;
+}
+
+@Override
+public void bootstrap(List addresses) {
+// Called by NetworkClient.ensureBootstrapped() after DNS
resolution succeeds
+AdminMetadataManager.this.update(Cluster.bootstrap(addresses), 0);
Review Comment:
is this "0" here intentional? this is supposed to be "now"
##
tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java:
##
@@ -664,14 +665,19 @@ private static class ReplicaFetcherBlockingSend {
metrics,
time,
"replica-fetcher",
-new HashMap() {{
+new HashMap<>() {{
put("broker-id", sourceNode.idString());
put("fetcher-id", String.valueOf(fetcherId));
}},
false,
channelBuilder,
logContext
);
+NetworkClient.BootstrapConfiguration bootstrapConfiguration =
NetworkClient.BootstrapConfiguration.enabled(
Review Comment:
do we need "enabled" here if this tool uses the ManualMetadataUpdater?
--
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,
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-4603604907 Thanks @frankvicky ! Could you take a look at the conflicts pls? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3340193316
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -736,6 +775,10 @@ private void ensureActive() {
public void close() {
state.compareAndSet(State.ACTIVE, State.CLOSING);
if (state.compareAndSet(State.CLOSING, State.CLOSED)) {
+cancelBootstrapResolution();
+if (bootstrapExecutor != null) {
+bootstrapExecutor.shutdown();
Review Comment:
make sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3335904308
##
clients/clients-integration-tests/src/test/java/org/apache/kafka/clients/consumer/PlaintextConsumerTest.java:
##
@@ -860,6 +860,9 @@ private void testListTopics(Map
consumerConfig) throws Exception
try (var consumer = cluster.consumer(consumerConfig)) {
// consumer some messages, and we can list the internal topic
__consumer_offsets
consumer.subscribe(List.of(topic1));
+// Poll multiple times to ensure DNS resolution and metadata are
fully loaded
+// With deferred DNS resolution (KIP-909), initial polls may be
needed to complete bootstrap
+consumer.poll(Duration.ofMillis(100));
Review Comment:
agree, to be fair it's not all on the metadata. The test is expecting 4
topics so the whole join group + creation of the offsets topic needs to happen
I expect.
To make this solid and avoid flakiness, what about instead of 2 explicit
polls, we reuse the existing "waitForCondition", with cond "4 topics found" and
action consumer.poll
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -53,6 +53,72 @@ public final class ClientUtils {
private ClientUtils() {
}
+/**
+ * Resolves a single URL to one or more InetSocketAddress based on the DNS
lookup strategy.
+ *
+ * @param url the original URL string (for logging)
+ * @param host the hostname extracted from the URL
+ * @param port the port extracted from the URL
+ * @param clientDnsLookup the DNS lookup strategy
+ * @return list of resolved addresses (may be empty if addresses are
unresolved)
+ * @throws UnknownHostException if DNS resolution fails
+ */
+private static List resolveAddress(
+String url,
+String host,
+Integer port,
+ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+List addresses = new ArrayList<>();
+
+if (clientDnsLookup ==
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+InetAddress[] inetAddresses = InetAddress.getAllByName(host);
+for (InetAddress inetAddress : inetAddresses) {
+String resolvedCanonicalName =
inetAddress.getCanonicalHostName();
+InetSocketAddress address = new
InetSocketAddress(resolvedCanonicalName, port);
+if (address.isUnresolved()) {
+log.warn("Couldn't resolve server {} from {} as DNS
resolution of the canonical hostname {} failed for {}",
+url, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
resolvedCanonicalName, host);
+} else {
+addresses.add(address);
+}
+}
+} else {
+InetSocketAddress address = new InetSocketAddress(host, port);
+if (address.isUnresolved()) {
+log.warn("Couldn't resolve server {} from {} as DNS resolution
failed for {}",
+url, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
host);
+} else {
+addresses.add(address);
+}
+}
+
+return addresses;
+}
+
+public static List parseAddresses(List urls,
ClientDnsLookup clientDnsLookup) {
+List addresses = new ArrayList<>();
+if (urls == null) {
+return addresses;
+}
+urls.forEach(url -> {
Review Comment:
related to the comment above, what if the NetworkClient closed
(executor.shutdown)? Should we handle it? (not sure if the interrupt will be
detected from within the resolveAddresses? but we probably should just early
return if the task/thread gets interrupted while we're parsing addresses
(network client close)
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -736,6 +775,10 @@ private void ensureActive() {
public void close() {
state.compareAndSet(State.ACTIVE, State.CLOSING);
if (state.compareAndSet(State.CLOSING, State.CLOSED)) {
+cancelBootstrapResolution();
+if (bootstrapExecutor != null) {
+bootstrapExecutor.shutdown();
Review Comment:
shouldn't we `shutdownNow` here? we cancelled the future already in the line
above, so we don't really want/need any running task to complete right?
Then, the other end to check would be that if this gets shutdown here, it
could happen in the middle of a `parseAddresses`.
##
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##
Review Comment:
nice! you got 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: [email protected]
For queries about
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080: URL: https://github.com/apache/kafka/pull/21080#discussion_r3332843397 ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## Review Comment: oh, good catch. I think it might also solve the failing test in RestoreIntegration test. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3311264270
##
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##
@@ -135,10 +135,19 @@ public class RestoreIntegrationTest {
@BeforeAll
public static void startCluster() throws IOException {
CLUSTER.start();
-
+
final Properties adminConfig = new Properties();
adminConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
admin = Admin.create(adminConfig);
+
+// Trigger initial bootstrap to ensure AdminClient thread doesn't exit
prematurely
+// With KIP-909 deferred DNS resolution, AdminClient thread may exit
before first use
Review Comment:
uhm ok, but this is still a workaround that I don't get, why do we need it?
(just concerned we may have something broken/missing in the admin client?)
##
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##
Review Comment:
don't we need here the same check we have in the NetworkClient to
rebootstrap only if we bootstrapped already?
https://github.com/frankvicky/kafka/blob/82752bb013060e3b96648b86fd5b5bacdd502f0e/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java#L1422
(relates to the workaround we're needing in the admin tests maybe?? I expect
this should be consistent for both clients, we don't allow it on the network
client but seems we allow it here?)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3246995608
##
streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java:
##
@@ -135,10 +135,19 @@ public class RestoreIntegrationTest {
@BeforeAll
public static void startCluster() throws IOException {
CLUSTER.start();
-
+
final Properties adminConfig = new Properties();
adminConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
CLUSTER.bootstrapServers());
admin = Admin.create(adminConfig);
+
+// Trigger initial bootstrap to ensure AdminClient thread doesn't exit
prematurely
+// With KIP-909 deferred DNS resolution, AdminClient thread may exit
before first use
Review Comment:
I made the admin as non-static in the test to avoid this workaround.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3212666275
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -201,6 +252,14 @@ public static NetworkClient
createNetworkClient(AbstractConfig config,
metricsGroupPrefix,
channelBuilder,
logContext);
+ClientDnsLookup dnsLookup =
ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG));
+
+NetworkClient.BootstrapConfiguration bootstrapConfiguration =
NetworkClient.BootstrapConfiguration.enabled(
+bootstrapServers != null ? bootstrapServers : List.of(),
Review Comment:
Does it make sense to create a "enabled config" with `null` brokers?
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1235,123 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Lazy initialization: create timer on first poll to ensure timeout
starts when polling begins
+if (bootstrapTimer == null) {
+bootstrapTimer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
+}
+
+bootstrapTimer.update(currentTimeMs);
+checkBootstrapTimeout();
+handleAsyncBootstrapResolution();
+}
+
+/**
+ * Check if bootstrap timeout has expired and throw exception if so.
+ */
+private void checkBootstrapTimeout() {
+if (bootstrapTimer.isExpired()) {
+cancelBootstrapResolution();
+throw new BootstrapResolutionException("Failed to resolve
bootstrap servers after " +
+bootstrapConfiguration.bootstrapResolveTimeoutMs + "ms. " +
+"Please check your bootstrap.servers configuration and DNS
settings.");
+}
+}
+
+/**
+ * Handle the async DNS resolution state machine.
+ * States: Not Started -> In Progress -> Completed (Success/Failure)
+ */
+private void handleAsyncBootstrapResolution() {
+if (pendingBootstrapResolution == null) {
+pendingBootstrapResolution = CompletableFuture.supplyAsync(
+() -> ClientUtils.parseAddresses(
+bootstrapConfiguration.bootstrapServers,
+
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3173612746
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1235,123 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public static final BootstrapConfiguration DISABLED =
+new BootstrapConfiguration(List.of(), null, 0, 0);
+
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a truly
asynchronous approach
+ * to avoid blocking on DNS resolution.
+ *
+ * DNS resolution is performed on a separate thread via {@link
CompletableFuture}. This ensures
+ * the event loop remains responsive even if DNS lookups block or take a
long time. The bootstrap
+ * timeout can interrupt a pending DNS resolution, unlike synchronous
approaches.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration == BootstrapConfiguration.DISABLED ||
metadataUpdater.isBootstrapped())
+return;
+
+if (Thread.interrupted()) {
+cancelBootstrapResolution();
+throw new InterruptException(new InterruptedException());
+}
+
+// Lazy initialization: create timer on first poll to ensure timeout
starts when polling begins
+if (bootstrapTimer == null) {
+bootstrapTimer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
+}
+
+bootstrapTimer.update(currentTimeMs);
+checkBootstrapTimeout();
+handleAsyncBootstrapResolution();
+}
+
+/**
+ * Check if bootstrap timeout has expired and throw exception if so.
+ */
+private void checkBootstrapTimeout() {
+if (bootstrapTimer.isExpired()) {
+cancelBootstrapResolution();
+throw new BootstrapResolutionException("Failed to resolve
bootstrap servers after " +
+bootstrapConfiguration.bootstrapResolveTimeoutMs + "ms. " +
+"Please check your bootstrap.servers configuration and DNS
settings.");
+}
+}
+
+/**
+ * Handle the async DNS resolution state machine.
+ * States: Not Started -> In Progress -> Completed (Success/Failure)
+ */
+private void handleAsyncBootstrapResolution() {
+if (pendingBootstrapResolution == null) {
+pendingBootstrapResolution = CompletableFuture.supplyAsync(
+() -> ClientUtils.parseAddresses(
+bootstrapConfiguration.bootstrapServers,
+bootstrapConfiguration.clientDnsLookup
+),
+bootstrapExecutor
+);
+return;
+}
+
+// check again on next poll
+if (!pendingBootstrapResolution.isDone()) {
+return;
+}
+
+processBootstrapResolutionResult();
+}
+
+/**
+ * Process the completed bootstrap DNS resolution result.
+ */
+private void processBootstrapResolutionResult() {
+List servers = List.of();
+try {
+servers = pendingBootstrapResolution.getNow(List.of());
+} catch (CompletionException e) {
+log.debug("DNS resolution failed: {}", e.getCause().getMessage());
+}
+
+if
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3087722598
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1205,84 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+private final boolean isBootstrapDisabled;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs,
+ final boolean isBootstrapDisabled) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+this.isBootstrapDisabled = isBootstrapDisabled;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs, false);
+}
+
+public static BootstrapConfiguration disabled() {
+return new BootstrapConfiguration(List.of(), null, 0, 0, true);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a
non-blocking approach.
+ *
+ * On each invocation, this method will attempt DNS resolution once. If
resolution fails,
+ * it returns immediately and will be retried on the next poll invocation.
This ensures the
+ * event loop remains responsive and doesn't block on DNS lookups.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration.isBootstrapDisabled ||
metadataUpdater.isBootstrapped())
+return;
+
+if (Thread.interrupted()) {
+throw new InterruptException(new InterruptedException());
+}
+
+// Lazy initialization: create timer on first poll to ensure timeout
starts when polling begins
+if (bootstrapTimer == null) {
+bootstrapTimer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
+}
+
+bootstrapTimer.update(currentTimeMs);
+
+// Attempt DNS resolution (single attempt per poll, typically fast)
+List servers =
ClientUtils.parseAddresses(bootstrapConfiguration.bootstrapServers,
bootstrapConfiguration.clientDnsLookup);
Review Comment:
yeap, agree, I had similar thoughts on this
[comment](https://github.com/apache/kafka/pull/21080#issuecomment-4143194531) ,
leaning towards running it truly async first (then having second thoughts about
the potential impact?).
But agree that this should ideally be truly async to make sure we're
delivering the experience of not failing fast on DNS issues
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3087523309
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -821,6 +842,19 @@ public LeastLoadedNode leastLoadedNode(long now) {
}
}
+/**
+ * Handle the case when there are no nodes available.
+ * During bootstrap phase, return null node to allow DNS resolution to
continue.
+ * After bootstrap, throw IllegalStateException.
+ */
+private LeastLoadedNode handleEmptyNodeList() {
+if (!metadataUpdater.isBootstrapped()) {
Review Comment:
It is a bit odd that we use `if (!metadataUpdater.isBootstrapped())` to
swallow the error here. It makes more sense to rely on the configuration (e.g.,
checking if lazy bootstrap is disabled). If the feature is disabled, we should
preserve the original fail-fast behavior and throw the exception immediately.
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1205,84 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+private final boolean isBootstrapDisabled;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs,
+ final boolean isBootstrapDisabled) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+this.isBootstrapDisabled = isBootstrapDisabled;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs, false);
+}
+
+public static BootstrapConfiguration disabled() {
+return new BootstrapConfiguration(List.of(), null, 0, 0, true);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a
non-blocking approach.
+ *
+ * On each invocation, this method will attempt DNS resolution once. If
resolution fails,
+ * it returns immediately and will be retried on the next poll invocation.
This ensures the
+ * event loop remains responsive and doesn't block on DNS lookups.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration.isBootstrapDisabled ||
metadataUpdater.isBootstrapped())
Review Comment:
Have you considered using sentry pattern?
```java
public static class BootstrapConfiguration {
public final List bootstrapServers;
public final ClientDnsLookup clientDnsLookup;
public final long bootstrapResolveTimeoutMs;
public final long retryBackoffMs;
public static final BootstrapConfiguration DISABLED =
new BootstrapConfiguration(List.of(), null, 0, 0, true);
```
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -230,7 +241,8 @@ public NetworkClient(Selectable selector,
ApiVersions apiVersions,
Sensor throttleTimeSensor,
LogContext logContext,
- MetadataRecoveryStrategy metadataRecoveryStrategy) {
+ MetadataRecoveryStrategy metadataRecoveryStrategy,
+ BootstrapConfiguration bootstrapConfiguration) {
Review Comment:
Since this constructor is only used for testing and always receives a
disabled configuration, we can revert this signature change and just pass
BootstrapConfiguration.DISABLED internally
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080: URL: https://github.com/apache/kafka/pull/21080#discussion_r3078077915 ## connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java: ## @@ -106,6 +107,11 @@ public WorkerGroupMember(DistributedConfig config, metadata.bootstrap(addresses); Review Comment: I suppose it was a miss. I will update the KIP. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3072744468
##
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java:
##
@@ -1479,4 +1500,126 @@ public KafkaException getAndClearFailure() {
return failure;
}
}
+
+private void bootstrapMetadataWithNodes(Metadata metadata, List
nodes) {
+List serverAddresses = new ArrayList<>();
+nodes.forEach(node -> serverAddresses.add(new
InetSocketAddress(node.host(), node.port(;
+metadata.bootstrap(serverAddresses);
+}
+
+private void bootstrapMetadata(Metadata metadata) {
+List serverAddresses = new ArrayList<>(List.of(
+new InetSocketAddress("localhost0", 8000),
+new InetSocketAddress("localhost1", 8000)
+));
+metadata.bootstrap(serverAddresses);
+}
+
+private void bootstrapMetadataUpdater(final MetadataUpdater
metadataUpdater) {
+List serverAddresses = new ArrayList<>(List.of(
+new InetSocketAddress("localhost0", 8000),
+new InetSocketAddress("localhost1", 8000)
+));
+metadataUpdater.bootstrap(serverAddresses);
+System.out.println("Bootstraping metadata--:" +
metadataUpdater.isBootstrapped());
Review Comment:
leaking my debug skill..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3065291068
##
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java:
##
@@ -1479,4 +1500,126 @@ public KafkaException getAndClearFailure() {
return failure;
}
}
+
+private void bootstrapMetadataWithNodes(Metadata metadata, List
nodes) {
+List serverAddresses = new ArrayList<>();
+nodes.forEach(node -> serverAddresses.add(new
InetSocketAddress(node.host(), node.port(;
+metadata.bootstrap(serverAddresses);
+}
+
+private void bootstrapMetadata(Metadata metadata) {
+List serverAddresses = new ArrayList<>(List.of(
+new InetSocketAddress("localhost0", 8000),
+new InetSocketAddress("localhost1", 8000)
+));
+metadata.bootstrap(serverAddresses);
+}
+
+private void bootstrapMetadataUpdater(final MetadataUpdater
metadataUpdater) {
+List serverAddresses = new ArrayList<>(List.of(
+new InetSocketAddress("localhost0", 8000),
+new InetSocketAddress("localhost1", 8000)
+));
+metadataUpdater.bootstrap(serverAddresses);
+System.out.println("Bootstraping metadata--:" +
metadataUpdater.isBootstrapped());
Review Comment:
remove?
##
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java:
##
@@ -106,6 +107,11 @@ public WorkerGroupMember(DistributedConfig config,
metadata.bootstrap(addresses);
Review Comment:
uhm we're still doing the early boostrap here (so could fail if it can't
resolve DNS right?). Looking back to the KIP there are no references to
Connect, but wonder if that was a miss on the discussion and we should remove
this early resolution here? We would need to update the KIP if we agree this is
the intention.
It's a bit confusing because we're setting the new boostrapt configs right
below (but I guess they are useless because by the time we hit the new
bootstraop logic in the network poll the resolution already happened, here)
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1191,123 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+private boolean isBootstrapDisabled;
+
+public BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.isBootstrapDisabled = false;
+}
+
+public static BootstrapConfiguration disabled() {
+BootstrapConfiguration bootstrapConfiguration = new
BootstrapConfiguration(List.of(), null, 0);
+bootstrapConfiguration.disableBootstrap();
+return bootstrapConfiguration;
+}
+
+public void disableBootstrap() {
+this.isBootstrapDisabled = true;
+}
+}
+
+private class BootstrapState {
+private final Timer timer;
Review Comment:
this seems to be a gap still , was it addressed?
Lazy init of the timer on the first poll could be one option (or reset the
deadline only on the first poll, but this feels riskier imo)
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -840,6 +841,19 @@ public LeastLoadedNode leastLoadedNode(long now) {
}
}
+/**
+ * Handle the case when there are no nodes available.
+ * During bootstrap phase, return null node to allow DNS resolution to
continue.
+ * After bootstrap, throw IllegalStateException.
+ */
+private LeastLoadedNode handleEmptyNodeList() {
+if (!metadataUpdater.isBootstrapped()) {
+log.debug("No nodes available yet, still in bootstrap phase");
+return new LeastLoadedNode(null, false);
+}
Review Comment:
interesting, was this revealed with the consumer poll test? nice!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3046157937
##
clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java:
##
@@ -4134,6 +4135,45 @@ public void
testConstructorFailsOnNetworkClientConstructorFailure(GroupProtocol
assertEquals("No LoginModule found for
org.example.InvalidLoginModule", cause.getMessage());
}
+@Test
+public void
testAsyncConsumerBootstrapResolutionExceptionPropagatedToPoll() throws
InterruptedException {
Review Comment:
this behaviour applies equally to both consumers. Should we better
parameterize the test to run it for group.protocol=classic too?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3002841918
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1190,80 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+private final boolean isBootstrapDisabled;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs,
+ final boolean isBootstrapDisabled) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+this.isBootstrapDisabled = isBootstrapDisabled;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs, false);
+}
+
+public static BootstrapConfiguration disabled() {
+return new BootstrapConfiguration(List.of(), null, 0, 0, true);
+}
+}
+
+/**
+ * Attempts to resolve bootstrap server addresses via DNS and create an
initial bootstrap cluster.
+ * This method is called from {@link #poll(long, long)} and uses a
non-blocking approach.
+ *
+ * On each invocation, this method will attempt DNS resolution once. If
resolution fails,
+ * it returns immediately and will be retried on the next poll invocation.
This ensures the
+ * event loop remains responsive and doesn't block on DNS lookups.
+ *
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
before DNS resolution succeeds
+ */
+void ensureBootstrapped(final long currentTimeMs) {
+if (bootstrapConfiguration.isBootstrapDisabled ||
metadataUpdater.isBootstrapped())
+return;
+
+if (Thread.interrupted()) {
+throw new InterruptException(new InterruptedException());
+}
+
+// Timer is already initialized in constructor, just update it with
current time
+bootstrapTimer.update(currentTimeMs);
+
+// Attempt DNS resolution (single attempt per poll, typically fast)
+List servers =
ClientUtils.parseAddresses(bootstrapConfiguration.bootstrapServers,
bootstrapConfiguration.clientDnsLookup);
+
+if (!servers.isEmpty()) {
+log.debug("Bootstrap DNS resolution succeeded, {} servers
resolved", servers.size());
+metadataUpdater.bootstrap(servers);
+return;
+}
+
+// DNS resolution failed
+if (bootstrapTimer.isExpired()) {
+throw new BootstrapResolutionException("Failed to resolve
bootstrap servers after " +
Review Comment:
from the KIP I get that we want this to propagate to the consumer.poll right?
If so, then we need to catch this exception and send an error event to the
app thread, otherwise this error will be swallowed in the background (this
https://github.com/apache/kafka/blob/73cb15e72c0809d624ffabdf76c51d4fa8c5a046/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkThread.java#L165)
One option I see is to catch this at the `NetworkClientDelegate` level,
where we have the `backgroundEventHandler`, and propagate the error there via
event (just like we do to propagate metadata errors)
Btw, we really need to have a test at the `KafkaConsumerTest` level covering
this behaviour. E.g, create consumer with invalid host, poll continuously, poll
should eventually throw the bootstrap exception (for both consumers). I expect
that wouldn't work now for the async.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r3002513262
##
clients/src/main/java/org/apache/kafka/common/Cluster.java:
##
@@ -217,6 +217,34 @@ public static Cluster bootstrap(List
addresses) {
Collections.emptySet(), Collections.emptySet(),
Collections.emptySet(), null, Collections.emptyMap());
}
+/**
+ * Create a cluster instance preserving the bootstrap state.
+ * This is used internally by MetadataSnapshot to maintain bootstrap flag
across metadata updates.
+ *
+ * @param clusterId the cluster id
+ * @param nodes the nodes in the cluster
+ * @param partitions information about partitions
+ * @param unauthorizedTopics unauthorized topics
+ * @param invalidTopics invalid topics
+ * @param internalTopics internal topics
+ * @param controller the controller node
+ * @param topicIds topic IDs
+ * @param isBootstrapConfigured whether the cluster is bootstrapped
+ * @return a new Cluster instance with the specified bootstrap state
+ */
+public static Cluster withBootstrapFlag(String clusterId,
Review Comment:
we had on offline discussion about this one. Another option to consider is
to completely remove the `isBootstrapConfigured` flag, and rely on checking if
metadata has nodes instead (having nodes, either bootstrap nodes or from real
metadata, means that we can skip DNS resolution, which is what the
isBootstrapConfigured was used for).
The changes are in place now, with that, no need for changes on this Cluster
class. Thoughts @chia7712 ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
lianetm commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-4143194531 About the blocking behaviour, agree with removing the while loop within the network client poll, and let the retry happen on the next poll. But even with that, we still have a blocking call to `parseAddresses`(single one, not in a while, but still blocking because it triggers `InetAddress.getAllByName`). I wonder if we could consider running this async? triggered only once when needed from the network poll (so the overhead of the extra thread is only to bootstrap), and with that we don't block on network poll and just conitnue to check if that asyn op finished in time, something like that. Thoughts? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r324458
##
clients/src/main/java/org/apache/kafka/common/Cluster.java:
##
@@ -217,6 +217,34 @@ public static Cluster bootstrap(List
addresses) {
Collections.emptySet(), Collections.emptySet(),
Collections.emptySet(), null, Collections.emptyMap());
}
+/**
+ * Create a cluster instance preserving the bootstrap state.
+ * This is used internally by MetadataSnapshot to maintain bootstrap flag
across metadata updates.
+ *
+ * @param clusterId the cluster id
+ * @param nodes the nodes in the cluster
+ * @param partitions information about partitions
+ * @param unauthorizedTopics unauthorized topics
+ * @param invalidTopics invalid topics
+ * @param internalTopics internal topics
+ * @param controller the controller node
+ * @param topicIds topic IDs
+ * @param isBootstrapConfigured whether the cluster is bootstrapped
+ * @return a new Cluster instance with the specified bootstrap state
+ */
+public static Cluster withBootstrapFlag(String clusterId,
Review Comment:
I understand your concern about adding a new public API. However, the
`Cluster.withBootstrapFlag()` method is necessary for the KIP-909
implementation. Here's why:
1. The isBootstrapConfigured flag must be preserved across metadata
updates:
- After DNS resolution succeeds, we create a cluster with
`isBootstrapConfigured=true`
- When a partial metadata update occurs (via
`MetadataSnapshot.mergeWith()`), this flag needs to be preserved
- This allows `AdminMetadataManager.update()` to correctly distinguish
between bootstrap cluster and real cluster metadata
2. Package structure constraints:
- Cluster is in the `org.apache.kafka.common package`
- `MetadataSnapshot` is in the `org.apache.kafka.clients` package
- We cannot use package-private methods to access each other
- The only options are: public static method or restructuring the package
layout
3. Issues with alternative approaches:
- Not preserving the flag: Would break `AdminMetadataManager`'s ability to
correctly identify bootstrap clusters, affecting re-bootstrap logic
- Modifying package structure: Too large a change with high risk
- Other designs: All require similar public API additions or more complex
refactoring
Alternative consideration:
If adding this public method is a blocker, the alternative would be to:
1. Move both Cluster and MetadataSnapshot to the same package, or
2. Add a ClusterBuilder class in the clients package that has access to
create clusters with specific bootstrap flags
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-4104437617 > did we discuss the blocking DNS resolution in poll before? I'm thinking we could use a separate BootstrapResolver class with an independent thread to handle this. We can pass it to Metadata or AdminMetadataManager so we don't have to change NetworkClient too much. NetworkClient#ensureBootstrapped can just poll the metadataUpdater's state with blocking. the while loop in `ensureBootstrapped` is problematic because it introduces an unexpected blocking logic into the event loop. The origin design likely didn't intend to have a loop here. To keep it non-blocking, it should just return immediately if the dns resolution is not ready for now -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2970091470
##
clients/src/main/java/org/apache/kafka/common/Cluster.java:
##
@@ -217,6 +217,34 @@ public static Cluster bootstrap(List
addresses) {
Collections.emptySet(), Collections.emptySet(),
Collections.emptySet(), null, Collections.emptyMap());
}
+/**
+ * Create a cluster instance preserving the bootstrap state.
+ * This is used internally by MetadataSnapshot to maintain bootstrap flag
across metadata updates.
+ *
+ * @param clusterId the cluster id
+ * @param nodes the nodes in the cluster
+ * @param partitions information about partitions
+ * @param unauthorizedTopics unauthorized topics
+ * @param invalidTopics invalid topics
+ * @param internalTopics internal topics
+ * @param controller the controller node
+ * @param topicIds topic IDs
+ * @param isBootstrapConfigured whether the cluster is bootstrapped
+ * @return a new Cluster instance with the specified bootstrap state
+ */
+public static Cluster withBootstrapFlag(String clusterId,
Review Comment:
This change seems to be unnecessary. Also, it is a public class so we should
not make this change here
##
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##
@@ -538,7 +538,14 @@ static KafkaAdminClient createInternal(
config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),
config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG),
adminAddresses.usingBootstrapControllers());
-
metadataManager.update(Cluster.bootstrap(adminAddresses.addresses()),
time.milliseconds());
Review Comment:
it seems `adminAddresses.addresses()` is not used any more. Maybe we could
just remove `AdminBootstrapAddresses` and use a helper method instead?
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1197,118 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+private final boolean isBootstrapDisabled;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs,
+ final boolean isBootstrapDisabled) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+this.isBootstrapDisabled = isBootstrapDisabled;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs, false);
+}
+
+public static BootstrapConfiguration disabled() {
+return new BootstrapConfiguration(List.of(), null, 0, 0, true);
+}
+}
+
+/**
+ * Ensures that the client has successfully resolved and bootstrapped with
the configured bootstrap servers.
+ * This method will retry DNS resolution until one of the following
conditions is met:
+ *
+ * DNS resolution succeeds and bootstrap is complete (method returns
normally)
+ * Bootstrap timeout expires (throws
BootstrapResolutionException)
+ * Poll timeout is reached but bootstrap timeout hasn't expired
(method returns to retry later)
+ *
+ *
+ * @param pollTimeoutMs The poll timeout in milliseconds, controlling how
long this method should block
+ * during a single poll invocation
+ * @param currentTimeMs The current time in milliseconds
+ * @throws BootstrapResolutionException if the bootstrap timeout expires
+ * before DNS resolution succeeds
+ *
+ * @implNote This method performs blocking DNS lookups via {@link
InetAddress#getAllByName(String)}.
+ * Each DNS lookup may block for an indefinite amount of time
depending on network conditions
+ * and DNS server responsiveness. As
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2960755692
##
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##
@@ -538,7 +539,22 @@ static KafkaAdminClient createInternal(
config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),
config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG),
adminAddresses.usingBootstrapControllers());
-
metadataManager.update(Cluster.bootstrap(adminAddresses.addresses()),
time.milliseconds());
+
+// Get the appropriate bootstrap configuration
+List bootstrapAddressesToUse =
adminAddresses.usingBootstrapControllers()
+?
config.getList(AdminClientConfig.BOOTSTRAP_CONTROLLERS_CONFIG)
+: config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
+
+// Create unresolved addresses for bootstrap cluster (defer DNS
resolution to NetworkClient.poll())
+List unresolvedAddresses = new ArrayList<>();
+for (String address : bootstrapAddressesToUse) {
+String host = Utils.getHost(address);
+Integer port = Utils.getPort(address);
+if (host != null && port != null) {
+
unresolvedAddresses.add(InetSocketAddress.createUnresolved(host, port));
+}
+}
+metadataManager.update(Cluster.bootstrap(unresolvedAddresses),
time.milliseconds());
Review Comment:
The cluster created by `Cluster.bootstrap` has `isBootstrapConfigured=true`,
so `bootstrapCluster` will be updated immediately. This makes
`isBootstrapped()` incorrectly return true
##
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##
@@ -538,7 +539,22 @@ static KafkaAdminClient createInternal(
config.getLong(AdminClientConfig.RETRY_BACKOFF_MS_CONFIG),
config.getLong(AdminClientConfig.METADATA_MAX_AGE_CONFIG),
adminAddresses.usingBootstrapControllers());
-
metadataManager.update(Cluster.bootstrap(adminAddresses.addresses()),
time.milliseconds());
+
+// Get the appropriate bootstrap configuration
+List bootstrapAddressesToUse =
adminAddresses.usingBootstrapControllers()
+?
config.getList(AdminClientConfig.BOOTSTRAP_CONTROLLERS_CONFIG)
+: config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
+
+// Create unresolved addresses for bootstrap cluster (defer DNS
resolution to NetworkClient.poll())
+List unresolvedAddresses = new ArrayList<>();
+for (String address : bootstrapAddressesToUse) {
+String host = Utils.getHost(address);
+Integer port = Utils.getPort(address);
+if (host != null && port != null) {
+
unresolvedAddresses.add(InetSocketAddress.createUnresolved(host, port));
+}
+}
+metadataManager.update(Cluster.bootstrap(unresolvedAddresses),
time.milliseconds());
Review Comment:
Maybe we should leverage `MetadataUpdater#bootstrap` to set the resolved
bootstrap servers once the DNS resolution succeeds in `NetworkClient`?
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1197,112 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+private final boolean isBootstrapDisabled;
+
+private BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs,
+ final long retryBackoffMs,
+ final boolean isBootstrapDisabled) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.retryBackoffMs = retryBackoffMs;
+this.isBootstrapDisabled = isBootstrapDisabled;
+}
+
+public static BootstrapConfiguration enabled(final List
bootstrapServers,
+ final ClientDnsLookup
clientDnsLookup,
+ final long
bootstrapResolveTimeoutMs,
+ final long
retryBackoffMs) {
+return new
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
Nikita-Shupletsov commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-4088067024 looks like the test failures may be related to the 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080: URL: https://github.com/apache/kafka/pull/21080#discussion_r2956324908 ## clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java: ## @@ -550,7 +550,14 @@ static KafkaAdminClient createInternal( MetricsContext metricsContext = new KafkaMetricsContext(JMX_PREFIX, config.originalsWithPrefix(CommonClientConfigs.METRICS_CONTEXT_PREFIX)); metrics = new Metrics(metricConfig, reporters, time, metricsContext); + +// Use the appropriate bootstrap configuration determined by AdminBootstrapAddresses Review Comment: It seems `AdminBootstrapAddresses.fromConfig(config);` will resolve the dns already, which is conflict with KIP-909 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2956271208
##
clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java:
##
@@ -81,6 +82,19 @@ public void handleSuccessfulResponse(RequestHeader
requestHeader, long now, Meta
// Do nothing
}
+@Override
+public boolean isBootstrapped() {
+// ManualMetadataUpdater is designed for cases where nodes are
manually set,
+// so we consider it bootstrapped if nodes have been provided
+return !nodes.isEmpty();
+}
+
+@Override
+public void bootstrap(List addresses) {
+// ManualMetadataUpdater doesn't use NetworkClient's bootstrap
mechanism
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2956271208
##
clients/src/main/java/org/apache/kafka/clients/ManualMetadataUpdater.java:
##
@@ -81,6 +82,19 @@ public void handleSuccessfulResponse(RequestHeader
requestHeader, long now, Meta
// Do nothing
}
+@Override
+public boolean isBootstrapped() {
+// ManualMetadataUpdater is designed for cases where nodes are
manually set,
+// so we consider it bootstrapped if nodes have been provided
+return !nodes.isEmpty();
+}
+
+@Override
+public void bootstrap(List addresses) {
+// ManualMetadataUpdater doesn't use NetworkClient's bootstrap
mechanism
Review Comment:
ditto
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2956270277
##
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminMetadataManager.java:
##
@@ -148,6 +149,18 @@ public void rebootstrap(long now) {
AdminMetadataManager.this.rebootstrap(now);
}
+@Override
+public boolean isBootstrapped() {
+// AdminClient bootstraps during construction via
metadataManager.update(Cluster.bootstrap(...))
+// so we check if the bootstrap cluster has been set
+return bootstrapCluster != null;
+}
+
+@Override
+public void bootstrap(List addresses) {
+// AdminClient handles bootstrap during construction, so this
method is not used
Review Comment:
KIP-909 states "Consumer, Producer, and Admin Clients: The bootstrap code
will be changed.", so this comment is not correct
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2956231905
##
tools/src/main/java/org/apache/kafka/tools/BrokerApiVersionsCommand.java:
##
@@ -182,6 +182,12 @@ static AdminClient create(AbstractConfig config) {
"admin",
ClientUtils.createChannelBuilder(config, time, logContext),
logContext);
+NetworkClient.BootstrapConfiguration bootstrapConfiguration = new
NetworkClient.BootstrapConfiguration(
Review Comment:
Could you please add a static method enabled to avoid passing false? This
would make it more readable and consistent.
```java
private BootstrapConfiguration(final List bootstrapServers,
final ClientDnsLookup clientDnsLookup,
final long bootstrapResolveTimeoutMs,
final long retryBackoffMs,
final boolean isBootstrapDisabled) {
this.bootstrapServers = bootstrapServers;
this.clientDnsLookup = clientDnsLookup;
this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
this.retryBackoffMs = retryBackoffMs;
this.isBootstrapDisabled = isBootstrapDisabled;
}
public static BootstrapConfiguration enabled(final List
bootstrapServers,
final ClientDnsLookup
clientDnsLookup,
final long
bootstrapResolveTimeoutMs,
final long
retryBackoffMs) {
return new BootstrapConfiguration(bootstrapServers,
clientDnsLookup, bootstrapResolveTimeoutMs, retryBackoffMs, true);
}
public static BootstrapConfiguration disabled() {
return new BootstrapConfiguration(List.of(), null, 0, 0, true);
}
```
##
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/WorkerGroupMember.java:
##
@@ -106,6 +107,11 @@ public WorkerGroupMember(DistributedConfig config,
metadata.bootstrap(addresses);
String metricGrpPrefix = "connect";
ChannelBuilder channelBuilder =
ClientUtils.createChannelBuilder(config, time, logContext);
+NetworkClient.BootstrapConfiguration bootstrapConfiguration = new
NetworkClient.BootstrapConfiguration(
+config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG),
+
ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG)),
+CommonClientConfigs.DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS,
Review Comment:
we should use `bootstrap.resolve.timeout.ms`, right?
##
tools/src/main/java/org/apache/kafka/tools/BrokerApiVersionsCommand.java:
##
@@ -182,6 +182,12 @@ static AdminClient create(AbstractConfig config) {
"admin",
ClientUtils.createChannelBuilder(config, time, logContext),
logContext);
+NetworkClient.BootstrapConfiguration bootstrapConfiguration = new
NetworkClient.BootstrapConfiguration(
+config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG),
+
ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG)),
+CommonClientConfigs.DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS,
Review Comment:
we should use `bootstrap.resolve.timeout.ms`, right?
##
trogdor/src/main/java/org/apache/kafka/trogdor/workload/ConnectionStressWorker.java:
##
@@ -164,6 +166,12 @@ public boolean tryConnect() {
try (Metrics metrics = new Metrics()) {
try (Selector selector = new
Selector(conf.getLong(AdminClientConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG),
metrics, Time.SYSTEM, "", channelBuilder, logContext))
{
+NetworkClient.BootstrapConfiguration
bootstrapConfiguration = new NetworkClient.BootstrapConfiguration(
+
conf.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG),
+
ClientDnsLookup.forConfig(conf.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG)),
+
CommonClientConfigs.DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS,
Review Comment:
we should use `bootstrap.resolve.timeout.ms`, right?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2954621012
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -228,6 +282,20 @@ public static NetworkClient
createNetworkClient(AbstractConfig config,
metricsGroupPrefix,
channelBuilder,
logContext);
+ClientDnsLookup dnsLookup =
ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG));
+
+// Validate bootstrap servers if provided (non-empty list)
+// This allows configurations that don't use bootstrap (e.g.,
broker-to-broker) to skip validation
+if (bootstrapServers != null && !bootstrapServers.isEmpty()) {
+parseAndValidateAddresses(bootstrapServers, dnsLookup);
+}
+
+bootstrapConfiguration = new NetworkClient.BootstrapConfiguration(
+bootstrapServers != null ? bootstrapServers : List.of(),
+dnsLookup,
+CommonClientConfigs.DEFAULT_BOOTSTRAP_RESOLVE_TIMEOUT_MS,
Review Comment:
Should we honour `bootstrap.resolve.timeout.ms` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2954580716
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -228,6 +282,20 @@ public static NetworkClient
createNetworkClient(AbstractConfig config,
metricsGroupPrefix,
channelBuilder,
logContext);
+ClientDnsLookup dnsLookup =
ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG));
+
+// Validate bootstrap servers if provided (non-empty list)
+// This allows configurations that don't use bootstrap (e.g.,
broker-to-broker) to skip validation
Review Comment:
Out of curiosity, why to resolve the dns for now? IIRC, KIP-909 tries to
handle the resolution error in the poll loop, right?
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1197,110 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+public final long retryBackoffMs;
+private boolean isBootstrapDisabled;
Review Comment:
`isBootstrapDisabled` could be a final field, since it is never modified
after initialization, right?
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -228,6 +282,20 @@ public static NetworkClient
createNetworkClient(AbstractConfig config,
metricsGroupPrefix,
channelBuilder,
logContext);
+ClientDnsLookup dnsLookup =
ClientDnsLookup.forConfig(config.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG));
+
+// Validate bootstrap servers if provided (non-empty list)
+// This allows configurations that don't use bootstrap (e.g.,
broker-to-broker) to skip validation
+if (bootstrapServers != null && !bootstrapServers.isEmpty()) {
+parseAndValidateAddresses(bootstrapServers, dnsLookup);
+}
+
+bootstrapConfiguration = new NetworkClient.BootstrapConfiguration(
Review Comment:
line#275 could be merged into this one.
##
clients/src/main/java/org/apache/kafka/clients/MetadataUpdater.java:
##
@@ -102,6 +103,19 @@ default boolean needsRebootstrap(long now, long
rebootstrapTriggerMs) {
*/
default void rebootstrap(long now) {}
+/**
+ * Returns true if the metadata has been bootstrapped.
+ */
+default boolean isBootstrapped() {
Review Comment:
It seems all subclasses have implemented this method. We could remove the
default implementation for now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
FrankYang0529 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2945865967
##
tools/src/main/java/org/apache/kafka/tools/ReplicaVerificationTool.java:
##
@@ -664,14 +665,18 @@ private static class ReplicaFetcherBlockingSend {
metrics,
time,
"replica-fetcher",
-new HashMap() {{
+new HashMap<>() {{
put("broker-id", sourceNode.idString());
put("fetcher-id", String.valueOf(fetcherId));
}},
false,
channelBuilder,
logContext
);
+NetworkClient.BootstrapConfiguration bootstrapConfiguration = new
NetworkClient.BootstrapConfiguration(
+
consumerConfig.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG),
+
ClientDnsLookup.forConfig(consumerConfig.getString(CommonClientConfigs.CLIENT_DNS_LOOKUP_CONFIG)),
+
consumerConfig.getLong(CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG));
Review Comment:
It looks like `BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG` is defined but not used
in this PR. Do you want to change `RECONNECT_BACKOFF_MS_CONFIG` to
`BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG` here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
FrankYang0529 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2945833633
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1195,6 +1327,7 @@ class DefaultMetadataUpdater implements MetadataUpdater {
@Override
public List fetchNodes() {
+ensureBootstrapped(Long.MAX_VALUE, time.milliseconds());
Review Comment:
The `pollTimeoutMs` + `currentTimeMs` will be negative for this 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
FrankYang0529 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2945775516
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -53,6 +53,67 @@ public final class ClientUtils {
private ClientUtils() {
}
+/**
+ * Resolves a single URL to one or more InetSocketAddress based on the DNS
lookup strategy.
+ *
+ * @param url the original URL string (for logging)
+ * @param host the hostname extracted from the URL
+ * @param port the port extracted from the URL
+ * @param clientDnsLookup the DNS lookup strategy
+ * @return list of resolved addresses (may be empty if addresses are
unresolved)
+ * @throws UnknownHostException if DNS resolution fails
+ */
+private static List resolveAddress(
+String url,
+String host,
+Integer port,
+ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+List addresses = new ArrayList<>();
+
+if (clientDnsLookup ==
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+InetAddress[] inetAddresses = InetAddress.getAllByName(host);
+for (InetAddress inetAddress : inetAddresses) {
+String resolvedCanonicalName =
inetAddress.getCanonicalHostName();
+InetSocketAddress address = new
InetSocketAddress(resolvedCanonicalName, port);
+if (address.isUnresolved()) {
+log.warn("Couldn't resolve server {} from {} as DNS
resolution of the canonical hostname {} failed for {}",
+url, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
resolvedCanonicalName, host);
+} else {
+addresses.add(address);
+}
+}
+} else {
+InetSocketAddress address = new InetSocketAddress(host, port);
+if (address.isUnresolved()) {
+log.warn("Couldn't resolve server {} from {} as DNS resolution
failed for {}",
+url, CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
host);
+} else {
+addresses.add(address);
+}
+}
+
+return addresses;
+}
+
+public static List validateAddresses(List urls,
ClientDnsLookup clientDnsLookup) {
+List addresses = new ArrayList<>();
+if (urls == null) {
+return addresses;
+}
+urls.forEach(url -> {
+final String host = getHost(url);
+final Integer port = getPort(url);
Review Comment:
The `getHost` and `getPort` may return null. Should we check 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-4031930280 @chia7712 @rajinisivaram thanks for the review. Comments addressed. PTAL 🙇🏼 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2902505693
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -53,6 +53,67 @@ public final class ClientUtils {
private ClientUtils() {
}
+/**
+ * Resolves a single URL to one or more InetSocketAddress based on the DNS
lookup strategy.
+ *
+ * @param url the original URL string (for logging)
+ * @param host the hostname extracted from the URL
+ * @param port the port extracted from the URL
+ * @param clientDnsLookup the DNS lookup strategy
+ * @return list of resolved addresses (may be empty if addresses are
unresolved)
+ * @throws UnknownHostException if DNS resolution fails
+ */
+private static List resolveAddress(
+String url,
+String host,
+Integer port,
+ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+List addresses = new ArrayList<>();
+
+if (clientDnsLookup ==
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+InetAddress[] inetAddresses = InetAddress.getAllByName(host);
Review Comment:
Updated:
https://cwiki.apache.org/confluence/display/KAFKA/KIP-909%3A+DNS+Resolution+Failure+Should+Not+Fail+the+Clients
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
rajinisivaram commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2890461322
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##
@@ -2193,7 +2188,6 @@ private void subscribeInternal(Collection topics,
Optionalhttps://cwiki.apache.org/confluence/display/KAFKA/Consumer+threading+refactor+design";>this
document
* for implementation detail.
- *
Review Comment:
nit: unnecessary formatting change
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -629,6 +643,7 @@ private void doSend(ClientRequest clientRequest, boolean
isInternalRequest, long
@Override
public List poll(long timeout, long now) {
ensureActive();
+ensureBootstrapped(now);
Review Comment:
The naming of `ensureBootstrapped` is odd. It looks like the method attempts
to resolve once. And if that fails and bootstrap timer hasn't expired, we
return here without bootstrapping and fall through. And that would just fail in
unexpected ways. We should probably retry resolution here up to the poll
timeout or bootstrap timeout, whichever expires first. And 1) return without an
error if bootstrap timer hasn't expired 2) throw BootstrapResolutionException
if bootstrap timer expired before poll timeout 3) Fall through only if
resolution has succeeded.
##
clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java:
##
@@ -1983,7 +1979,6 @@ private boolean isCommittedOffsetsManagementEnabled() {
* This method signals the background thread to {@link
CreateFetchRequestsEvent create fetch requests} for the
* pre-fetch case, i.e. right before {@link #poll(Duration)} exits. In the
pre-fetch case, the application thread
* will not wait for confirmation of the request creation before
continuing.
- *
Review Comment:
nit: unnecessary formatting changes in this file.
##
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java:
##
@@ -91,26 +91,31 @@
import static org.mockito.Mockito.when;
public class NetworkClientTest {
+protected static final MockTime TIME = new MockTime();
Review Comment:
Why was this changed to static? The old instance variable `time` seems to be
sufficient?
##
clients/src/main/java/org/apache/kafka/clients/KafkaClient.java:
##
@@ -212,5 +212,4 @@ ClientRequest newClientRequest(String nodeId,
* was invoked for this client.
*/
boolean active();
-
Review Comment:
nit: unnecessary formatting change in a file with no other changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2873859822
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -53,6 +53,67 @@ public final class ClientUtils {
private ClientUtils() {
}
+/**
+ * Resolves a single URL to one or more InetSocketAddress based on the DNS
lookup strategy.
+ *
+ * @param url the original URL string (for logging)
+ * @param host the hostname extracted from the URL
+ * @param port the port extracted from the URL
+ * @param clientDnsLookup the DNS lookup strategy
+ * @return list of resolved addresses (may be empty if addresses are
unresolved)
+ * @throws UnknownHostException if DNS resolution fails
+ */
+private static List resolveAddress(
+String url,
+String host,
+Integer port,
+ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+List addresses = new ArrayList<>();
+
+if (clientDnsLookup ==
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+InetAddress[] inetAddresses = InetAddress.getAllByName(host);
Review Comment:
If not, we should update the KIP as well
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
chia7712 commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2873784701
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1186,87 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+private boolean isBootstrapDisabled;
+
+public BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.isBootstrapDisabled = false;
+}
+
+public static BootstrapConfiguration disabled() {
+BootstrapConfiguration bootstrapConfiguration = new
BootstrapConfiguration(List.of(), null, 0);
+bootstrapConfiguration.disableBootstrap();
+return bootstrapConfiguration;
+}
+
+public void disableBootstrap() {
+this.isBootstrapDisabled = true;
+}
+}
+
+private class BootstrapState {
+private final Timer timer;
+private final List bootstrapServers;
+private final ClientDnsLookup clientDnsLookup;
+private final long dnsResolutionTimeoutMs;
+private final boolean isDisabled;
+
+BootstrapState(BootstrapConfiguration bootstrapConfiguration) {
+this.dnsResolutionTimeoutMs =
bootstrapConfiguration.bootstrapResolveTimeoutMs;
+this.timer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
+this.bootstrapServers = bootstrapConfiguration.bootstrapServers;
+this.clientDnsLookup = bootstrapConfiguration.clientDnsLookup;
+this.isDisabled = bootstrapConfiguration.isBootstrapDisabled;
+}
+
+List tryResolveAddresses(final long currentTimeMs) {
+timer.update(currentTimeMs);
+List addresses =
ClientUtils.validateAddresses(bootstrapServers, clientDnsLookup);
+if (!addresses.isEmpty()) {
+timer.reset(dnsResolutionTimeoutMs);
+return addresses;
+}
+
+if (timer.isExpired()) {
+throw new BootstrapResolutionException("Timeout while
attempting to resolve bootstrap " +
+"servers. ");
+}
+return ClientUtils.validateAddresses(bootstrapServers,
clientDnsLookup);
Review Comment:
`tryResolveAddresses` and `ensureBootstrapped` are doing the similar thing,
so could you try to refactor them?
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1186,87 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public List bootstrapServers;
Review Comment:
We could add the final modifier to these fields, right?
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -53,6 +53,67 @@ public final class ClientUtils {
private ClientUtils() {
}
+/**
+ * Resolves a single URL to one or more InetSocketAddress based on the DNS
lookup strategy.
+ *
+ * @param url the original URL string (for logging)
+ * @param host the hostname extracted from the URL
+ * @param port the port extracted from the URL
+ * @param clientDnsLookup the DNS lookup strategy
+ * @return list of resolved addresses (may be empty if addresses are
unresolved)
+ * @throws UnknownHostException if DNS resolution fails
+ */
+private static List resolveAddress(
+String url,
+String host,
+Integer port,
+ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+List addresses = new ArrayList<>();
+
+if (clientDnsLookup ==
ClientDnsLookup.RESOLVE_CANONICAL_BOOTSTRAP_SERVERS_ONLY) {
+InetAddress[] inetAddresses = InetAddress.getAllByName(host);
Review Comment:
`InetAddress.getAllByName` is a blocking call, which will cause `poll` loop
to block. Did the KIP mention this behavior?
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -629,6 +643,7 @@ private void doSend(ClientRequest clientRequest, boolean
isInternalRequest, long
@Override
public List poll(long timeout, long now) {
ensureActive();
+
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-3985625378 @kirktrue Could you please take a look when you have free 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
frankvicky commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2855389316
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1186,87 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+private boolean isBootstrapDisabled;
+
+public BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.isBootstrapDisabled = false;
+}
+
+public static BootstrapConfiguration disabled() {
+BootstrapConfiguration bootstrapConfiguration = new
BootstrapConfiguration(List.of(), null, 0);
+bootstrapConfiguration.disableBootstrap();
+return bootstrapConfiguration;
+}
+
+public void disableBootstrap() {
+this.isBootstrapDisabled = true;
+}
+}
+
+private class BootstrapState {
+private final Timer timer;
+private final List bootstrapServers;
+private final ClientDnsLookup clientDnsLookup;
+private final long dnsResolutionTimeoutMs;
+private final boolean isDisabled;
+
+BootstrapState(BootstrapConfiguration bootstrapConfiguration) {
+this.dnsResolutionTimeoutMs =
bootstrapConfiguration.bootstrapResolveTimeoutMs;
+this.timer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
+this.bootstrapServers = bootstrapConfiguration.bootstrapServers;
+this.clientDnsLookup = bootstrapConfiguration.clientDnsLookup;
+this.isDisabled = bootstrapConfiguration.isBootstrapDisabled;
+}
+
+List tryResolveAddresses(final long currentTimeMs) {
+timer.update(currentTimeMs);
+List addresses =
ClientUtils.validateAddresses(bootstrapServers, clientDnsLookup);
+if (!addresses.isEmpty()) {
+timer.reset(dnsResolutionTimeoutMs);
+return addresses;
+}
+
+if (timer.isExpired()) {
+throw new BootstrapResolutionException("Timeout while
attempting to resolve bootstrap " +
+"servers. ");
+}
+return ClientUtils.validateAddresses(bootstrapServers,
clientDnsLookup);
+}
+
+boolean isDisabled() {
+return isDisabled;
+}
+
+boolean isTimerExpired() {
+return timer.isExpired();
+}
+}
+
+void ensureBootstrapped(final long currentTimeMs) {
Review Comment:
yes, it is intend to call multiple times(on each `poll`)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
Nikita-Shupletsov commented on code in PR #21080:
URL: https://github.com/apache/kafka/pull/21080#discussion_r2818232114
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
@@ -53,6 +53,44 @@ public final class ClientUtils {
private ClientUtils() {
}
+public static List validateAddresses(List urls,
ClientDnsLookup clientDnsLookup) {
Review Comment:
as the old one is called parseAndValidateAddresses, this one is probably
closer to parseAddresses, as we omit the validation part, not the parsing part
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1171,6 +1186,87 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+private boolean isBootstrapDisabled;
+
+public BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.isBootstrapDisabled = false;
+}
+
+public static BootstrapConfiguration disabled() {
+BootstrapConfiguration bootstrapConfiguration = new
BootstrapConfiguration(List.of(), null, 0);
+bootstrapConfiguration.disableBootstrap();
+return bootstrapConfiguration;
+}
+
+public void disableBootstrap() {
+this.isBootstrapDisabled = true;
+}
+}
+
+private class BootstrapState {
+private final Timer timer;
+private final List bootstrapServers;
+private final ClientDnsLookup clientDnsLookup;
+private final long dnsResolutionTimeoutMs;
+private final boolean isDisabled;
+
+BootstrapState(BootstrapConfiguration bootstrapConfiguration) {
+this.dnsResolutionTimeoutMs =
bootstrapConfiguration.bootstrapResolveTimeoutMs;
+this.timer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
+this.bootstrapServers = bootstrapConfiguration.bootstrapServers;
+this.clientDnsLookup = bootstrapConfiguration.clientDnsLookup;
+this.isDisabled = bootstrapConfiguration.isBootstrapDisabled;
+}
+
+List tryResolveAddresses(final long currentTimeMs) {
+timer.update(currentTimeMs);
+List addresses =
ClientUtils.validateAddresses(bootstrapServers, clientDnsLookup);
+if (!addresses.isEmpty()) {
+timer.reset(dnsResolutionTimeoutMs);
+return addresses;
+}
+
+if (timer.isExpired()) {
+throw new BootstrapResolutionException("Timeout while
attempting to resolve bootstrap " +
+"servers. ");
+}
+return ClientUtils.validateAddresses(bootstrapServers,
clientDnsLookup);
+}
+
+boolean isDisabled() {
+return isDisabled;
+}
+
+boolean isTimerExpired() {
+return timer.isExpired();
+}
+}
+
+void ensureBootstrapped(final long currentTimeMs) {
Review Comment:
is this method intended to call multiple times? or is it supposed to block
till it bootstraps the metadata or the timer expires?
##
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##
Review Comment:
this and the new methods look alike. would it be possible to extract the
shared part and reuse it in both of them? to prevent the implementation drift
overtime
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
github-actions[bot] commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-3850915319 A label of 'needs-attention' was automatically added to this PR in order to raise the attention of the committers. Once this issue has been triaged, the `triage` label should be removed to prevent this automation from happening again. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
github-actions[bot] commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-3757969959 A label of 'needs-attention' was automatically added to this PR in order to raise the attention of the committers. Once this issue has been triaged, the `triage` label should be removed to prevent this automation from happening again. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
github-actions[bot] commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-3747540890 A label of 'needs-attention' was automatically added to this PR in order to raise the attention of the committers. Once this issue has been triaged, the `triage` label should be removed to prevent this automation from happening again. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
github-actions[bot] commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-3731770600 A label of 'needs-attention' was automatically added to this PR in order to raise the attention of the committers. Once this issue has been triaged, the `triage` label should be removed to prevent this automation from happening again. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648: Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
github-actions[bot] commented on PR #21080: URL: https://github.com/apache/kafka/pull/21080#issuecomment-3717195369 A label of 'needs-attention' was automatically added to this PR in order to raise the attention of the committers. Once this issue has been triaged, the `triage` label should be removed to prevent this automation from happening again. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648 - Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
philipnee commented on code in PR #16629: URL: https://github.com/apache/kafka/pull/16629#discussion_r1690598152 ## clients/src/main/java/org/apache/kafka/clients/ClientUtils.java: ## @@ -228,6 +229,10 @@ public static NetworkClient createNetworkClient(AbstractConfig config, metricsGroupPrefix, channelBuilder, logContext); +bootstrapConfig = new NetworkClient.BootstrapConfiguration( Review Comment: KIP-899 allows clients to rebootstrap. so it stores the bootstrap server addresses in the metadata. check `Metadata.java` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648 - Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
brenden20 commented on code in PR #16629: URL: https://github.com/apache/kafka/pull/16629#discussion_r1690532335 ## clients/src/main/java/org/apache/kafka/clients/ClientUtils.java: ## @@ -228,6 +229,10 @@ public static NetworkClient createNetworkClient(AbstractConfig config, metricsGroupPrefix, channelBuilder, logContext); +bootstrapConfig = new NetworkClient.BootstrapConfiguration( Review Comment: Could you elaborate on that? I am not sure what configs you are talking about, I am only seeing one. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
Re: [PR] KAFKA-14648 - Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
philipnee commented on code in PR #16629:
URL: https://github.com/apache/kafka/pull/16629#discussion_r1688567705
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1106,6 +1125,104 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+private final boolean bootstrapDisabled;
+
+public BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.bootstrapDisabled = false;
+}
+}
+
+private class BootstrapState {
+private final Timer timer;
+private final List bootstrapServers;
+private final ClientDnsLookup clientDnsLookup;
+private final long dnsResolutionTimeoutMs;
+private final boolean isDisabled;
+private boolean isBootstrapped;
+
+BootstrapState(BootstrapConfiguration bootstrapConfiguration) {
+if (bootstrapConfiguration == null) {
+this.dnsResolutionTimeoutMs = -1;
+this.timer = null;
+this.bootstrapServers = null;
+this.clientDnsLookup = null;
+this.isDisabled = true;
+this.isBootstrapped = false;
+} else {
+this.dnsResolutionTimeoutMs =
bootstrapConfiguration.bootstrapResolveTimeoutMs;
+this.timer =
time.timer(bootstrapConfiguration.bootstrapResolveTimeoutMs);
+this.bootstrapServers =
bootstrapConfiguration.bootstrapServers;
+this.clientDnsLookup = bootstrapConfiguration.clientDnsLookup;
+this.isDisabled = bootstrapConfiguration.bootstrapDisabled;
+}
+}
+
+List tryResolveAddresses() {
+timer.update(time.milliseconds());
+
+do {
Review Comment:
this will cause poll to block on this loop. Let's not do that. My initial
thought was doing the resolution once per poll(). DNS could take 10s of
minutes to get updated, and I think one use case was to let the application
continue to run, until bootstrap is completed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648 - Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
philipnee commented on code in PR #16629:
URL: https://github.com/apache/kafka/pull/16629#discussion_r1688562760
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -581,6 +593,13 @@ private void doSend(ClientRequest clientRequest, boolean
isInternalRequest, long
public List poll(long timeout, long now) {
ensureActive();
+if (!this.isBootstrapped() && !this.isBootstrapDisabled()) {
Review Comment:
let's move this to ensureBootstrapped().
I also wonder if we could make bootstrapState and optional, so we just need
to check bootstrapState.isEmpty() - true: don't need to bootstrap, false:
perform bootstrap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648 - Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
philipnee commented on code in PR #16629:
URL: https://github.com/apache/kafka/pull/16629#discussion_r1688563938
##
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##
@@ -1106,6 +1125,104 @@ private boolean isTelemetryApi(ApiKeys apiKey) {
return apiKey == ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS || apiKey ==
ApiKeys.PUSH_TELEMETRY;
}
+public static class BootstrapConfiguration {
+public final List bootstrapServers;
+public final ClientDnsLookup clientDnsLookup;
+public final long bootstrapResolveTimeoutMs;
+private final boolean bootstrapDisabled;
+
+public BootstrapConfiguration(final List bootstrapServers,
+ final ClientDnsLookup clientDnsLookup,
+ final long bootstrapResolveTimeoutMs) {
+this.bootstrapServers = bootstrapServers;
+this.clientDnsLookup = clientDnsLookup;
+this.bootstrapResolveTimeoutMs = bootstrapResolveTimeoutMs;
+this.bootstrapDisabled = false;
+}
+}
+
+private class BootstrapState {
+private final Timer timer;
+private final List bootstrapServers;
+private final ClientDnsLookup clientDnsLookup;
+private final long dnsResolutionTimeoutMs;
+private final boolean isDisabled;
+private boolean isBootstrapped;
+
+BootstrapState(BootstrapConfiguration bootstrapConfiguration) {
Review Comment:
or just make bootstrapConfiguration Optional so we
don't need to check isDisabled and isBootstrapped
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
Re: [PR] KAFKA-14648 - Do not fail clients if bootstrap servers is not immediately resolvable [kafka]
philipnee commented on code in PR #16629: URL: https://github.com/apache/kafka/pull/16629#discussion_r1688560266 ## clients/src/main/java/org/apache/kafka/clients/ClientUtils.java: ## @@ -228,6 +229,10 @@ public static NetworkClient createNetworkClient(AbstractConfig config, metricsGroupPrefix, channelBuilder, logContext); +bootstrapConfig = new NetworkClient.BootstrapConfiguration( Review Comment: The bootstrap server addresses are stored in metadata per KIP-899 but metadata.rebootstrap() is called within metadata.maybeUpdate(). So I don't know if it is easy to combine both config. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
