IGNITE-26580 Finish test, fix bug in an MDC filter

Author: sergey-chugunov-1985

modules/core/src/main/java/org/apache/ignite/cache/affinity/rendezvous/MdcAffinityBackupFilter.java

@@ -20,7 +20,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.lang.IgniteBiPredicate;
 
@@ -50,8 +49,11 @@ public MdcAffinityBackupFilter(int dcsCount, int backups) {
 
     /** {@inheritDoc} */
     @Override public boolean apply(ClusterNode node, List<ClusterNode> list) {
-        if (list.size() == 1) //account for primary node which is assigned beforehand
+        if (list.size() == 1) { //list contains only primary node, thus we started new assignment round.
+            partsDistrMap.replaceAll((e, v) -> -1);
+
             partsDistrMap.put(list.get(0).dataCenterId(), 1);
+        }
 
         String candidateDcId = node.dataCenterId();
         Integer candDcPartsCopies = partsDistrMap.get(candidateDcId);
@@ -72,11 +74,6 @@ public MdcAffinityBackupFilter(int dcsCount, int backups) {
             }
         }
 
-        Optional<Integer> sum = partsDistrMap.values().stream().reduce(Integer::sum);
-
-        if (sum.isPresent() && sum.get() == primaryAndBackups)
-            partsDistrMap.replaceAll((e, v) -> -1);
-
         return res;
     }
 }

modules/core/src/test/java/org/apache/ignite/cache/affinity/rendezvous/MdcAffinityBackupFilterSelfTest.java

@@ -18,13 +18,15 @@
 package org.apache.ignite.cache.affinity.rendezvous;
 
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteSystemProperties;
 import org.apache.ignite.cache.affinity.Affinity;
-import org.apache.ignite.cache.affinity.AffinityFunction;
 import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.configuration.CacheConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
@@ -33,48 +35,174 @@
 /** */
 public class MdcAffinityBackupFilterSelfTest extends GridCommonAbstractTest {
     /** */
-    private static final String DC_0_ID = "DC_0";
+    private static final int PARTS_CNT = 1024;
 
     /** */
-    private static final String DC_1_ID = "DC_1";
+    private int backups;
+
+    /** */
+    private String[] dcIds;
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+        cfg.setCacheConfiguration(defaultCacheConfiguration()
+            .setBackups(backups)
+            .setAffinity(
+                new RendezvousAffinityFunction(false, PARTS_CNT)
+                    .setAffinityBackupFilter(new MdcAffinityBackupFilter(dcIds.length, backups))));
+
+        return cfg;
+    }
 
     /** {@inheritDoc} */
     @Override protected IgniteConfiguration optimize(IgniteConfiguration cfg) throws IgniteCheckedException {
         return super.optimize(cfg).setIncludeProperties((String[])null);
     }
 
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        super.afterTest();
+
+        stopAllGrids();
+    }
+
     /**
-     * Verifies that partition copies are assigned evenly to all data centers.
+     * Verifies that partition copies are assigned evenly across a cluster in two data centers.
+     * <p/>
+     * When a node from one data center is stopped, partition distribution is that data center should stay uniform.
      *
      * @throws Exception If failed.
      */
     @Test
