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<String, Object>
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<InetSocketAddress> resolveAddress(
+ String url,
+ String host,
+ Integer port,
+ ClientDnsLookup clientDnsLookup) throws UnknownHostException {
+
+ List<InetSocketAddress> 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<InetSocketAddress> parseAddresses(List<String> urls,
ClientDnsLookup clientDnsLookup) {
+ List<InetSocketAddress> 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 this service, please contact Infrastructure at:
[email protected]
