Propagate client thumbstone when a client creation fails

nicktorwald · nicktorwald · commit 868b377eb1e6 · 2019-09-11T18:47:41.000+07:00
Closes: #30
diff --git a/src/main/java/org/tarantool/TarantoolClientImpl.java b/src/main/java/org/tarantool/TarantoolClientImpl.java
@@ -28,9 +28,6 @@
 
 public class TarantoolClientImpl extends TarantoolBase<Future<?>> implements TarantoolClient {
 
-    public static final CommunicationException NOT_INIT_EXCEPTION
-        = new CommunicationException("Not connected, initializing connection");
-
     protected TarantoolClientConfig config;
     protected long operationTimeout;
 
@@ -101,7 +98,6 @@ public TarantoolClientImpl(SocketChannelProvider socketProvider, TarantoolClient
     }
 
     private void initClient(SocketChannelProvider socketProvider, TarantoolClientConfig config) {
-        this.thumbstone = NOT_INIT_EXCEPTION;
         this.config = config;
         this.initialRequestSize = config.defaultRequestSize;
         this.operationTimeout = config.operationExpiryTimeMillis;
@@ -130,8 +126,8 @@ private void startConnector(long initTimeoutMillis) {
                 CommunicationException e = new CommunicationException(
                     initTimeoutMillis +
                         "ms is exceeded when waiting for client initialization. " +
-                        "You could configure init timeout in TarantoolConfig"
-                );
+                        "You could configure init timeout in TarantoolConfig",
+                    thumbstone);
 
                 close(e);
                 throw e;
@@ -147,7 +143,7 @@ protected void reconnect(Throwable lastError) {
         int retryNumber = 0;
         while (!Thread.currentThread().isInterrupted()) {
             try {
-                channel = socketProvider.get(retryNumber++, lastError == NOT_INIT_EXCEPTION ? null : lastError);
+                channel = socketProvider.get(retryNumber++, lastError);
             } catch (Exception e) {
                 closeChannel(channel);
                 lastError = e;
diff --git a/src/test/java/org/tarantool/ClientReconnectClusterIT.java b/src/test/java/org/tarantool/ClientReconnectClusterIT.java
@@ -2,6 +2,8 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.tarantool.TestUtils.findCause;
 import static org.tarantool.TestUtils.makeDefaultClusterClientConfig;
 import static org.tarantool.TestUtils.makeDiscoveryFunction;
 
@@ -393,6 +395,53 @@ void testDelayFunctionResultFetch() {
         expectDisconnected(client, spaceId, pkId);
     }
 
+    @Test
+    void testRoundRobinSocketProviderRefused() {
+        stopInstancesAndAwait(SRV1);
+        stopInstancesAndAwait(SRV2);
+        stopInstancesAndAwait(SRV3);
+
+        RuntimeException error = new RuntimeException("Fake error");
+        TarantoolClusterClientConfig config = makeDefaultClusterClientConfig();
+        config.initTimeoutMillis = 1000;
+        Throwable exception = assertThrows(
+            CommunicationException.class,
+            () -> {
+                new TarantoolClusterClient(
+                    config,
+                    TestUtils.wrapByErroredProvider(new RoundRobinSocketProviderImpl(
+                        "localhost:" + PORTS[0],
+                        "localhost:" + PORTS[1],
+                        "localhost:" + PORTS[2]
+                    ), error)
+                );
+            }
+        );
+        assertTrue(findCause(exception, error));
+    }
+
+    @Test
+    void testRoundRobinSocketProviderRefusedAfterConnect() {
+        final TarantoolClientImpl client = makeClusterClient(
+            "localhost:" + PORTS[0],
+            "localhost:" + PORTS[1],
+            "localhost:" + PORTS[2]
+        );
+
+        client.ping();
+        stopInstancesAndAwait(SRV1);
+
+        client.ping();
+        stopInstancesAndAwait(SRV2);
+
+        client.ping();
+        stopInstancesAndAwait(SRV3);
+
+        CommunicationException exception = assertThrows(CommunicationException.class, client::ping);
+        Throwable origin = exception.getCause();
+        assertEquals(origin, client.getThumbstone());
+    }
+
     private void tryAwait(CyclicBarrier barrier) {
         try {
             barrier.await(6000, TimeUnit.MILLISECONDS);
diff --git a/src/test/java/org/tarantool/ClientReconnectIT.java b/src/test/java/org/tarantool/ClientReconnectIT.java
@@ -7,6 +7,7 @@
 import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
+import static org.tarantool.TestUtils.findCause;
 import static org.tarantool.TestUtils.makeDefaultClientConfig;
 import static org.tarantool.TestUtils.makeTestClient;
 
@@ -27,6 +28,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.atomic.AtomicReferenceArray;
 import java.util.concurrent.locks.LockSupport;
 
@@ -412,13 +414,86 @@ public void testReconnectWrongAuth() throws Exception {
         client.close();
     }
 
-    private TestSocketChannelProvider makeZeroLingerProvider() {
+    @Test
+    void testFirstConnectionRefused() {
+        RuntimeException error = new RuntimeException("Fake error");
+        TarantoolClientConfig config = makeDefaultClientConfig();
+        config.initTimeoutMillis = 100;
+        Throwable exception = assertThrows(
+            CommunicationException.class,
+            () -> new TarantoolClientImpl(makeErroredProvider(error), config)
+        );
+        assertTrue(findCause(exception, error));
+    }
+
+    @Test
+    void testConnectionRefusedAfterConnect() {
+        TarantoolClientImpl client = new TarantoolClientImpl(makeErroredProvider(null), makeDefaultClientConfig());
+        client.ping();
+
+        testHelper.stopInstance();
+        CommunicationException exception = assertThrows(CommunicationException.class, client::ping);
+
+        Throwable origin = exception.getCause();
+        assertEquals(origin, client.getThumbstone());
+
+        testHelper.startInstance();
+    }
+
+    @Test
+    void testSingleSocketProviderRefused() {
+        TarantoolClientConfig config = makeDefaultClientConfig();
+        RuntimeException error = new RuntimeException("Fake error");
+        config.initTimeoutMillis = 1000;
+
+        SingleSocketChannelProviderImpl socketProvider = new SingleSocketChannelProviderImpl("localhost:3301");
+
+        testHelper.stopInstance();
+        Throwable exception = assertThrows(
+            CommunicationException.class,
+            () -> new TarantoolClientImpl(TestUtils.wrapByErroredProvider(socketProvider, error), config)
+        );
+        testHelper.startInstance();
+        assertTrue(findCause(exception, error));
+    }
+
+    @Test
+    void testSingleSocketProviderRefusedAfterConnect() {
+        TarantoolClientImpl client = new TarantoolClientImpl(socketChannelProvider, makeDefaultClientConfig());
+
+        client.ping();
+        testHelper.stopInstance();
+
+        CommunicationException exception = assertThrows(CommunicationException.class, client::ping);
+        Throwable origin = exception.getCause();
+        assertEquals(origin, client.getThumbstone());
+
+        testHelper.startInstance();
+    }
+
+    private SocketChannelProvider makeZeroLingerProvider() {
         return new TestSocketChannelProvider(
             TarantoolTestHelper.HOST, TarantoolTestHelper.PORT, RESTART_TIMEOUT
         ).setSoLinger(0);
     }
 
-    TarantoolClient makeClient(SocketChannelProvider provider) {
+    private SocketChannelProvider makeErroredProvider(RuntimeException error) {
+        return new SocketChannelProvider() {
+            private final SocketChannelProvider delegate = makeZeroLingerProvider();
+            private AtomicReference<RuntimeException> errorReference = new AtomicReference<>(error);
+
+            @Override
+            public SocketChannel get(int retryNumber, Throwable lastError) {
+                RuntimeException rawError = errorReference.get();
+                if (rawError != null) {
+                    throw rawError;
+                }
+                return delegate.get(retryNumber, lastError);
+            }
+        };
+    }
+
+    private TarantoolClient makeClient(SocketChannelProvider provider) {
         return new TarantoolClientImpl(provider, makeDefaultClientConfig());
     }
 
diff --git a/src/test/java/org/tarantool/TestUtils.java b/src/test/java/org/tarantool/TestUtils.java
@@ -5,6 +5,7 @@
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.net.SocketAddress;
+import java.nio.channels.SocketChannel;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
@@ -290,4 +291,50 @@ public static TarantoolClientConfig makeDefaultClientConfig() {
         return config;
     }
 
+    /**
+     * Wraps a socket channel provider
+     * {@link SocketChannelProvider#get(int, Throwable)} method.
+     * When an error is raised the wrapper substitutes
+     * this error by the predefined one. The original value is
+     * still accessible as a cause of the injected error.
+     *
+     * @param provider provider to be wrapped
+     * @param error error to be thrown instead of original
+     *
+     * @return wrapped provider
+     */
+    public static SocketChannelProvider wrapByErroredProvider(SocketChannelProvider provider, RuntimeException error) {
+        return new SocketChannelProvider() {
+            private final SocketChannelProvider delegate = provider;
+
+            @Override
+            public SocketChannel get(int retryNumber, Throwable lastError) {
+                try {
+                    return delegate.get(retryNumber, lastError);
+                } catch (Exception e) {
+                    error.initCause(e);
+                    throw error;
+                }
+            }
+        };
+    }
+
+    /**
+     * Searches recursively the given cause for a root error.
+     *
+     * @param error root error
+     * @param cause cause to be found
+     *
+     * @return {@literal true} if cause is found within a cause chain
+     */
+    public static boolean findCause(Throwable error, Throwable cause) {
+        while (error.getCause() != null) {
+            error = error.getCause();
+            if (cause.equals(error)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
 }