-    public void testBasicDistribution() throws Exception {
-        System.setProperty(IgniteSystemProperties.IGNITE_DATA_CENTER_ID, DC_0_ID);
-        startGrids(2);
+    public void test2DcDistribution() throws Exception {
+        dcIds = new String[] {"DC_0", "DC_1"};
+        int nodesPerDc = 4;
+        backups = 3;
 
-        System.setProperty(IgniteSystemProperties.IGNITE_DATA_CENTER_ID, DC_1_ID);
-        startGrid(2);
-        IgniteEx srv = startGrid(3);
+        IgniteEx srv = startClusterAcrossDataCenters(dcIds, nodesPerDc);
 
-        CacheConfiguration ccfg = defaultCacheConfiguration()
-            .setBackups(3)
-            .setAffinity(
-                new RendezvousAffinityFunction(false, 4)
-                    .setAffinityBackupFilter(new MdcAffinityBackupFilter(2, 3)));
+        checkPartitionsDistribution(srv, dcIds, nodesPerDc, backups);
+
+        //stopping one node in DC_1 should not compromise distribution as there are additional nodes in the same DC
+        stopGrid(5);
+
+        awaitPartitionMapExchange();
 
-        IgniteCache cache = srv.getOrCreateCache(ccfg);
+        checkPartitionsDistribution(srv, dcIds, nodesPerDc, backups);
 
-        checkPartitions(srv, cache, 3);
+        //stopping another node in DC_1 should not compromise distribution as well
+        stopGrid(6);
+
+        awaitPartitionMapExchange();
+
+        checkPartitionsDistribution(srv, dcIds, nodesPerDc, backups);
     }
 
-    /** */
-    private void checkPartitions(IgniteEx srv, IgniteCache cache, int backups) throws IgniteCheckedException {) {
+    /**
+     * Verifies that partition copies are assigned evenly across a cluster in three data centers.
+     *
+     * @throws Exception If failed.
+     */
+    @Test
+    public void test3DcDistribution() throws Exception {
+        dcIds = new String[] {"DC_0", "DC_1", "DC_2"};
+        int nodesPerDc = 2;
+        backups = 5;
+
+        IgniteEx srv = startClusterAcrossDataCenters(dcIds, 2);
+
+        checkPartitionsDistribution(srv, dcIds, nodesPerDc, backups);
+    }
+
+    /** Starts specified number of nodes in each DC. */
+    private IgniteEx startClusterAcrossDataCenters(String[] dcIds, int nodesPerDc) throws Exception {
+        int nodeIdx = 0;
+        IgniteEx lastNode = null;
+
+        for (String dcId : dcIds) {
+            System.setProperty(IgniteSystemProperties.IGNITE_DATA_CENTER_ID, dcId);
+
+            for (int i = 0; i < nodesPerDc; i++)
+                lastNode = startGrid(nodeIdx++);
+        }
+
+        return lastNode;
+    }
+
+    /**
+     * Checks that copies of each partition are distributed evenly across data centers.
+     * If checkUniformity is true, then also checks that number of copies per node is close enough to ideal value (less than 10% variation).
+     */
+    private void checkPartitionsDistribution(
+        IgniteEx srv,
+        String[] dcIds,
+        int nodesPerDc,
+        int backups
+    ) {
+        IgniteCache cache = srv.getOrCreateCache(DEFAULT_CACHE_NAME);
+
         int partCnt = cacheConfiguration(srv.configuration(), cache.getName()).getAffinity().partitions();
         Affinity aff = affinity(cache);
+        int expectedCopiesPerNode = (backups + 1) / dcIds.length;
+
+        Map<ClusterNode, Integer> overallCopiesPerNode = new HashMap<>();
+        int[] copiesPerNode = new int[dcIds.length * nodesPerDc];
 
         for (int i = 0; i < partCnt; i++) {
+            int[] partCopiesPerDc = new int[dcIds.length];
+
             Collection<ClusterNode> nodes = aff.mapKeyToPrimaryAndBackups(i);
+
+            //calculate actual number of copies in each data center
+            //aggregate copies per each node
+            for (ClusterNode node : nodes) {
+                copiesPerNode[(int)(node.order() - 1)]++;
+
+                overallCopiesPerNode.compute(node, (k, v) -> {
+                    if (v == null)
+                        return 1;
+                    else
+                        return v + 1;
+                });
+
+                for (int j = 0; j < dcIds.length; j++) {
+                    if (node.dataCenterId().equals(dcIds[j])) {
+                        partCopiesPerDc[j]++;
+                        break;
+                    }
+                }
+            }
+
+            //check that each data center has expected number of copies
+            for (int dcIdx = 0; dcIdx < dcIds.length; dcIdx++) {
+                assertEquals(String.format("Unexpected number of copies of partition %d in data center %s", i, dcIds[dcIdx]),
+                    expectedCopiesPerNode,
+                    partCopiesPerDc[dcIdx]);
+            }
+        }
+
+        for (String dcId : dcIds) {
+            long nodesInDc = overallCopiesPerNode.entrySet().stream().filter(e -> e.getKey().dataCenterId().equals(dcId)).count();
+
+            double idealCopiesPerNode = (double)((PARTS_CNT * (backups + 1)) / (dcIds.length * nodesInDc));
+
+            List<Integer> numOfCopiesPerNode = overallCopiesPerNode.entrySet().stream()
+                .filter(e -> e.getKey().dataCenterId().equals(dcId)).map(Map.Entry::getValue).collect(Collectors.toList());
+
+            for (int copiesOnNode : numOfCopiesPerNode) {
+                double deviation = (Math.abs(copiesOnNode - idealCopiesPerNode) / idealCopiesPerNode);
+
+                assertTrue(
+                    String.format("Too big deviation from ideal distribution: partitions assigned = %d, " +
+                            "ideal partitions assigned = %d, deviation = %d",
+                        copiesOnNode,
+                        (int)idealCopiesPerNode,
+                        (int)(deviation * 100)
+                    ),
+                    deviation < 0.1);
+            }
         }
     }
 }