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<ClientResponse> 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<String> bootstrapServers;
+ public final ClientDnsLookup clientDnsLookup;
+ public final long bootstrapResolveTimeoutMs;
+ public final long retryBackoffMs;
+
+ private BootstrapConfiguration(final List<String> 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<String>
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.
+ *
+ * <p>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 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?
##########
clients/src/main/java/org/apache/kafka/clients/ClientUtils.java:
##########
@@ -53,6 +53,76 @@ 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;
+ }
+ for (String url : urls) {
+ if (Thread.currentThread().isInterrupted()) {
+ break;
+ }
+
+ final String host = getHost(url);
+ final Integer port = getPort(url);
+
+ if (host == null || port == null) {
+ log.warn("Skipping invalid bootstrap URL: {}", url);
Review Comment:
isn't this something that we should catch higher up and throw a
ConfigException for?
We used to, but seems we lost the ConfigException and just have this warn?
(I tracked this and can only see it from the ShareConsumer, which I expect will
disappear too in the follow-up PR)
https://github.com/apache/kafka/blob/b8429b93a76ffa7ad9a2db62fa32dbafd4ac2f51/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java#L73-L74
I guess we should keep throwing ConfigException in this case, and that can
really happen at a higher level, not down here in the async path
--
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]
