IGNITE-26909 Thin client optimizations for MultiDC

Author: alex-plekhanov

modules/core/src/test/java/org/apache/ignite/client/ReliabilityTest.java

@@ -356,7 +356,7 @@ public void testRetryPolicyConvertOpAllOperationsSupported() {
 
         String nullOpsNames = nullOps.stream().map(Enum::name).collect(Collectors.joining(", "));
 
-        long expectedNullCnt = 23;
+        long expectedNullCnt = 24;
 
         String msg = nullOps.size()
                 + " operation codes do not have public equivalent. When adding new codes, update ClientOperationType too. Missing ops: "

modules/core/src/test/java/org/apache/ignite/internal/client/thin/ThinClientPartitionAwarenessMultiDcTest.java

@@ -32,6 +32,8 @@
 import org.apache.ignite.client.ClientTransaction;
 import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.ClientConnectorConfiguration;
+import org.apache.ignite.configuration.ThinClientConfiguration;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.IgniteInterruptedCheckedException;
 import org.apache.ignite.internal.util.typedef.F;
@@ -64,6 +66,9 @@ public class ThinClientPartitionAwarenessMultiDcTest extends ThinClientAbstractP
     /** */
     private IgniteEx startGrid(int idx, String dcId) throws Exception {
         return startGrid(getConfiguration(getTestIgniteInstanceName(idx))
+            .setClientConnectorConfiguration(new ClientConnectorConfiguration()
+                .setThinClientConfiguration(new ThinClientConfiguration()
+                    .setMaxActiveComputeTasksPerConnection(1)))
             .setUserAttributes(F.asMap(IgniteSystemProperties.IGNITE_DATA_CENTER_ID, dcId)));
     }
 
@@ -146,9 +151,17 @@ private void checkDcAwarePaRequests(
 
                 assertOpOnChannel(channels[primaryNodeIdx], ClientOperation.CACHE_PUT);
 
+                cache.putAsync(key, 0).get();
+
+                assertOpOnChannel(channels[primaryNodeIdx], ClientOperation.CACHE_PUT);
+
                 cache.get(key);
 
                 assertOpOnChannel(channels[primaryNodeIdx], ClientOperation.CACHE_GET);
+
+                cache.getAsync(key).get();
+
+                assertOpOnChannel(channels[primaryNodeIdx], ClientOperation.CACHE_GET);
             }
         }
 
@@ -158,9 +171,13 @@ private void checkDcAwarePaRequests(
             for (Integer key : keys) {
                 int primaryNodeIdx = getTestIgniteInstanceIndex(primaryNode.name());
 
+                // If primary in another DC, write requests always sent to primary node.
                 cache.put(key, 0);
 
-                // If primary in another DC, write requests always sent to primary node.
+                assertOpOnChannel(channels[primaryNodeIdx], ClientOperation.CACHE_PUT);
+
+                cache.putAsync(key, 0).get();
+
                 assertOpOnChannel(channels[primaryNodeIdx], ClientOperation.CACHE_PUT);
 
                 int backupNodeIdx = getTestIgniteInstanceIndex(backupNode(key, cacheName).name());
@@ -172,6 +189,10 @@ private void checkDcAwarePaRequests(
 
                 assertOpOnChannel(channels[expReqToBackup ? backupNodeIdx : primaryNodeIdx], ClientOperation.CACHE_GET);
 
+                cache.getAsync(key).get();
+
+                assertOpOnChannel(channels[expReqToBackup ? backupNodeIdx : primaryNodeIdx], ClientOperation.CACHE_GET);
+
                 cache.containsKey(key);
 
                 assertOpOnChannel(channels[expReqToBackup ? backupNodeIdx : primaryNodeIdx], ClientOperation.CACHE_CONTAINS_KEY);
@@ -219,19 +240,15 @@ private void checkDcAwareNonPaRequests(String cacheName, String dcId, int... exp
         for (int i = 0; i < 10; i++) {
             cache.query(new ScanQuery<>()).getAll();
 
-            int channelIdx = nextOpChannelIdx();
-
-            assertTrue(F.contains(expNodeIdxs, channelIdx));
-
-            assertTrue("Ops queue not empty: " + opsQueue, F.isEmpty(opsQueue));
+            assertExpectedChannel(expNodeIdxs);
 
             client.cluster().state();
 
-            channelIdx = nextOpChannelIdx();
+            assertExpectedChannel(expNodeIdxs);
 
-            assertTrue(F.contains(expNodeIdxs, channelIdx));
+            client.compute().executeAsync2(TestTask.class.getName(), null).get();
 
-            assertTrue("Ops queue not empty: " + opsQueue, F.isEmpty(opsQueue));
+            assertExpectedChannel(expNodeIdxs);
 
             try (ClientTransaction tx = client.transactions().txStart()) {
                 cache.put(ThreadLocalRandom.current().nextInt(10), 0);
@@ -241,7 +258,7 @@ private void checkDcAwareNonPaRequests(String cacheName, String dcId, int... exp
             }
 
             while (true) {
-                channelIdx = nextOpChannelIdx();
+                int channelIdx = nextOpChannelIdx();
 
                 if (channelIdx < 0)
                     break;
@@ -250,4 +267,13 @@ private void checkDcAwareNonPaRequests(String cacheName, String dcId, int... exp
             }
         }
     }
+
+    /** */
+    private void assertExpectedChannel(int... expChannelIdxs) {
+        int channelIdx = nextOpChannelIdx();
+
+        assertTrue(F.contains(expChannelIdxs, channelIdx));
+
+        assertTrue("Ops queue not empty: " + opsQueue, F.isEmpty(opsQueue));
+    }
 }