IGNITE-26580 Add test for baseline

Author: sergey-chugunov-1985

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

@@ -26,17 +26,21 @@
 import java.util.UUID;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 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.cluster.ClusterNode;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.lang.IgniteBiPredicate;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.Test;
 
+import static org.apache.ignite.cluster.ClusterState.ACTIVE;
 import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
 
 /**
@@ -54,7 +58,10 @@ public class MdcAffinityBackupFilterSelfTest extends GridCommonAbstractTest {
     private String[] dcIds;
 
     /** */
-    private IgniteBiPredicate<ClusterNode, List<ClusterNode>> filter;
+    private IgniteBiPredicate<ClusterNode, List<ClusterNode>> backupFilter;
+
+    /** */
+    private boolean persistenceEnabled;
 
     /** {@inheritDoc} */
     @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
@@ -64,7 +71,17 @@ public class MdcAffinityBackupFilterSelfTest extends GridCommonAbstractTest {
             .setBackups(backups)
             .setAffinity(
                 new RendezvousAffinityFunction(false, PARTS_CNT)
-                    .setAffinityBackupFilter(filter)));
+                    .setAffinityBackupFilter(backupFilter)));
+
+        if (persistenceEnabled) {
+            cfg.setDataStorageConfiguration(
+                new DataStorageConfiguration()
+                    .setDefaultDataRegionConfiguration(
+                        new DataRegionConfiguration()
+                            .setPersistenceEnabled(true)
+                    )
+            );
+        }
 
         return cfg;
     }
