Copilot commented on code in PR #22897:
URL: https://github.com/apache/kafka/pull/22897#discussion_r3636337251
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -835,6 +909,21 @@ public LeastLoadedNode leastLoadedNode(long now) {
}
}
+ /**
+ * Handle the case when there are no nodes available.
+ * <p>
+ * 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 Javadoc says to “return null” during bootstrap, but the method returns
a LeastLoadedNode instance with a null node. Update the comment to match the
actual behavior.
This issue also appears in the following locations of the same file:
- line 1335
- line 1340
- line 1365
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -11881,4 +11880,34 @@ public void testOutOfMemoryErrorPropagation() throws
Exception {
TestUtils.assertFutureThrows(OutOfMemoryError.class,
result.names());
}
}
+
+ @Test
+ public void testAdminBootstrapResolutionExceptionPropagated() throws
Exception {
+ String invalidHost = "unresolvable.invalid:9092";
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, invalidHost);
+ configs.put(CommonClientConfigs.BOOTSTRAP_RESOLVE_TIMEOUT_MS_CONFIG,
"3000");
+
+ try (Admin admin = Admin.create(configs)) {
+ assertThrows(BootstrapResolutionException.class, () -> {
+ long startTime = System.currentTimeMillis();
+ long maxWaitTime = 15000;
+ while (System.currentTimeMillis() - startTime < maxWaitTime) {
+ try {
+ admin.listTopics().names().get();
+ } catch (ExecutionException e) {
+ if (e.getCause() instanceof
BootstrapResolutionException) {
+ throw (BootstrapResolutionException) e.getCause();
+ }
+ }
+ }
Review Comment:
This while-loop may spin very fast when listTopics().names().get() fails
quickly (e.g., before the bootstrap timeout is reached), consuming CPU
unnecessarily. Adding a small backoff makes the test less resource-intensive
while still bounding runtime.
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -516,6 +516,43 @@ static String prettyPrintException(Throwable throwable) {
return throwable.getClass().getSimpleName();
}
+ /**
+ * Determines which bootstrap configuration to use based on the provided
config.
+ * Validates that exactly one of bootstrap.servers or
bootstrap.controllers is configured.
+ *
+ * @param config The admin client configuration
+ * @return true if using bootstrap.controllers, false if using
bootstrap.servers
+ * @throws ConfigException if both or neither bootstrap configurations are
set
+ */
+ static boolean determineBootstrapType(AdminClientConfig config) {
+ List<String> bootstrapServers =
config.getList(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG);
+ if (bootstrapServers == null) {
+ bootstrapServers = Collections.emptyList();
+ }
Review Comment:
determineBootstrapType introduces new validation logic for bootstrap.servers
vs bootstrap.controllers, but the dedicated unit tests that previously covered
this behavior (AdminBootstrapAddressesTest) were removed. Add focused tests to
cover the three cases: neither set, both set, and exactly one set.
##########
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<Node>
nodes) {
+ List<InetSocketAddress> serverAddresses = new ArrayList<>();
+ nodes.forEach(node -> serverAddresses.add(new
InetSocketAddress(node.host(), node.port())));
+ metadata.bootstrap(serverAddresses);
+ }
+
+ private void bootstrapMetadata(Metadata metadata) {
+ List<InetSocketAddress> serverAddresses = new ArrayList<>(List.of(
+ new InetSocketAddress("localhost0", 8000),
+ new InetSocketAddress("localhost1", 8000)
+ ));
+ metadata.bootstrap(serverAddresses);
+ }
+
+ private void bootstrapMetadataUpdater(final MetadataUpdater
metadataUpdater) {
+ List<InetSocketAddress> serverAddresses = new ArrayList<>(List.of(
+ new InetSocketAddress("localhost0", 8000),
+ new InetSocketAddress("localhost1", 8000)
+ ));
+ metadataUpdater.bootstrap(serverAddresses);
+ }
Review Comment:
These helpers use new InetSocketAddress("localhost0", …) / ("localhost1",
…), which can trigger real DNS lookups during tests. Using
InetSocketAddress.createUnresolved avoids external DNS dependency and speeds
up/derisks the test suite.
##########
clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java:
##########
@@ -3416,4 +3424,59 @@ public static void resetCounters() {
CLOSE_COUNT.set(0);
}
}
+
+ @Test
+ public void testProducerBootstrapResolutionExceptionPropagated() {
+ String invalidHost = "unresolvable.invalid:9092";
+ Map<String, Object> 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<String, String> 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:
This loop can become a tight spin if partitionsFor() returns quickly before
the bootstrap timeout elapses, which can waste CPU and make the test noisy
under load. Adding a small sleep/backoff makes the test friendlier and still
validates the behavior.
--
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]
