diff --git a/core/src/main/java/tech/ydb/core/impl/YdbTransportImpl.java b/core/src/main/java/tech/ydb/core/impl/YdbTransportImpl.java index a8bc6255a..d0d8e1e4d 100644 --- a/core/src/main/java/tech/ydb/core/impl/YdbTransportImpl.java +++ b/core/src/main/java/tech/ydb/core/impl/YdbTransportImpl.java @@ -6,6 +6,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import com.google.common.base.Strings; import org.slf4j.Logger; @@ -135,11 +136,14 @@ public AuthCallOptions getAuthCallOptions() { protected GrpcChannel getChannel(GrpcRequestSettings settings) { EndpointRecord endpoint = endpointPool.getEndpoint(channelPool.getReadyEndpoints(), settings); if (endpoint == null) { - long timeout = -1; + // negative value tells waitReady to use the default discovery timeout + long timeoutMs = -1; if (settings.getDeadlineAfter() != 0) { - timeout = settings.getDeadlineAfter() - System.nanoTime(); + long leftNanos = settings.getDeadlineAfter() - System.nanoTime(); + // an already expired deadline must not fall back to the default timeout + timeoutMs = Math.max(TimeUnit.NANOSECONDS.toMillis(leftNanos), 1); } - discovery.waitReady(timeout); + discovery.waitReady(timeoutMs); endpoint = endpointPool.getEndpoint(Collections.emptySet(), settings); } return channelPool.getChannel(endpoint); diff --git a/core/src/test/java/tech/ydb/core/impl/YdbTransportImplTest.java b/core/src/test/java/tech/ydb/core/impl/YdbTransportImplTest.java index d708ab576..c1a61dcf4 100644 --- a/core/src/test/java/tech/ydb/core/impl/YdbTransportImplTest.java +++ b/core/src/test/java/tech/ydb/core/impl/YdbTransportImplTest.java @@ -27,6 +27,7 @@ import tech.ydb.core.UnexpectedResultException; import tech.ydb.core.grpc.GrpcRequestSettings; import tech.ydb.core.grpc.GrpcTransport; +import tech.ydb.core.grpc.GrpcTransportBuilder; import tech.ydb.core.grpc.YdbHeaders; import tech.ydb.core.impl.pool.EndpointRecord; import tech.ydb.core.impl.pool.ManagedChannelFactory; @@ -183,6 +184,41 @@ public void asyncBuildGoodTest() { Assert.assertTrue(isReady.isDone()); } + @Test(timeout = 30_000) + public void requestDeadlineLimitsWaitingForDiscovery() { + Mockito.when(discoveryChannel.newCall(Mockito.eq(DiscoveryServiceGrpc.getListEndpointsMethod()), Mockito.any())) + .thenReturn(MockedCall.neverAnswer(testScheduler)); + + // deliberately much longer than the deadline of the request below + Duration discoveryTimeout = Duration.ofMinutes(10); + + try (GrpcTransport transport = GrpcTransport.forConnectionString("grpc://mocked:2136/local") + .withInitMode(GrpcTransportBuilder.InitMode.ASYNC) + .withDiscoveryTimeout(discoveryTimeout) + .withChannelFactoryBuilder(builder -> channelFactory) + .build() + ) { + + GrpcRequestSettings settings = GrpcRequestSettings.newBuilder() + .withDeadline(Duration.ofMillis(200)) + .build(); + + long startedAt = System.nanoTime(); + Result result = transport + .unaryCall( + DiscoveryServiceGrpc.getWhoAmIMethod(), + settings, + DiscoveryProtos.WhoAmIRequest.newBuilder().build() + ) + .join(); + Duration elapsed = Duration.ofNanos(System.nanoTime() - startedAt); + + Assert.assertFalse(result.isSuccess()); + // the call must give up near its own deadline instead of waiting for the discovery timeout + Assert.assertTrue("discovery waiting took " + elapsed, elapsed.compareTo(discoveryTimeout) < 0); + } + } + @Test public void failFastOnMissingPort() { String endpoint = "127.1.2.3";