Fix discovery losing connection problem

Losing address connections in discovery if restart happens faster than discovery task

Closes #404

Author: ArtDu

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## [Unreleased]
 
+### Bugfixes
+- Fix losing address connections in discovery if restart happens faster than discovery task ([#404](https://github.com/tarantool/cartridge-springdata/issues/404))
+
 ## [0.12.0] - 2023-06-09
 
 ### Internal and API changes

src/main/java/io/tarantool/driver/cluster/AbstractDiscoveryClusterAddressProvider.java

@@ -36,20 +36,13 @@ public AbstractDiscoveryClusterAddressProvider(TarantoolClusterDiscoveryConfig d
     protected void startDiscoveryTask() throws TarantoolClientException {
         Runnable discoveryTask = () -> {
             try {
-                Collection<TarantoolServerAddress> currentAddresses = this.addressesHolder.get();
-                Collection<TarantoolServerAddress> addresses = discoverAddresses();
-                setAddresses(addresses);
-
-                if (currentAddresses != null
-                    && addresses.size() != currentAddresses.size()
-                    || !addresses.equals(currentAddresses)) {
-                    this.refreshCallback.get().run();
-                }
+                setAddresses(discoverAddresses());
             } finally {
                 if (initLatch.getCount() > 0) {
                     initLatch.countDown();
                 }
             }
+            this.refreshCallback.get().run();
         };
 
         this.scheduledExecutorService.scheduleWithFixedDelay(

src/main/java/io/tarantool/driver/core/connection/AbstractTarantoolConnectionManager.java

@@ -17,6 +17,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -95,6 +96,25 @@ public boolean refresh() {
         return connectionMode.compareAndSet(ConnectionMode.OFF, ConnectionMode.PARTIAL);
     }
 
+    protected boolean areAddressesChanged() {
+        Collection<TarantoolServerAddress> addresses = getAddresses();
+        if (addresses == null) {
+            logger.debug("The list of server addresses is not defined");
+            return true;
+        }
+        return !connectionRegistry.keySet().equals(new HashSet<>(addresses));
+    }
+
+    protected boolean areConnectionsAlive() {
+        for (List<TarantoolConnection> connections : connectionRegistry.values()) {
+            int isAliveConnections = (int) connections.stream().filter(TarantoolConnection::isConnected).count();
+            if (isAliveConnections != config.getConnections()) {
+                return false;
+            }
+        }
+        return true;
+    }
+
     private CompletableFuture<TarantoolConnection> getConnectionInternal() {
         CompletableFuture<TarantoolConnection> result;
         ConnectionMode currentMode = connectionMode.get();

src/main/java/io/tarantool/driver/core/connection/TarantoolClusterConnectionManager.java

@@ -30,7 +30,13 @@ public TarantoolClusterConnectionManager(
         TarantoolClusterAddressProvider addressProvider) {
         super(config, connectionFactory, listeners);
         this.addressProvider = addressProvider;
-        this.addressProvider.setRefreshCallback(super::refresh);
+        this.addressProvider.setRefreshCallback(this::refreshIfConditionsChanged);
+    }
+
+    private void refreshIfConditionsChanged() {
+        if (areAddressesChanged() || !areConnectionsAlive()) {
+            super.refresh();
+        }
     }
 
     @Override

src/test/java/io/tarantool/driver/integration/ProxyTarantoolClientIT.java

@@ -1,7 +1,5 @@
 package io.tarantool.driver.integration;
 
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -12,12 +10,10 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.testcontainers.shaded.org.apache.commons.lang3.reflect.FieldUtils;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -32,6 +28,7 @@
 import io.tarantool.driver.api.TarantoolResult;
 import io.tarantool.driver.api.TarantoolServerAddress;
 import io.tarantool.driver.api.conditions.Conditions;
+import io.tarantool.driver.api.connection.TarantoolConnection;
 import io.tarantool.driver.api.metadata.TarantoolIndexMetadata;
 import io.tarantool.driver.api.metadata.TarantoolIndexType;
 import io.tarantool.driver.api.metadata.TarantoolMetadataOperations;
@@ -52,7 +49,6 @@
 import io.tarantool.driver.core.ClusterTarantoolTupleClient;
 import io.tarantool.driver.core.ProxyTarantoolTupleClient;
 import io.tarantool.driver.core.RetryingTarantoolTupleClient;
-import io.tarantool.driver.core.connection.TarantoolClusterConnectionManager;
 import io.tarantool.driver.core.tuple.TarantoolTupleImpl;
 import io.tarantool.driver.exceptions.TarantoolInternalException;
 import io.tarantool.driver.exceptions.TarantoolNoSuchProcedureException;
@@ -84,9 +80,7 @@ public class ProxyTarantoolClientIT extends SharedCartridgeContainer {
     private static final String PK_FIELD_NAME = "profile_id";
     private static final String BUCKET_ID_FIELD_NAME = "bucket_id";
     private static final int DEFAULT_TIMEOUT = 5 * 1000;
-    private static TestWrappedClusterAddressProvider addressProvider;
-    private static Method getAliveConnections;
-    private static TarantoolClusterConnectionManager connectionManager;
+    private static List<TarantoolConnection> connections = new ArrayList<>();
 
     @BeforeAll
     public static void setUp() throws Exception {
@@ -152,6 +146,10 @@ public static void initClient() {
                 .builder(10, thr -> thr instanceof TarantoolNoSuchProcedureException)
                 .withDelay(100)
                 .build());
+        client.getConnectionListeners().add(conn -> {
+            connections.add(conn);
+            return CompletableFuture.completedFuture(conn);
+        });
     }
 
     @Test
@@ -706,37 +704,7 @@ public void test_should_reconnect_ifReconnectIsInvoked() throws Exception {
         });
     }
 
-    private static int getAliveConnections() throws IllegalAccessException, NoSuchMethodException {
-        if (addressProvider == null) {
-            ClusterTarantoolTupleClient
-                innerClient = (ClusterTarantoolTupleClient) FieldUtils.readField(
-                FieldUtils.readField(client, "client", true),
-                "client", true);
-            connectionManager =
-                (TarantoolClusterConnectionManager) FieldUtils.readField(innerClient, "connectionManager", true);
-
-            getAliveConnections =
-                connectionManager.getClass().getSuperclass()
-                    .getDeclaredMethod("getAliveConnections", TarantoolServerAddress.class);
-            getAliveConnections.setAccessible(true);
-
-            addressProvider =
-                (TestWrappedClusterAddressProvider) FieldUtils.readField(innerClient, "addressProvider", true);
-        }
-
-        return getAliveConnectionsAux();
-    }
-
-    private static int getAliveConnectionsAux() {
-        AtomicInteger res = new AtomicInteger();
-        addressProvider.getAddresses().forEach(r -> {
-            try {
-                res.addAndGet(((List<?>) getAliveConnections.invoke(connectionManager, r)).size());
-            } catch (IllegalAccessException | InvocationTargetException e) {
-                throw new RuntimeException(e);
-            }
-        });
-
-        return res.get();
+    private static long getAliveConnections() throws IllegalAccessException, NoSuchMethodException {
+        return connections.stream().filter(TarantoolConnection::isConnected).count();
     }
 }