Ignite 4756

modules/core/src/main/java/org/apache/ignite/IgniteSystemProperties.java

@@ -637,6 +637,15 @@ public final class IgniteSystemProperties {
     public static final String IGNITE_CLIENT_CACHE_CHANGE_MESSAGE_TIMEOUT =
         "IGNITE_CLIENT_CACHE_CHANGE_MESSAGE_TIMEOUT";
 
+    /**
+     * Property controlling printing warning message and statistic only when nodes count differs more then threshold.
+     * The problem message. If calculated statistic is more than IGNITE_PART_DISTRIBUTION_WARN_THRESHOLD
+     * there is problem message "Partition map has been built (distribution is not even for caches) ... " else
+     * "Partition map has been built (distribution is even)."
+     * Printing enabled by default.
+     */
+    public static final String IGNITE_PART_DISTRIBUTION_WARN_THRESHOLD = "IGNITE_PART_DISTRIBUTION_WARN_THRESHOLD";
+
     /**
      * Enforces singleton.
      */

modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java

@@ -18,7 +18,9 @@
 package org.apache.ignite.internal.processors.affinity;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -48,6 +50,7 @@
 import org.apache.ignite.lang.IgnitePredicate;
 import org.jetbrains.annotations.Nullable;
 import org.jsr166.ConcurrentHashMap8;
+import org.apache.ignite.IgniteSystemProperties;
 
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_AFFINITY_HISTORY_SIZE;
 import static org.apache.ignite.IgniteSystemProperties.getInteger;
@@ -295,6 +298,8 @@ public List<List<ClusterNode>> calculate(AffinityTopologyVersion topVer, Discove
 
         assert assignment != null;
 
+        printDistribution(assignment, sorted, cacheOrGrpName, ctx.localNodeId().toString());
+
         idealAssignment = assignment;
 
         if (locCache)
@@ -303,6 +308,114 @@ public List<List<ClusterNode>> calculate(AffinityTopologyVersion topVer, Discove
         return assignment;
     }
 
+    /**
+     * @param partitionsByNodes Affinity result.
+     * @param nodes Topology.
+     * @param cacheName
+     * @param localNodeID
+     */
+    private void printDistribution(List<List<ClusterNode>> partitionsByNodes,
+        Collection<ClusterNode> nodes, String cacheName, String localNodeID) {
+
+        String ignitePartDistribution = System.getProperty(IgniteSystemProperties.IGNITE_PART_DISTRIBUTION_WARN_THRESHOLD);
+
+        if (ignitePartDistribution == null || ignitePartDistribution == "")
+            ignitePartDistribution = "0";
+
+        int[] nodesParts = freqDistribution(partitionsByNodes, nodes);
+
+        if (nodesParts.length != 0) {
+
+            for (int i = 0; i < nodesParts.length; i++) {
+
+                if (nodesParts[i] != partitionsByNodes.size()) {
+                    int percentage = (int)Math.round((double)nodesParts[i] / partitionsByNodes.size() * 100);
+
+                    if (percentage >= Integer.valueOf(ignitePartDistribution) || ignitePartDistribution == null) {
+                        log.info("Partition map has been built (distribution is not even for caches) [cacheName="
+                            + cacheName + ", " + " nodeId=" + localNodeID +
+                            ", totalPartitionsCount=" + partitionsByNodes.size() + ", percentageOfTotalPartsCount=" + percentage + "%"
+                            + ", parts=" + nodesParts[i] + "]");
+                    }
+                    else {
+                        log.info("Partition map has been built (distribution is even)");
+                    }
+                }
+            }
+        }
+    }
+
+    /**
+     * The array with count of partitions on node:
+     *
+     * element 0 - primary partitions counts
+     * element 1 - backup#0 partitions counts
+     * etc
+     *
+     * Rows correspond to the nodes.
+     *
+     * @param partitionsByNodes Affinity result.
+     * @param nodes All nodes for current topology.
+     * @return Frequency distribution array with counts of partitions on node.
+     */
+    private int[] freqDistribution(List<List<ClusterNode>> partitionsByNodes,
+        Collection<ClusterNode> nodes) {
+        List<Map<ClusterNode, AtomicInteger>> nodeMaps = new ArrayList<>();
+
+        int backups = partitionsByNodes.get(0).size();
+
+        for (int i = 0; i < backups; ++i) {
+            Map<ClusterNode, AtomicInteger> nodeMap = new HashMap<>();
+
+            for (List<ClusterNode> partitionByNodes : partitionsByNodes) {
+                ClusterNode node = partitionByNodes.get(i);
+
+                if (node.isLocal()) {
+                    if (!nodeMap.containsKey(node)) {
+                        nodeMap.put(node, new AtomicInteger(1));
+                    }
+                    else
+                        nodeMap.get(node).incrementAndGet();
+                }
+            }
+
+            nodeMaps.add(nodeMap);
+        }
+
+        List<List<Integer>> byNodes = new ArrayList<>(nodes.size());
+
+        int resultCount = 0;
+
+        for (ClusterNode node : nodes) {
+            if (node.isLocal()) {
+                List<Integer> byBackups = new ArrayList<>(backups);
+
+                for (int j = 0; j < backups; ++j) {
+                    if (nodeMaps.get(j).get(node) == null)
+                        byBackups.add(0);
+                    else
+                        byBackups.add(nodeMaps.get(j).get(node).get());
+
+                    resultCount++;
+                }
+
+                byNodes.add(byBackups);
+            }
+        }
+
+        int[] result = new int[resultCount];
+
+        int i = 0;
+
+        for (List<Integer> byNode : byNodes) {
+            for (Integer partsCount : byNode) {
+                result[i++] = partsCount;
+            }
+        }
+
+        return result;
+    }
+
     /**
      * Copies previous affinity assignment when discovery event does not cause affinity assignment changes
      * (e.g. client node joins on leaves).

modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityFunctionContextImpl.java

@@ -45,7 +45,10 @@ public class GridAffinityFunctionContextImpl implements AffinityFunctionContext
 
     /**
      * @param topSnapshot Topology snapshot.
+     * @param prevAssignment Prevent Assignment.
+     * @param discoEvt Discovery Event.
      * @param topVer Topology version.
+     * @param backups Quantity of backups.
      */
     public GridAffinityFunctionContextImpl(List<ClusterNode> topSnapshot, List<List<ClusterNode>> prevAssignment,
         DiscoveryEvent discoEvt, @NotNull AffinityTopologyVersion topVer, int backups) {

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

@@ -104,7 +104,6 @@ public void testRandomReassignmentThreeBackups() throws Exception {
     }
 
     /**
-     * @param backups Number of backups.
      * @throws Exception If failed.
      */
     public void testNullKeyForPartitionCalculation() throws Exception {

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

@@ -0,0 +1,195 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.cache.affinity.rendezvous;
+
+
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.processors.affinity.GridAffinityAssignmentCache;
+import org.apache.ignite.internal.processors.cache.GridCacheContext;
+import org.apache.ignite.internal.processors.cache.GridCacheProcessor;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.spi.eventstorage.memory.MemoryEventStorageSpi;
+import org.apache.ignite.testframework.GridStringLogger;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ * Tests for {@link RendezvousAffinityFunction}.
+ */
+public class RendezvousAffinityFunctionCalculateDistributionSelfTest extends GridCommonAbstractTest {
+
+    /** Number of backups */
+    private int backups = 3;
+
+    /** Backup value of IGNITE_PART_DISTRIBUTION_WARN_THRESHOLD */
+    String backup;
+
+    /**  */
+    private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        super.beforeTest();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTestsStopped() throws Exception {
+        super.afterTest();
+    }
+
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+        ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+
+        TcpCommunicationSpi commSpi = new TcpCommunicationSpi();
+
+        commSpi.setIdleConnectionTimeout(100);
+
+        cfg.setCommunicationSpi(commSpi);
+
+        MemoryEventStorageSpi evtSpi = new MemoryEventStorageSpi();
+
+        evtSpi.setExpireCount(50);
+
+        cfg.setEventStorageSpi(evtSpi);
+
+        CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME);
+
+        ccfg.setCacheMode(CacheMode.PARTITIONED);
+        ccfg.setAtomicityMode(CacheAtomicityMode.ATOMIC);
+        ccfg.setBackups(backups);
+        ccfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_ASYNC);
+        ccfg.setNearConfiguration(null);
+
+        cfg.setCacheConfiguration(ccfg);
+
+        cfg.setClientMode(false);
+
+        return cfg;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testDistributionCalculationDefault() throws Exception {
+        System.setProperty(IgniteSystemProperties.IGNITE_PART_DISTRIBUTION_WARN_THRESHOLD, "");
+
+        Ignite ignite = startGrid(0);
+
+        awaitPartitionMapExchange();
+
+        final GridStringLogger log = new GridStringLogger(false, this.log);
+
+        GridCacheProcessor proc = ((IgniteKernal)ignite).context().cache();
+
+        for (GridCacheContext cctx : proc.context().cacheContexts()) {
+            GridAffinityAssignmentCache aff = GridTestUtils.getFieldValue(cctx.affinity(), "aff");
+
+            GridTestUtils.setFieldValue(aff, "log", log);
+        }
+
+        startGrid(1);
+
+        String res = log.toString();
+
+        assertTrue(res.contains("Partition map has been built (distribution is not even for caches)"));
+
+        assertFalse(res.contains("Partition map has been built (distribution is even)"));
+
+        log.reset();
+
+        stopAllGrids();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testDistributionCalculationProblemMessage() throws Exception {
+        System.setProperty(IgniteSystemProperties.IGNITE_PART_DISTRIBUTION_WARN_THRESHOLD, String.valueOf(1));
+
+        Ignite ignite = startGrid(0);
+
+        awaitPartitionMapExchange();
+
+        final GridStringLogger log = new GridStringLogger(false, this.log);
+
+        GridCacheProcessor proc = ((IgniteKernal)ignite).context().cache();
+
+        for (GridCacheContext cctx : proc.context().cacheContexts()) {
+            GridAffinityAssignmentCache aff = GridTestUtils.getFieldValue(cctx.affinity(), "aff");
+
+            GridTestUtils.setFieldValue(aff, "log", log);
+        }
+
+        startGrid(1);
+
+        String res = log.toString();
+
+        assertTrue(res.contains("Partition map has been built (distribution is not even for caches)"));
+
+        assertFalse(res.contains("Partition map has been built (distribution is even)"));
+
+        log.reset();
+
+        stopAllGrids();
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testDistributionCalculationOKMessage() throws Exception {
+        System.setProperty(IgniteSystemProperties.IGNITE_PART_DISTRIBUTION_WARN_THRESHOLD, String.valueOf(75));
+
+        Ignite ignite = startGrid(0);
+
+        awaitPartitionMapExchange();
+
+        final GridStringLogger log = new GridStringLogger(false, this.log);
+
+        GridCacheProcessor proc = ((IgniteKernal)ignite).context().cache();
+
+        for (GridCacheContext cctx : proc.context().cacheContexts()) {
+            GridAffinityAssignmentCache aff = GridTestUtils.getFieldValue(cctx.affinity(), "aff");
+
+            GridTestUtils.setFieldValue(aff, "log", log);
+        }
+
+        startGrid(1);
+
+        String res = log.toString();
+
+        assertTrue(res.contains("Partition map has been built (distribution is even)"));
+
+        assertFalse(res.contains("Partition map has been built (distribution is not even for caches)"));
+
+        log.reset();
+
+        stopAllGrids();
+    }
+}