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..08bbcbef9 100644 --- a/core/src/test/java/tech/ydb/core/impl/YdbTransportImplTest.java +++ b/core/src/test/java/tech/ydb/core/impl/YdbTransportImplTest.java @@ -8,9 +8,11 @@ import java.util.Queue; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentLinkedQueue; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.Executor; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import javax.annotation.Nonnull; @@ -27,6 +29,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 +186,69 @@ public void asyncBuildGoodTest() { Assert.assertTrue(isReady.isDone()); } + @Test + public void asyncWaitingForReadyTest() throws Exception { + CountDownLatch discoveryLatch = new CountDownLatch(1); + Queue lazyTasks = new ConcurrentLinkedQueue<>(); + Executor lazyExecutor = (Runnable r) -> { + lazyTasks.add(r); + discoveryLatch.countDown(); + }; + + Mockito.when(discoveryChannel.newCall(Mockito.eq(DiscoveryServiceGrpc.getListEndpointsMethod()), Mockito.any())) + .thenReturn(MockedCall.discovery(lazyExecutor, "self", new EndpointRecord("node", 2136))); + Mockito.when(transportChannel.newCall(Mockito.eq(DiscoveryServiceGrpc.getWhoAmIMethod()), Mockito.any())) + .thenReturn(MockedCall.whoAmICall("i am node")); + + Duration discoveryTimeout = Duration.ofSeconds(5); + Duration callTimeout = Duration.ofMillis(50); + + Assert.assertEquals(0, lazyTasks.size()); + try (GrpcTransport transport = GrpcTransport.forConnectionString("grpc://mocked:2136/local") + .withInitMode(GrpcTransportBuilder.InitMode.ASYNC) + .withDiscoveryTimeout(discoveryTimeout) + .withChannelFactoryBuilder(builder -> channelFactory) + .build() + ) { + Assert.assertTrue(discoveryLatch.await(1, TimeUnit.SECONDS)); + Assert.assertEquals(1, lazyTasks.size()); // a discovery call + + long startedAt = System.nanoTime(); + CompletableFuture> call1 = CompletableFuture.supplyAsync( + () ->transport.unaryCall( + DiscoveryServiceGrpc.getWhoAmIMethod(), + GrpcRequestSettings.newBuilder().withDeadline(callTimeout).build(), + DiscoveryProtos.WhoAmIRequest.newBuilder().build() + ).join(), + testScheduler + ); + + Assert.assertFalse(call1.isDone()); + Result res1 = call1.get(5, TimeUnit.SECONDS); + Assert.assertEquals(StatusCode.CLIENT_INTERNAL_ERROR, res1.getStatus().getCode()); + Assert.assertEquals(1, lazyTasks.size()); // a new call wasn't executed + + long nanos = System.nanoTime() - startedAt; + Assert.assertTrue(nanos >= callTimeout.toNanos()); + Assert.assertTrue(nanos < discoveryTimeout.toNanos()); + + CompletableFuture> call2 = CompletableFuture.supplyAsync( + () ->transport.unaryCall( + DiscoveryServiceGrpc.getWhoAmIMethod(), + GrpcRequestSettings.newBuilder().build(), + DiscoveryProtos.WhoAmIRequest.newBuilder().build() + ).join(), + testScheduler + ); + Assert.assertFalse(call2.isDone()); + Assert.assertEquals(1, lazyTasks.size()); // a new call wasn't executed + + lazyTasks.poll().run(); // complete discovery + Result res2 = call2.join(); + Assert.assertTrue(res2.isSuccess()); + } + } + @Test public void failFastOnMissingPort() { String endpoint = "127.1.2.3";