@@ -74,6 +91,8 @@ public class MdcAffinityBackupFilterSelfTest extends GridCommonAbstractTest {
         super.afterTest();
 
         stopAllGrids();
+
+        cleanPersistenceDir();
     }
 
     /**
@@ -103,6 +122,61 @@ public void testUniformNumberOfPartitionCopiesPerDcIsEnforced() {
             "recommended values are 5 and 8");
     }
 
+    /**
+     * Verifies that filter doesn't assing additional copies if baseline topology is not changed,
+     * and it could lead to breaking of the guarantee `at least one copy of partition per datacenter`.
+     * <p/>
+     * Verifies that after baseline topology change, guarantee is restored.
+     *
+     * @throws Exception If failed.
+     */
+    @Test
+    public void testMdcFilterWithBaselineTopology() throws Exception {
+        persistenceEnabled = true;
+        dcIds = new String[] {"DC_0", "DC_1"};
+        int nodesPerDc = 2;
+        backups = 1;
+        backupFilter = new MdcAffinityBackupFilter(dcIds.length, backups);
+
+        IgniteEx srv = startClusterAcrossDataCenters(dcIds, nodesPerDc);
+
+        srv.cluster().state(ACTIVE);
+
+        IgniteCache cache = srv.getOrCreateCache(DEFAULT_CACHE_NAME);
+
+        Map<Integer, Set<UUID>> dc1OldAffDistr = affinityForPartitions(cache, node -> node.dataCenterId().equals(dcIds[1]));
+
+        UUID srv0Id = grid(0).localNode().id();
+
+        Map<Integer, Set<UUID>> dc0AffDistr = affinityForPartitions(cache, node -> node.dataCenterId().equals(dcIds[0]));
+
+        Set<Integer> srv0Partitions = dc0AffDistr
+            .entrySet().stream()
+            .filter(entry -> entry.getValue().contains(srv0Id))
+            .map(Map.Entry::getKey)
+            .collect(Collectors.toSet());
+
+        stopGrid(0);
+
+        dc0AffDistr = affinityForPartitions(cache, node -> node.dataCenterId().equals(dcIds[0]));
+
+        // Partitions from stopped node are not assigned to any node without baseline topology change.
+        assertFalse(dc0AffDistr.keySet().containsAll(srv0Partitions));
+
+        Map<Integer, Set<UUID>> dc1NewAffDistr = affinityForPartitions(cache, node -> node.dataCenterId().equals(dcIds[1]));
+
+        // Assignment of partitions in remote DC has not changed.
+        assertEquals(dc1OldAffDistr, dc1NewAffDistr);
+
+        srv.cluster().setBaselineTopology(srv.cluster().topologyVersion());
+        awaitPartitionMapExchange();
+
+        Set<Integer> allParts = Stream.iterate(0, i -> i + 1).limit(PARTS_CNT).collect(Collectors.toSet());
+        dc0AffDistr = affinityForPartitions(cache, node -> node.dataCenterId().equals(dcIds[0]));
+        // After baseline topology change, guarantee is restored.
+        assertTrue(allParts.containsAll(dc0AffDistr.keySet()));
+    }
+
     /**
      * Verifies that partition copies are assigned evenly across a cluster in two data centers.
      * <p/>
@@ -115,25 +189,25 @@ public void test2DcDistribution() throws Exception {
         dcIds = new String[] {"DC_0", "DC_1"};
         int nodesPerDc = 4;
         backups = 3;
-        filter = new MdcAffinityBackupFilter(dcIds.length, backups);
+        backupFilter = new MdcAffinityBackupFilter(dcIds.length, backups);
 
         IgniteEx srv = startClusterAcrossDataCenters(dcIds, nodesPerDc);
 
-        verifyDistributionProperties(srv, dcIds, nodesPerDc, backups);
+        verifyDistributionGuarantees(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();
 
-        verifyDistributionProperties(srv, dcIds, nodesPerDc, backups);
+        verifyDistributionGuarantees(srv, dcIds, nodesPerDc, backups);
 
         //stopping another node in DC_1 should not compromise distribution as well
         stopGrid(6);
 
         awaitPartitionMapExchange();
 
-        verifyDistributionProperties(srv, dcIds, nodesPerDc, backups);
+        verifyDistributionGuarantees(srv, dcIds, nodesPerDc, backups);
     }
 
     /**
@@ -146,11 +220,11 @@ public void test3DcDistribution() throws Exception {
         dcIds = new String[] {"DC_0", "DC_1", "DC_2"};
         int nodesPerDc = 2;
         backups = 5;
-        filter = new MdcAffinityBackupFilter(dcIds.length, backups);
+        backupFilter = new MdcAffinityBackupFilter(dcIds.length, backups);
 
         IgniteEx srv = startClusterAcrossDataCenters(dcIds, 2);
 
-        verifyDistributionProperties(srv, dcIds, nodesPerDc, backups);
+        verifyDistributionGuarantees(srv, dcIds, nodesPerDc, backups);
     }
 
     /**
@@ -163,10 +237,10 @@ public void test3DcDistribution() throws Exception {
     public void testAffinityFilterConfigurationValidation() throws Exception {
         dcIds = new String[] {"DC_0", "DC_1"};
         backups = 3;
-        filter = new MdcAffinityBackupFilter(dcIds.length, backups);
+        backupFilter = new MdcAffinityBackupFilter(dcIds.length, backups);
         startGrid(0);
 
-        filter = new ClusterNodeAttributeAffinityBackupFilter("DC_ID");
+        backupFilter = new ClusterNodeAttributeAffinityBackupFilter("DC_ID");
         try {
             startGrid(1);
 
@@ -180,7 +254,7 @@ public void testAffinityFilterConfigurationValidation() throws Exception {
             assertTrue(errMsg.contains("Affinity backup filter class mismatch"));
         }
 
-        filter = new MdcAffinityBackupFilter(dcIds.length, backups + dcIds.length);
+        backupFilter = new MdcAffinityBackupFilter(dcIds.length, backups + dcIds.length);
         try {
             startGrid(1);
 
@@ -207,15 +281,21 @@ 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());
+        backupFilter = new MdcAffinityBackupFilter(dcIds.length, backups);
+        Predicate<ClusterNode> dc1NodesFilter = node -> dcIds[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);
+        oldDistribution = verifyNoRebalancing(0, true, null, srv, oldDistribution, dc1NodesFilter);
+        oldDistribution = verifyNoRebalancing(1, true, null, srv, oldDistribution, dc1NodesFilter);
+        // Start srv0 back in DC_0 to make sure that neither starting nor stopping a server doesn't affect
+        // affinity distribution in remote DC.
+        oldDistribution = verifyNoRebalancing(0, false, dcIds[0], srv, oldDistribution, dc1NodesFilter);
+        // Stop it again.
+        oldDistribution = verifyNoRebalancing(0, true, null, srv, oldDistribution, dc1NodesFilter);
+        verifyNoRebalancing(2, true, null, srv, oldDistribution, dc1NodesFilter);
     }
 
     /** Starts specified number of nodes in each DC. */
@@ -236,11 +316,18 @@ private IgniteEx startClusterAcrossDataCenters(String[] dcIds, int nodesPerDc) t
     /** */
     private Map<Integer, Set<UUID>> verifyNoRebalancing(
         int srvIdx,
+        boolean stopSrv,
+        String startingSrvDcId,
         IgniteEx srv,
         Map<Integer, Set<UUID>> oldDistribution,
         Predicate<ClusterNode> dc1NodesFilter
-    ) throws InterruptedException {
-        stopGrid(srvIdx);
+    ) throws Exception {
+        if (stopSrv)
+            stopGrid(srvIdx);
+        else {
+            System.setProperty(IgniteSystemProperties.IGNITE_DATA_CENTER_ID, startingSrvDcId);
+            startGrid(srvIdx);
+        }
         awaitPartitionMapExchange();
         IgniteCache<Integer, Integer> cache = srv.getOrCreateCache(DEFAULT_CACHE_NAME);
 
@@ -287,7 +374,7 @@ private Map<Integer, Set<UUID>> affinityForPartitions(IgniteCache<Integer, Integ
     /**
      * Checks that copies of each partition are distributed evenly across data centers and copies are spread evenly across nodes.
      */
-    private void verifyDistributionProperties(
+    private void verifyDistributionGuarantees(
         IgniteEx srv,
         String[] dcIds,
         int nodesPerDc,