m1a2st commented on code in PR #23227:
URL: https://github.com/apache/kafka/pull/23227#discussion_r3913714539
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -731,6 +741,83 @@ public void
testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
assertPoll(1, commitRequestManager);
}
+ @Test
+ public void testMaximumTimeToWaitWhenCoordinatorUnknownDoesNotSpin() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
+
+ time.sleep(100);
+ long result =
commitRequestManager.maximumTimeToWait(time.milliseconds());
+
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 when the coordinator is unknown to
avoid a busy-spin; got " + result);
+ assertEquals(100, result);
+ }
+
+ @Test
+ public void
testMaximumTimeToWaitDoesNotSpinDuringRealBootstrapDnsResolution() throws
Exception {
+ long bootstrapResolveTimeoutMs = 1000;
+
+ BootstrapConfiguration bootstrapConfiguration =
BootstrapConfiguration.enabled(
+ List.of("unresolvable.invalid:9092"),
+ ClientDnsLookup.USE_ALL_DNS_IPS,
+ bootstrapResolveTimeoutMs,
+ retryBackoffMs
+ );
+
+ ConsumerConfig config = new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class,
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class,
+ ConsumerConfig.GROUP_ID_CONFIG, DEFAULT_GROUP_ID,
+ ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"unresolvable.invalid:9092",
+ // Much shorter than bootstrapResolveTimeoutMs, so the auto-commit
timer expires several
+ // times while the coordinator is still (and will remain, since
DNS never resolves) unknown.
+ ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100",
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"
+ ));
+
+ ConsumerMetadata metadata = new ConsumerMetadata(config,
subscriptionState, logContext, new ClusterResourceListeners());
+
+ MockSelector selector = new MockSelector(time);
+ NetworkClient networkClient = new NetworkClient(selector, metadata,
"test-client",
+ Integer.MAX_VALUE, 50, 1000, 64 * 1024, 64 * 1024, 1000, 5000,
30000,
+ time, false, new ApiVersions(), logContext,
+ MetadataRecoveryStrategy.NONE, bootstrapConfiguration, false);
+
+
+ CoordinatorRequestManager realCoordinatorRequestManager = new
CoordinatorRequestManager(
+ logContext, retryBackoffMs, retryBackoffMaxMs, DEFAULT_GROUP_ID);
+
+ CommitRequestManager realCommitRequestManager = new
CommitRequestManager(
+ time, logContext, subscriptionState, config,
realCoordinatorRequestManager,
+ mock(OffsetCommitCallbackInvoker.class), DEFAULT_GROUP_ID,
Optional.empty(), retryBackoffMs,
+ retryBackoffMaxMs, OptionalDouble.of(0), new Metrics(), metadata);
+
+ try (NetworkClientDelegate networkClientDelegate = new
NetworkClientDelegate(time, config, logContext, networkClient, metadata,
+ mock(BackgroundEventHandler.class), false,
mock(AsyncConsumerMetrics.class));
+ ) {
+ long deadline = time.milliseconds() + bootstrapResolveTimeoutMs +
3000;
+ boolean sawBootstrapException = false;
+
+ while (time.milliseconds() < deadline) {
Review Comment:
Yes, MockSelector addresses that, its `poll(timeout)` calls
`time.sleep(timeout)`
##########
clients/src/test/java/org/apache/kafka/clients/consumer/internals/CommitRequestManagerTest.java:
##########
@@ -731,6 +741,83 @@ public void
testAutoCommitOnIntervalSkippedIfPreviousOneInFlight() {
assertPoll(1, commitRequestManager);
}
+ @Test
+ public void testMaximumTimeToWaitWhenCoordinatorUnknownDoesNotSpin() {
+ CommitRequestManager commitRequestManager = create(true, 100);
+
when(coordinatorRequestManager.coordinator()).thenReturn(Optional.empty());
+
+ time.sleep(100);
+ long result =
commitRequestManager.maximumTimeToWait(time.milliseconds());
+
+ assertTrue(result > 0,
+ "maximumTimeToWait must be > 0 when the coordinator is unknown to
avoid a busy-spin; got " + result);
+ assertEquals(100, result);
+ }
+
+ @Test
+ public void
testMaximumTimeToWaitDoesNotSpinDuringRealBootstrapDnsResolution() throws
Exception {
+ long bootstrapResolveTimeoutMs = 1000;
+
+ BootstrapConfiguration bootstrapConfiguration =
BootstrapConfiguration.enabled(
+ List.of("unresolvable.invalid:9092"),
+ ClientDnsLookup.USE_ALL_DNS_IPS,
+ bootstrapResolveTimeoutMs,
+ retryBackoffMs
+ );
+
+ ConsumerConfig config = new ConsumerConfig(Map.of(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class,
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class,
+ ConsumerConfig.GROUP_ID_CONFIG, DEFAULT_GROUP_ID,
+ ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
"unresolvable.invalid:9092",
+ // Much shorter than bootstrapResolveTimeoutMs, so the auto-commit
timer expires several
+ // times while the coordinator is still (and will remain, since
DNS never resolves) unknown.
+ ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100",
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true"
+ ));
+
+ ConsumerMetadata metadata = new ConsumerMetadata(config,
subscriptionState, logContext, new ClusterResourceListeners());
+
+ MockSelector selector = new MockSelector(time);
+ NetworkClient networkClient = new NetworkClient(selector, metadata,
"test-client",
+ Integer.MAX_VALUE, 50, 1000, 64 * 1024, 64 * 1024, 1000, 5000,
30000,
+ time, false, new ApiVersions(), logContext,
+ MetadataRecoveryStrategy.NONE, bootstrapConfiguration, false);
+
+
+ CoordinatorRequestManager realCoordinatorRequestManager = new
CoordinatorRequestManager(
+ logContext, retryBackoffMs, retryBackoffMaxMs, DEFAULT_GROUP_ID);
+
+ CommitRequestManager realCommitRequestManager = new
CommitRequestManager(
+ time, logContext, subscriptionState, config,
realCoordinatorRequestManager,
+ mock(OffsetCommitCallbackInvoker.class), DEFAULT_GROUP_ID,
Optional.empty(), retryBackoffMs,
+ retryBackoffMaxMs, OptionalDouble.of(0), new Metrics(), metadata);
+
+ try (NetworkClientDelegate networkClientDelegate = new
NetworkClientDelegate(time, config, logContext, networkClient, metadata,
+ mock(BackgroundEventHandler.class), false,
mock(AsyncConsumerMetrics.class));
+ ) {
+ long deadline = time.milliseconds() + bootstrapResolveTimeoutMs +
3000;
+ boolean sawBootstrapException = false;
+
+ while (time.milliseconds() < deadline) {
+ // Drives the real NetworkClient's ensureBootstrapped()/async
DNS resolution forward;
+ // the coordinator never becomes known since there is no real
broker to respond.
+ networkClientDelegate.poll(50, time.milliseconds());
+
+ Optional<Exception> metadataError =
networkClientDelegate.getAndClearMetadataError();
+ if (metadataError.isPresent()) {
+ assertInstanceOf(BootstrapResolutionException.class,
metadataError.get());
+ sawBootstrapException = true;
+ break;
+ }
+
+ long waitMs =
realCommitRequestManager.maximumTimeToWait(time.milliseconds());
+ assertTrue(waitMs > 0, "maximumTimeToWait must be > 0 while
real bootstrap DNS resolution is pending; got " + waitMs);
Review Comment:
Sure, addressed 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]
