unknowntpo commented on code in PR #23673:
URL: https://github.com/apache/kafka/pull/23673#discussion_r4277429354
##########
clients/src/main/java/org/apache/kafka/clients/NetworkClient.java:
##########
@@ -94,6 +80,13 @@ private enum State {
CLOSED
}
+ private enum ApiVersionsResponseOutcome {
Review Comment:
Thanks for the update, this looks good to me. One tiny thing: the enum is
`ApiVersionsResponseOutcome` but the method is `determineApiVersionsAction`;
either name is fine, just pick one.
I also left a comment on `NetworkClientTest` about the bootstrap tests.
##########
clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java:
##########
@@ -1735,86 +1817,173 @@ private void bootstrapMetadataUpdater(final
MetadataUpdater metadataUpdater) {
}
@Test
- public void testEnsureBootstrappedSuccess() throws InterruptedException {
+ public void testEnsureBootstrappedSuccess() {
Metadata metadata = new Metadata(50, 50, 5000, new LogContext(), new
ClusterResourceListeners());
- BootstrapConfiguration config = BootstrapConfiguration.enabled(
- BOOTSTRAP_ADDRESSES,
- ClientDnsLookup.USE_ALL_DNS_IPS,
- 5000,
- CommonClientConfigs.DEFAULT_RETRY_BACKOFF_MS);
- NetworkClient client = new NetworkClient(selector, metadata, "mock",
Integer.MAX_VALUE,
- reconnectBackoffMsTest, 0, 64 * 1024, 64 * 1024,
- defaultRequestTimeoutMs, connectionSetupTimeoutMsTest,
connectionSetupTimeoutMaxMsTest,
- time, false, new ApiVersions(), new LogContext(),
- MetadataRecoveryStrategy.NONE, config, false);
+ List<InetSocketAddress> addresses = List.of(new
InetSocketAddress(InetAddress.getLoopbackAddress(), 9092));
+ try (MockedConstruction<BootstrapResolver> resolvers =
mockConstruction(BootstrapResolver.class);
+ NetworkClient client = createBootstrapNetworkClient(metadata)) {
+ BootstrapResolver resolver = resolvers.constructed().get(0);
+ when(resolver.isEnabled()).thenReturn(true);
+
when(resolver.poll(anyLong())).thenReturn(Optional.of(BootstrapResolver.Result.resolved(addresses)));
- // Async DNS resolution: first poll starts the resolution
- client.poll(1000, time.milliseconds());
-
- // Wait for async DNS resolution to complete and poll again to process
result
- MetadataUpdater metadataUpdater = TestUtils.fieldValue(client,
NetworkClient.class, "metadataUpdater");
- TestUtils.waitForCondition(() -> {
- client.poll(100, time.milliseconds());
- return metadataUpdater.isBootstrapped();
- }, "Bootstrap should complete");
-
- assertTrue(metadataUpdater.isBootstrapped());
+ client.poll(0, time.milliseconds());
+ assertEquals(Cluster.bootstrap(addresses).nodes(),
metadata.fetch().nodes());
+ client.poll(0, time.milliseconds());
+ verify(resolver).poll(anyLong());
+ }
}
@Test
- public void testEnsureBootstrappedPollTimeoutReturnsWithoutError() {
+ public void testPendingBootstrapAllowsEmptyNodeList() {
Metadata metadata = new Metadata(50, 50, 5000, new LogContext(), new
ClusterResourceListeners());
- // Use invalid addresses that cannot be resolved (using RFC 6761
reserved .invalid TLD)
- List<String> invalidAddresses = List.of("unresolvable.invalid:9092");
- BootstrapConfiguration config = BootstrapConfiguration.enabled(
- invalidAddresses,
- ClientDnsLookup.USE_ALL_DNS_IPS,
- 5000, // Long bootstrap timeout
- CommonClientConfigs.DEFAULT_RETRY_BACKOFF_MS);
- NetworkClient client = new NetworkClient(selector, metadata, "mock",
Integer.MAX_VALUE,
- reconnectBackoffMsTest, 0, 64 * 1024, 64 * 1024,
- defaultRequestTimeoutMs, connectionSetupTimeoutMsTest,
connectionSetupTimeoutMaxMsTest,
- time, false, new ApiVersions(), new LogContext(),
- MetadataRecoveryStrategy.NONE, config, false);
+ try (MockedConstruction<BootstrapResolver> resolvers =
mockConstruction(BootstrapResolver.class);
+ NetworkClient client = createBootstrapNetworkClient(metadata)) {
+ BootstrapResolver resolver = resolvers.constructed().get(0);
+ when(resolver.isEnabled()).thenReturn(true);
+ when(resolver.poll(anyLong())).thenReturn(Optional.empty());
+
+ client.poll(0, time.milliseconds());
+ assertTrue(metadata.fetch().nodes().isEmpty());
+ LeastLoadedNode leastLoadedNode =
client.leastLoadedNode(time.milliseconds());
+ assertNull(leastLoadedNode.node());
+ assertFalse(leastLoadedNode.hasNodeAvailableOrConnectionReady());
+ }
+ }
- // Directly call ensureBootstrapped
- // Should return without error even though bootstrap hasn't succeeded
(will retry on next poll)
- // DNS resolution will fail but timeout hasn't been reached yet
- client.ensureBootstrapped(time.milliseconds());
+ @Test
+ public void testBootstrapFailureIsRecordedInMetadata() {
+ Metadata metadata = spy(new Metadata(50, 50, 5000, new LogContext(),
new ClusterResourceListeners()));
+ BootstrapResolutionException exception = new
BootstrapResolutionException("DNS resolution timed out");
+ try (MockedConstruction<BootstrapResolver> resolvers =
mockConstruction(BootstrapResolver.class);
+ NetworkClient client = createBootstrapNetworkClient(metadata)) {
+ BootstrapResolver resolver = resolvers.constructed().get(0);
+ when(resolver.isEnabled()).thenReturn(true);
+ when(resolver.poll(anyLong()))
+
.thenReturn(Optional.of(BootstrapResolver.Result.failed(exception)))
+ .thenReturn(Optional.empty());
- // Verify that no exception was thrown and metadata is still empty
- assertEquals(0, metadata.fetch().nodes().size(), "Metadata should have
no nodes after failed DNS resolution");
+ client.poll(0, time.milliseconds());
+ assertSame(exception,
assertThrows(BootstrapResolutionException.class,
metadata::maybeThrowBootstrapFatalException));
+ client.poll(0, time.milliseconds());
+ assertSame(exception,
assertThrows(BootstrapResolutionException.class,
metadata::maybeThrowBootstrapFatalException));
+ verify(metadata).bootstrapFatalError(exception);
+
assertFalse(client.leastLoadedNode(time.milliseconds()).hasNodeAvailableOrConnectionReady());
+ }
}
@Test
- public void testEnsureBootstrappedRetryUntilSuccess() throws
InterruptedException {
+ public void testAlreadyBootstrappedSkipsResolution() {
Metadata metadata = new Metadata(50, 50, 5000, new LogContext(), new
ClusterResourceListeners());
- BootstrapConfiguration config = BootstrapConfiguration.enabled(
- BOOTSTRAP_ADDRESSES,
- ClientDnsLookup.USE_ALL_DNS_IPS,
- 5000,
- CommonClientConfigs.DEFAULT_RETRY_BACKOFF_MS);
- NetworkClient client = new NetworkClient(selector, metadata, "mock",
Integer.MAX_VALUE,
- reconnectBackoffMsTest, 0, 64 * 1024, 64 * 1024,
- defaultRequestTimeoutMs, connectionSetupTimeoutMsTest,
connectionSetupTimeoutMaxMsTest,
- time, false, new ApiVersions(), new LogContext(),
- MetadataRecoveryStrategy.NONE, config, false);
+ metadata.bootstrap(List.of(new
InetSocketAddress(InetAddress.getLoopbackAddress(), 9092)));
+ try (MockedConstruction<BootstrapResolver> resolvers =
mockConstruction(BootstrapResolver.class);
+ NetworkClient client = createBootstrapNetworkClient(metadata)) {
+ BootstrapResolver resolver = resolvers.constructed().get(0);
+ when(resolver.isEnabled()).thenReturn(true);
+ time.sleep(10_000);
+ Thread.currentThread().interrupt();
Review Comment:
This assertion cannot fail here: the resolver is a mock, so the
`Thread.interrupted()` check in `BootstrapResolver.poll` is never reached, and
`ensureBootstrapped` returns early because metadata is already bootstrapped.
`verify(resolver, never()).poll(anyLong())` already covers this test, so the
interrupt setup and `time.sleep(10_000)` can go.
`BootstrapResolver` now owns the state machine but has no test of its own.
`testRealBootstrapResolverDeliversOutcomeToMetadata` covers success (including
a result arriving exactly at the deadline) and timeout with no resubmit. Not
covered today:
- interrupt: cancel the pending future, throw `InterruptException`, keep the
flag set
- retry after a failed resolution (empty list or `CompletionException`),
including the backoff
- timer starting on the first `poll`, not at construction
These paths depend on time and on when the pending future completes, so they
are hard to reach from `NetworkClientTest` without mocking the resolver away
(as the new tests do) or mocking `Executors`. A `BootstrapResolverTest` with
`MockTime` and an injected `CompletableFuture` can cover each of them in a few
lines. `BootstrapResolver` already has a package-private constructor that takes
an `executor` and a `submitResolution` supplier (the 5-argument one) and has no
caller outside the class; that looks like the seam for it. Otherwise it can be
folded back into the private 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]
