Merge branch 'release_v2.4.0' into update-dependency-ydb-proto-api

Author: asmyasnikov

core/pom.xml

@@ -29,7 +29,7 @@
         <dependency>
             <groupId>io.grpc</groupId>
             <artifactId>grpc-netty</artifactId>
-            <version>1.59.1</version>
+            <version>1.68.3</version>
             <optional>true</optional>
         </dependency>
 

core/src/main/java/tech/ydb/core/grpc/GrpcTransportBuilder.java

@@ -79,7 +79,6 @@ public enum InitMode {
     private Executor callExecutor = MoreExecutors.directExecutor();
     private AuthRpcProvider<? super GrpcAuthRpc> authProvider = NopAuthProvider.INSTANCE;
     private long readTimeoutMillis = 0;
-    private long connectTimeoutMillis = 30_000;
     private long discoveryTimeoutMillis = 60_000;
     private boolean useDefaultGrpcResolver = false;
     private GrpcCompression compression = GrpcCompression.NO_COMPRESSION;
@@ -151,8 +150,9 @@ public long getReadTimeoutMillis() {
         return readTimeoutMillis;
     }
 
+    @Deprecated
     public long getConnectTimeoutMillis() {
-        return connectTimeoutMillis;
+        return 10_000;
     }
 
     public long getDiscoveryTimeoutMillis() {
@@ -315,15 +315,13 @@ public GrpcTransportBuilder withReadTimeout(long timeout, TimeUnit unit) {
         return this;
     }
 
+    @Deprecated
     public GrpcTransportBuilder withConnectTimeout(Duration timeout) {
-        this.connectTimeoutMillis = timeout.toMillis();
-        Preconditions.checkArgument(connectTimeoutMillis > 0, "connectTimeoutMillis must be greater than 0");
         return this;
     }
 
+    @Deprecated
     public GrpcTransportBuilder withConnectTimeout(long timeout, TimeUnit unit) {
-        this.connectTimeoutMillis = unit.toMillis(timeout);
-        Preconditions.checkArgument(connectTimeoutMillis > 0, "connectTimeoutMillis must be greater than 0");
         return this;
     }
 

core/src/main/java/tech/ydb/core/impl/YdbTransportImpl.java

@@ -53,12 +53,8 @@ public YdbTransportImpl(GrpcTransportBuilder builder) {
 
         this.channelFactory = builder.getManagedChannelFactory();
         this.scheduler = builder.getSchedulerFactory().get();
-        this.callOptions = new AuthCallOptions(
-                scheduler,
-                Collections.singletonList(discoveryEndpoint),
-                channelFactory,
-                builder
-        );
+        this.callOptions = new AuthCallOptions(scheduler, Collections.singletonList(discoveryEndpoint),
+                channelFactory, builder);
         this.channelPool = new GrpcChannelPool(channelFactory, scheduler);
         this.endpointPool = new EndpointPool(balancingSettings);
         this.discovery = new YdbDiscovery(new DiscoveryHandler(), scheduler, database, discoveryTimeout);
@@ -178,8 +174,14 @@ protected GrpcChannel getChannel(GrpcRequestSettings settings) {
     }
 
     @Override
-    protected void pessimizeEndpoint(EndpointRecord endpoint, String reason) {
-        endpointPool.pessimizeEndpoint(endpoint, reason);
+    protected void updateChannelStatus(GrpcChannel channel, io.grpc.Status status) {
+        // Usally CANCELLED is received when ClientCall is canceled on client side
+        if (!status.isOk() && status.getCode() != io.grpc.Status.Code.CANCELLED &&
+                status.getCode() != io.grpc.Status.Code.DEADLINE_EXCEEDED &&
+                status.getCode() != io.grpc.Status.Code.RESOURCE_EXHAUSTED
+        ) {
+            endpointPool.pessimizeEndpoint(channel.getEndpoint());
+        }
     }
 
     private class DiscoveryHandler implements YdbDiscovery.Handler {

core/src/main/java/tech/ydb/core/impl/pool/GrpcChannel.java

@@ -1,9 +1,6 @@
 package tech.ydb.core.impl.pool;
 
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 import io.grpc.Channel;
 import io.grpc.ConnectivityState;
@@ -14,25 +11,21 @@
 /**
  * @author Nikolay Perfilov
  */
-public class GrpcChannel {
+public final class GrpcChannel implements Runnable {
+
     /* Channel shutdown waits for finish of active grpc calls, so there must be enough time to complete them all */
     private static final long WAIT_FOR_CLOSING_MS = 5000;
     private static final Logger logger = LoggerFactory.getLogger(GrpcChannel.class);
 
     private final EndpointRecord endpoint;
     private final ManagedChannel channel;
-    private final long connectTimeoutMs;
-    private final ReadyWatcher readyWatcher;
 
     public GrpcChannel(EndpointRecord endpoint, ManagedChannelFactory factory) {
         try {
             logger.debug("Creating grpc channel with {}", endpoint);
             this.endpoint = endpoint;
-            this.channel = factory.newManagedChannel(endpoint.getHost(), endpoint.getPort(),
-                endpoint.getAuthority());
-            this.connectTimeoutMs = factory.getConnectTimeoutMs();
-            this.readyWatcher = new ReadyWatcher();
-            this.readyWatcher.checkState();
+            this.channel = factory.newManagedChannel(endpoint.getHost(), endpoint.getPort(), endpoint.getAuthority());
+            checkState();
         } catch (Throwable th) {
             throw new RuntimeException("cannot create channel", th);
         }
@@ -43,7 +36,7 @@ public EndpointRecord getEndpoint() {
     }
 
     public Channel getReadyChannel() {
-        return readyWatcher.getReadyChannel();
+        return channel;
     }
 
     public boolean isShutdown() {
@@ -72,50 +65,14 @@ public boolean shutdown() {
         }
     }
 
-    private class ReadyWatcher implements Runnable {
-        private final CompletableFuture<ManagedChannel> future = new CompletableFuture<>();
-
-        public Channel getReadyChannel() {
-            try {
-                return future.get(connectTimeoutMs, TimeUnit.MILLISECONDS);
-            } catch (InterruptedException ex) {
-                logger.warn("Grpc channel {} ready waiting is interrupted: ", endpoint, ex);
-                Thread.currentThread().interrupt();
-            } catch (ExecutionException ex) {
-                logger.warn("Grpc channel {} connecting problem: ", endpoint, ex);
-                throw new RuntimeException("Channel " + endpoint + " connecting problem", ex);
-            } catch (TimeoutException ex) {
-                logger.warn("Grpc channel {} connect timeout exceeded", endpoint);
-                throw new RuntimeException("Channel " + endpoint + " connecting timeout");
-            }
-            return null;
-        }
-
-        public void checkState() {
-            ConnectivityState state = channel.getState(true);
-            logger.debug("Grpc channel {} new state: {}", endpoint, state);
-            switch (state) {
-                case READY:
-                    future.complete(channel);
-                    // keep tracking channel state
-                    channel.notifyWhenStateChanged(state, this);
-                    break;
-                case SHUTDOWN:
-                    future.completeExceptionally(new IllegalStateException("Grpc channel already closed"));
-                    break;
-                case TRANSIENT_FAILURE:
-                case CONNECTING:
-                case IDLE:
-                default:
-                    // keep tracking channel state
-                    channel.notifyWhenStateChanged(state, this);
-                    break;
-            }
-        }
+    private void checkState() {
+        ConnectivityState state = channel.getState(true);
+        logger.debug("Grpc channel {} new state: {}", endpoint, state);
+        channel.notifyWhenStateChanged(state, this);
+    }
 
-        @Override
-        public void run() {
-            checkState();
-        }
+    @Override
+    public void run() {
+        checkState();
     }
 }

core/src/main/java/tech/ydb/core/impl/pool/ManagedChannelFactory.java

@@ -14,6 +14,4 @@ interface Builder {
     }
 
     ManagedChannel newManagedChannel(String host, int port, String authority);
-
-    long getConnectTimeoutMs();
 }

core/src/main/java/tech/ydb/core/impl/pool/NettyChannelFactory.java

@@ -38,7 +38,6 @@ public class NettyChannelFactory implements ManagedChannelFactory {
     private final boolean useTLS;
     private final byte[] cert;
     private final boolean retryEnabled;
-    private final long connectTimeoutMs;
     private final boolean useDefaultGrpcResolver;
     private final Long grpcKeepAliveTimeMillis;
     private final List<Consumer<? super ManagedChannelBuilder<?>>> initializers;
@@ -49,17 +48,11 @@ private NettyChannelFactory(GrpcTransportBuilder builder) {
         this.useTLS = builder.getUseTls();
         this.cert = builder.getCert();
         this.retryEnabled = builder.isEnableRetry();
-        this.connectTimeoutMs = builder.getConnectTimeoutMillis();
         this.useDefaultGrpcResolver = builder.useDefaultGrpcResolver();
         this.grpcKeepAliveTimeMillis = builder.getGrpcKeepAliveTimeMillis();
         this.initializers = builder.getChannelInitializers();
     }
 
-    @Override
-    public long getConnectTimeoutMs() {
-        return this.connectTimeoutMs;
-    }
-
     @SuppressWarnings("deprecation")
     @Override
     public ManagedChannel newManagedChannel(String host, int port, String sslHostOverride) {

core/src/main/java/tech/ydb/core/impl/pool/ShadedNettyChannelFactory.java

@@ -38,7 +38,6 @@ public class ShadedNettyChannelFactory implements ManagedChannelFactory {
     private final boolean useTLS;
     private final byte[] cert;
     private final boolean retryEnabled;
-    private final long connectTimeoutMs;
     private final boolean useDefaultGrpcResolver;
     private final Long grpcKeepAliveTimeMillis;
     private final List<Consumer<? super ManagedChannelBuilder<?>>> initializers;
@@ -49,17 +48,11 @@ public ShadedNettyChannelFactory(GrpcTransportBuilder builder) {
         this.useTLS = builder.getUseTls();
         this.cert = builder.getCert();
         this.retryEnabled = builder.isEnableRetry();
-        this.connectTimeoutMs = builder.getConnectTimeoutMillis();
         this.useDefaultGrpcResolver = builder.useDefaultGrpcResolver();
         this.grpcKeepAliveTimeMillis = builder.getGrpcKeepAliveTimeMillis();
         this.initializers = builder.getChannelInitializers();
     }
 
-    @Override
-    public long getConnectTimeoutMs() {
-        return this.connectTimeoutMs;
-    }
-
     @SuppressWarnings("deprecation")
     @Override
     public ManagedChannel newManagedChannel(String host, int port, String sslHostOverride) {

core/src/test/java/tech/ydb/core/impl/pool/DefaultChannelFactoryTest.java

@@ -5,8 +5,6 @@
 import java.io.IOException;
 import java.nio.file.Files;
 import java.security.cert.CertificateException;
-import java.time.Duration;
-import java.util.concurrent.TimeUnit;
 
 import com.google.common.io.ByteStreams;
 import io.grpc.ClientInterceptor;
@@ -79,7 +77,6 @@ public void defaultParams() {
         ManagedChannelFactory factory = ChannelFactoryLoader.load().buildFactory(builder);
         channelStaticMock.verify(FOR_ADDRESS, Mockito.times(0));
 
-        Assert.assertEquals(30_000l, factory.getConnectTimeoutMs());
         Assert.assertSame(channelMock, factory.newManagedChannel(MOCKED_HOST, MOCKED_PORT, null));
 
         channelStaticMock.verify(FOR_ADDRESS, Mockito.times(1));
@@ -100,13 +97,11 @@ public void defaultParams() {
     public void defaultSslFactory() {
         GrpcTransportBuilder builder = GrpcTransport.forHost(MOCKED_HOST, MOCKED_PORT, "/Root")
                 .withSecureConnection()
-                .withGrpcRetry(true)
-                .withConnectTimeout(Duration.ofMinutes(1));
+                .withGrpcRetry(true);
 
         ManagedChannelFactory factory = ChannelFactoryLoader.load().buildFactory(builder);
         channelStaticMock.verify(FOR_ADDRESS, Mockito.times(0));
 
-        Assert.assertEquals(60000l, factory.getConnectTimeoutMs());
         Assert.assertSame(channelMock, factory.newManagedChannel(MOCKED_HOST, MOCKED_PORT, null));
 
         channelStaticMock.verify(FOR_ADDRESS, Mockito.times(1));
@@ -129,7 +124,7 @@ public void customChannelInitializer() {
                 .withUseDefaultGrpcResolver(true);
 
         ManagedChannelFactory factory = ShadedNettyChannelFactory
-                .withInterceptor(ForwardingChannelBuilder2::enableFullStreamDecompression)
+                .withInterceptor(ForwardingChannelBuilder2::useTransportSecurity)
                 .buildFactory(builder);
 
         channelStaticMock.verify(FOR_ADDRESS, Mockito.times(0));
@@ -146,7 +141,7 @@ public void customChannelInitializer() {
         Mockito.verify(channelBuilderMock, Mockito.times(1))
                 .withOption(ChannelOption.ALLOCATOR, ByteBufAllocator.DEFAULT);
         Mockito.verify(channelBuilderMock, Mockito.times(1)).withOption(ChannelOption.TCP_NODELAY, Boolean.TRUE);
-        Mockito.verify(channelBuilderMock, Mockito.times(1)).enableFullStreamDecompression();
+        Mockito.verify(channelBuilderMock, Mockito.times(1)).useTransportSecurity();
     }
 
     @Test
@@ -186,12 +181,10 @@ public void customSslFactory() throws CertificateException, IOException {
 
             GrpcTransportBuilder builder = GrpcTransport.forHost(MOCKED_HOST, MOCKED_PORT, "/Root")
                     .withSecureConnection(baos.toByteArray())
-                    .withGrpcRetry(false)
-                    .withConnectTimeout(4, TimeUnit.SECONDS);
+                    .withGrpcRetry(false);
 
             ManagedChannelFactory factory = ChannelFactoryLoader.load().buildFactory(builder);
 
-            Assert.assertEquals(4000l, factory.getConnectTimeoutMs());
             Assert.assertSame(channelMock, factory.newManagedChannel(MOCKED_HOST, MOCKED_PORT, null));
 
         } finally {

core/src/test/java/tech/ydb/core/impl/pool/EndpointPoolTest.java

@@ -39,11 +39,11 @@ public class EndpointPoolTest {
 
     @Before
     public void setUp() throws IOException {
+        Mockito.when(socketFactory.createSocket()).thenReturn(socket);
+        Mockito.doNothing().when(socket).connect(Mockito.any(SocketAddress.class));
         mocks = MockitoAnnotations.openMocks(this);
         threadLocalStaticMock.when(ThreadLocalRandom::current).thenReturn(random);
         socketFactoryStaticMock.when(SocketFactory::getDefault).thenReturn(socketFactory);
-        Mockito.when(socketFactory.createSocket()).thenReturn(socket);
-        Mockito.doNothing().when(socket).connect(Mockito.any(SocketAddress.class));
     }
 
     @After

core/src/test/java/tech/ydb/core/impl/pool/GrpcChannelPoolTest.java

@@ -22,7 +22,6 @@ public class GrpcChannelPoolTest {
 
     @Before
     public void setUp() {
-        Mockito.when(factoryMock.getConnectTimeoutMs()).thenReturn(500l); // timeout for ready watcher
         Mockito.when(factoryMock.newManagedChannel(Mockito.any(), Mockito.anyInt(), Mockito.isNull()))
                 .then((args) -> ManagedChannelMock.good());
     }