IGNITE-26580 Add test for stability of partition distribution

Author: sergey-chugunov-1985

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

@@ -106,14 +106,7 @@ public MdcAffinityBackupFilter(int dcsNum, int backups) {
 
     }
 
-    /**
-     * Defines a predicate which returns {@code true} if a node is acceptable for a backup
-     * or {@code false} otherwise.
-     * An acceptable node is the one that belongs to a data center that has some additional copies of partition to assign to.
-     * @param candidate A node that is a candidate for becoming a backup node for a partition.
-     * @param previouslySelected A list of primary/backup nodes already chosen for a partition.
-     *                           The primary is first.
-     */
+    /** {@inheritDoc} */
     @Override public boolean apply(ClusterNode candidate, List<ClusterNode> previouslySelected) {
         String candidateDcId = candidate.dataCenterId();
         int candDcCopiesAssigned = 0;

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

@@ -19,8 +19,12 @@
 
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import org.apache.ignite.IgniteCache;
 import org.apache.ignite.IgniteCheckedException;
@@ -41,7 +45,7 @@
  */
 public class MdcAffinityBackupFilterSelfTest extends GridCommonAbstractTest {
     /** */
-    private static final int PARTS_CNT = 1024;
+    private static final int PARTS_CNT = 512;
 
     /** */
     private int backups;
@@ -189,6 +193,29 @@ public void testAffinityFilterConfigurationValidation() throws Exception {
         }
     }
 
+    /**
+     * Verifies that distribution of partitions in one datacenter
+     * doesn't change if nodes leave in another or if the other DC goes down completely.
+     * In other words, no rebalance is triggered in one dc if some topology changes happen in another.
+     *
+     * @throws Exception If failed.
+     */
+    @Test
+    public void testNoRebalanceInOneDcIfTopologyChangesInAnother() throws Exception {
+        dcIds = new String[] {"DC_0", "DC_1"};
+        int nodesPerDc = 3;
+        backups = 3;
+        filter = new MdcAffinityBackupFilter(dcIds.length, backups);
+        Predicate<ClusterNode> dc1NodesFilter = node -> "DC_1".equals(node.dataCenterId());
+
+        IgniteEx srv = startClusterAcrossDataCenters(dcIds, nodesPerDc);
+        awaitPartitionMapExchange();
+        Map<Integer, Set<UUID>> oldDistribution = affinityForPartitions(srv.getOrCreateCache(DEFAULT_CACHE_NAME), dc1NodesFilter);
+
+        for (int srvIdx = 0; srvIdx < 3; srvIdx++)
+            oldDistribution = verifyNoRebalancing(srvIdx, srv, oldDistribution, dc1NodesFilter);
+    }
+
     /** Starts specified number of nodes in each DC. */
     private IgniteEx startClusterAcrossDataCenters(String[] dcIds, int nodesPerDc) throws Exception {
         int nodeIdx = 0;
@@ -204,6 +231,51 @@ private IgniteEx startClusterAcrossDataCenters(String[] dcIds, int nodesPerDc) t
         return lastNode;
     }
 
+    /** */
+    private Map<Integer, Set<UUID>> verifyNoRebalancing(int srvIdx, IgniteEx srv, Map<Integer, Set<UUID>> oldDistribution, Predicate<ClusterNode> dc1NodesFilter) throws InterruptedException {
+        stopGrid(srvIdx);
+        awaitPartitionMapExchange();
+        IgniteCache<Integer, Integer> cache = srv.getOrCreateCache(DEFAULT_CACHE_NAME);
+
+        Map<Integer, Set<UUID>> newDistribution = affinityForPartitions(cache, dc1NodesFilter);
+
+        assertEquals(
+            String.format("Affinity distribution changed after server node %d was stopped", srvIdx),
+            oldDistribution,
+            newDistribution);
+
+        return newDistribution;
+    }
+
+    /** */
+    private Map<Integer, Set<UUID>> affinityForPartitions(IgniteCache<Integer, Integer> cache, Predicate<ClusterNode> dcFilter) {
+        Map<Integer, Set<UUID>> result = new HashMap<>(PARTS_CNT);
+        Affinity<Integer> aff = affinity(cache);
+
+        for (int i = 0; i < PARTS_CNT; i++) {
+            int j = i;
+
+            aff.mapKeyToPrimaryAndBackups(i)
+                .stream()
+                .filter(dcFilter)
+                .forEach(
+                    node -> result.compute(j,
+                        (k, v) -> {
+                            if (v == null) {
+                                Set<UUID> s = new HashSet<>();
+                                s.add(node.id());
+                                return s;
+                            }
+                            else {
+                                v.add(node.id());
+                                return v;
+                            }
+                        }));
+        }
+
+        return result;
+    }
+
     /**
      * Checks that copies of each partition are distributed evenly across data centers and copies are spread evenly across nodes.
      */