IGNITE-26580 Add documentation for MDC affinity backup filter

Author: sergey-chugunov-1985

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

@@ -23,39 +23,92 @@
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.lang.IgniteBiPredicate;
 
-/** */
+/**
+ * Multi-data center affinity backup filter that ensures each partition's data is distributed across multiple data centers,
+ * providing high availability and fault tolerance. This implementation guarantees at least one copy of the data in each
+ * data center and attempts to maintain the configured backup factor without discarding copies.
+ * <p>
+ * The filter works by grouping nodes based on their data center identification attribute (@see {@link ClusterNode#dataCenterId()})
+ * and ensuring that for every partition, at least one node from each data center is included in the primary-backup set.
+ * <p>
+ * The filter will discard backup copies only if the number of available nodes in a given data center is less
+ * than the number of copies assigned to that data center.
+ * For example, if a partition has 4 copies (1 primary and 3 backups) and the cluster has 2 data centers,
+ * than 2 copies are assigned to each data center. The only scenario when just a single copy is assigned to a node in a data center is when
+ * the number of nodes in that data center is one.
+ * <p>
+ * This class is constructed with a number of data centers the cluster spans and a number of backups of the cache this filter is applied to.
+ * Implementation expects that all copies can be spread evenly across all data centers. In other words, (backups + 1) is divisible by
+ * number of data centers without remainder. Uneven distributions of copies are not supported.
+ * <p>
+ * Warning: Ensure that all nodes have a consistent and valid data center identifier attribute. Missing or inconsistent values
+ * may lead to unexpected placement of data.
+ * </pre>
+ * <h2 class="header">Spring Example</h2>
+ * Create a partitioned cache template where each data center has at least one copy of the data, and the backup count is maintained.
+ * <pre name="code" class="xml">
+ * &lt;property name="cacheConfiguration"&gt;
+ *     &lt;list&gt;
+ *         &lt;bean id="cache-template-bean" abstract="true" class="org.apache.ignite.configuration.CacheConfiguration"&gt;
+ *             &lt;property name="name" value="JobcaseDefaultCacheConfig*"/&gt;
+ *             &lt;property name="cacheMode" value="PARTITIONED" /&gt;
+ *             &lt;property name="backups" value="3" /&gt;
+ *             &lt;property name="affinity"&gt;
+ *                 &lt;bean class="org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction"&gt;
+ *                     &lt;property name="affinityBackupFilter"&gt;
+ *                         &lt;bean class="org.apache.ignite.cache.affinity.rendezvous.MdcAffinityBackupFilter"&gt;
+ *                             &lt;constructor-arg value="2"/&gt; <!-- dcsNumber -->
+ *                             &lt;constructor-arg value="3"/&gt; <!-- backups, the same as in the cache template -->
+ *                         &lt;/bean&gt;
+ *                     &lt;/property&gt;
+ *                 &lt;/bean&gt;
+ *             &lt;/property&gt;
+ *         &lt;/bean&gt;
+ *     &lt;/list&gt;
+ * &lt;/property&gt;
+ * </pre>
+ * <p>
+ * With more backups, additional replicas can be distributed across different data centers to further improve redundancy.
+ */
 public class MdcAffinityBackupFilter implements IgniteBiPredicate<ClusterNode, List<ClusterNode>> {
     /** */
     private static final long serialVersionUID = 1L;
 
-    /** */
-    private final int dcsCount;
+    /** Number of data centers. */
+    private final int dcsNum;
 
-    /** */
+    /** Number of copies of each partition, including primary. */
     private final int primaryAndBackups;
 
-    /** */
+    /** Map is used to optimize the time it takes to perform a partition assignment procedure. */
     private final Map<String, Integer> partsDistrMap;
 
     /**
-     * @param dcsCount
-     * @param backups
+     * @param dcsNum Number of data centers.
+     * @param backups Number of backups.
      */
-    public MdcAffinityBackupFilter(int dcsCount, int backups) {
-        this.dcsCount = dcsCount;
-        partsDistrMap = new HashMap<>(dcsCount + 1);
+    public MdcAffinityBackupFilter(int dcsNum, int backups) {
+        this.dcsNum = dcsNum;
+        partsDistrMap = new HashMap<>(dcsNum + 1);
         primaryAndBackups = backups + 1;
     }
 
-    /** {@inheritDoc} */
-    @Override public boolean apply(ClusterNode node, List<ClusterNode> list) {
-        if (list.size() == 1) { //list contains only primary node, thus we started new assignment round.
+    /**
+     * 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.
+     */
+    @Override public boolean apply(ClusterNode candidate, List<ClusterNode> previouslySelected) {
+        if (previouslySelected.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);
+            partsDistrMap.put(previouslySelected.get(0).dataCenterId(), 1);
         }
 
-        String candidateDcId = node.dataCenterId();
+        String candidateDcId = candidate.dataCenterId();
         Integer candDcPartsCopies = partsDistrMap.get(candidateDcId);
         boolean res = false;
 
@@ -65,7 +118,7 @@ public MdcAffinityBackupFilter(int dcsCount, int backups) {
             res = true;
         }
         else {
-            int partCopiesPerDc = primaryAndBackups / dcsCount;
+            int partCopiesPerDc = primaryAndBackups / dcsNum;
 
             if (candDcPartsCopies < partCopiesPerDc) {
                 partsDistrMap.put(candidateDcId, candDcPartsCopies + 1);

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

@@ -32,7 +32,10 @@
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.Test;
 
-/** */
+/**
+ * Verifies behaviour of {@link MdcAffinityBackupFilter} - guarantees that each DC has at least one copy of every partition,
+ * distribution uniformity.
+ */
 public class MdcAffinityBackupFilterSelfTest extends GridCommonAbstractTest {
     /** */
     private static final int PARTS_CNT = 1024;
@@ -83,21 +86,21 @@ public void test2DcDistribution() throws Exception {
 
         IgniteEx srv = startClusterAcrossDataCenters(dcIds, nodesPerDc);
 
-        checkPartitionsDistribution(srv, dcIds, nodesPerDc, backups);
+        verifyDistributionProperties(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();
 
-        checkPartitionsDistribution(srv, dcIds, nodesPerDc, backups);
+        verifyDistributionProperties(srv, dcIds, nodesPerDc, backups);
 
         //stopping another node in DC_1 should not compromise distribution as well
         stopGrid(6);
 
         awaitPartitionMapExchange();
 
-        checkPartitionsDistribution(srv, dcIds, nodesPerDc, backups);
+        verifyDistributionProperties(srv, dcIds, nodesPerDc, backups);
     }
 
     /**
@@ -113,7 +116,7 @@ public void test3DcDistribution() throws Exception {
 
         IgniteEx srv = startClusterAcrossDataCenters(dcIds, 2);
 
-        checkPartitionsDistribution(srv, dcIds, nodesPerDc, backups);
+        verifyDistributionProperties(srv, dcIds, nodesPerDc, backups);
     }
 
     /** Starts specified number of nodes in each DC. */
@@ -132,10 +135,9 @@ private IgniteEx startClusterAcrossDataCenters(String[] dcIds, int nodesPerDc) t
     }
 
     /**
-     * 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).
+     * Checks that copies of each partition are distributed evenly across data centers and copies are spread evenly across nodes.
      */
-    private void checkPartitionsDistribution(
+    private void verifyDistributionProperties(
         IgniteEx srv,
         String[] dcIds,
         int nodesPerDc,
@@ -150,10 +152,10 @@ private void checkPartitionsDistribution(
         Map<ClusterNode, Integer> overallCopiesPerNode = new HashMap<>();
         int[] copiesPerNode = new int[dcIds.length * nodesPerDc];
 
-        for (int i = 0; i < partCnt; i++) {
+        for (int partId = 0; partId < partCnt; partId++) {
             int[] partCopiesPerDc = new int[dcIds.length];
 
-            Collection<ClusterNode> nodes = aff.mapKeyToPrimaryAndBackups(i);
+            Collection<ClusterNode> nodes = aff.mapKeyToPrimaryAndBackups(partId);
 
             //calculate actual number of copies in each data center
             //aggregate copies per each node
@@ -175,14 +177,23 @@ private void checkPartitionsDistribution(
                 }
             }
 
-            //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]);
-            }
+            verifyCopyInEachDcGuarantee(partId, expectedCopiesPerNode, partCopiesPerDc);
         }
 
+        verifyDistributionUniformity(dcIds, overallCopiesPerNode);
+    }
+
+    /** */
+    private void verifyCopyInEachDcGuarantee(int partId, int expectedCopiesPerNode, int[] partCopiesPerDc) {
+        for (int dcIdx = 0; dcIdx < dcIds.length; dcIdx++) {
+            assertEquals(String.format("Unexpected number of copies of partition %d in data center %s", partId, dcIds[dcIdx]),
+                expectedCopiesPerNode,
+                partCopiesPerDc[dcIdx]);
+        }
+    }
+
+    /** */
+    private void verifyDistributionUniformity(String[] dcIds, Map<ClusterNode, Integer> overallCopiesPerNode) {
         for (String dcId : dcIds) {
             long nodesInDc = overallCopiesPerNode.entrySet().stream().filter(e -> e.getKey().dataCenterId().equals(dcId)).count();