[ignite] branch master updated: IGNITE-17217 Java thin: reload addresses from ClientAddressFinder on connection loss (#10110)

[pt...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

ptupitsyn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git


The following commit(s) were added to refs/heads/master by this push:
     new 85f54de74a4 IGNITE-17217 Java thin: reload addresses from ClientAddressFinder on connection loss (#10110)
85f54de74a4 is described below

commit 85f54de74a4fd27eb461db255e7be6f0405d6b12
Author: wkhappy1 <54...@users.noreply.github.com>
AuthorDate: Tue Jun 28 19:59:48 2022 +0800

    IGNITE-17217 Java thin: reload addresses from ClientAddressFinder on connection loss (#10110)
    
    Server address can change after restart (applies to Kubernetes, among other things). Refresh addresses from AddressFinder on connection loss.
---
 .../internal/client/thin/ReliableChannel.java      | 13 +++++-
 .../discovery/TestClusterClientConnection.java     | 47 ++++++++++++++++++++++
 2 files changed, 59 insertions(+), 1 deletion(-)

diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
index f9529349d0e..50a1dc5213e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
@@ -529,10 +529,21 @@ final class ReliableChannel implements AutoCloseable {
         rollCurrentChannel(hld);
 
         // For partiton awareness it's already initializing asynchronously in #onTopologyChanged.
-        if (scheduledChannelsReinit.get() && !partitionAwarenessEnabled)
+        if (addressFinderAddressesChanged() || (scheduledChannelsReinit.get() && !partitionAwarenessEnabled))
             channelsInit();
     }
 
+    /**
+     * Checks whether addressFinder returns a different set of addresses.
+     */
+    private boolean addressFinderAddressesChanged() {
+        if (clientCfg.getAddressesFinder() == null)
+            return false;
+
+        String[] hostAddrs = clientCfg.getAddressesFinder().getAddresses();
+        return !Arrays.equals(hostAddrs, prevHostAddrs);
+    }
+
     /**
      * Asynchronously try to establish a connection to all configured servers.
      */
diff --git a/modules/kubernetes/src/test/java/org/apache/ignite/kubernetes/discovery/TestClusterClientConnection.java b/modules/kubernetes/src/test/java/org/apache/ignite/kubernetes/discovery/TestClusterClientConnection.java
index 839e38c27a6..b70c1b88b4d 100644
--- a/modules/kubernetes/src/test/java/org/apache/ignite/kubernetes/discovery/TestClusterClientConnection.java
+++ b/modules/kubernetes/src/test/java/org/apache/ignite/kubernetes/discovery/TestClusterClientConnection.java
@@ -22,11 +22,13 @@ import org.apache.ignite.client.ClientCache;
 import org.apache.ignite.client.IgniteClient;
 import org.apache.ignite.client.ThinClientKubernetesAddressFinder;
 import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.ClientConnectorConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.IgniteEx;
 import org.junit.Test;
 
 /** Test that thin client connects to cluster with {@link ThinClientKubernetesAddressFinder}. */
+@SuppressWarnings({"rawtypes", "unchecked"})
 public class TestClusterClientConnection extends KubernetesDiscoveryAbstractTest {
     /** */
     @Test
@@ -49,4 +51,49 @@ public class TestClusterClientConnection extends KubernetesDiscoveryAbstractTest
         cache.put(1, 2);
         assertEquals(2, cache.get(1));
     }
+
+    /**
+     * Tests that client updates addresses from address finder and can connect to the cluster after pod restart.
+     */
+    @Test
+    public void testClientReConnectsToClusterAfterPodIpChange() throws Exception {
+        mockServerResponse();
+
+        IgniteConfiguration cfg = getConfiguration(getTestIgniteInstanceName(), false);
+
+        IgniteEx crd = startGrid(cfg);
+        String crdAddr = crd.localNode().addresses().iterator().next();
+
+        mockServerResponse(crdAddr);
+
+        ClientConfiguration ccfg = new ClientConfiguration();
+        ccfg.setAddressesFinder(new ThinClientKubernetesAddressFinder(prepareConfiguration()));
+        IgniteClient client = Ignition.startClient(ccfg);
+
+        ClientCache cache = client.createCache("cache");
+        cache.put(1, 2);
+        assertEquals(2, cache.get(1));
+
+        // Stop node and change port => still can connect.
+        Ignition.stop(crd.name(), true);
+        int newPort = 10801;
+        ccfg.setAddressesFinder(new ThinClientKubernetesAddressFinder(prepareConfiguration(), newPort));
+        mockServerResponse(5, crdAddr);
+
+        cfg = getConfiguration(getTestIgniteInstanceName(), false);
+        cfg.setClientConnectorConfiguration(new ClientConnectorConfiguration().setPort(newPort));
+        startGrid(cfg);
+
+        try {
+            cache = client.getOrCreateCache("cache");
+            cache.put(1, 3);
+        }
+        catch (Exception ignored) {
+            // No-op.
+        }
+
+        cache = client.getOrCreateCache("cache");
+        cache.put(1, 3);
+        assertEquals(3, cache.get(1));
+    }
 }