IGNITE-26584 Discovery optimizations for MultiDC

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java

@@ -41,6 +41,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridNearAtomicAbstractUpdateRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemander.RebalanceFuture;
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition;
+import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState;
 import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology;
 import org.apache.ignite.internal.util.future.GridCompoundFuture;
 import org.apache.ignite.internal.util.future.GridFinishedFuture;
@@ -221,12 +222,14 @@ public boolean disableRebalancingCancellationOptimization() {
                 assert part != null;
                 assert part.id() == p;
 
+                GridDhtPartitionState state = part.state();
+
                 // Do not rebalance OWNING or LOST partitions.
-                if (part.state() == OWNING || part.state() == LOST)
+                if (state == OWNING || state == LOST)
                     continue;
 
                 // State should be switched to MOVING during PME.
-                if (part.state() != MOVING) {
+                if (state != MOVING) {
                     throw new AssertionError("Partition has invalid state for rebalance "
                         + aff.topologyVersion() + " " + part);
                 }

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java

@@ -920,7 +920,7 @@ private Collection<ClusterNode> updateTopologyHistory(long topVer, @Nullable Tcp
 
             Collection<ClusterNode> top = topHist.get(topVer);
 
-            assert top != null : "Failed to find topology history [msg=" + msg + ", hist=" + topHist + ']';
+            assert top != null : "Failed to find topology history [top=" + topVer + ", msg=" + msg + ", hist=" + topHist + ']';
 
             return top;
         }

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java

@@ -1837,6 +1837,7 @@ private void prepareNodeAddedMessage(
         TcpDiscoveryAbstractMessage msg,
         UUID destNodeId,
         @Nullable Collection<PendingMessage> msgs,
+        @Nullable IgniteUuid discardMsgId,
         @Nullable IgniteUuid discardCustomMsgId
     ) {
         assert destNodeId != null;
@@ -1874,10 +1875,7 @@ private void prepareNodeAddedMessage(
                     }
                 }
 
-                // No need to send discardMsgId because we already filtered out
-                // cleaned up messages.
-                // TODO IGNITE-11271
-                nodeAddedMsg.messages(msgs0, null, discardCustomMsgId);
+                nodeAddedMsg.messages(msgs0, discardMsgId, discardCustomMsgId);
 
                 Map<Long, Collection<ClusterNode>> hist;
 
@@ -2659,7 +2657,7 @@ private TcpDiscoveryAbstractMessage prepare(TcpDiscoveryAbstractMessage msg, UUI
 
                     TcpDiscoveryNodeAddedMessage msg0 = new TcpDiscoveryNodeAddedMessage(addedMsg);
 
-                    prepareNodeAddedMessage(msg0, destNodeId, null, null);
+                    prepareNodeAddedMessage(msg0, destNodeId, null, null, null);
 
                     msg0.topology(addedMsg.clientTopology());
 
@@ -3401,7 +3399,7 @@ private void sendMessageToClients(TcpDiscoveryAbstractMessage msg) {
                                 msg0 = U.unmarshal(spi.marshaller(), msgBytes,
                                     U.resolveClassLoader(spi.ignite().configuration()));
 
-                                prepareNodeAddedMessage(msg0, clientMsgWorker.clientNodeId, null, null);
+                                prepareNodeAddedMessage(msg0, clientMsgWorker.clientNodeId, null, null, null);
 
                                 msgBytes0 = null;
                             }
@@ -3751,7 +3749,7 @@ else if (!spi.failureDetectionTimeoutEnabled() && (e instanceof
                                     long tsNanos = System.nanoTime();
 
                                     prepareNodeAddedMessage(pendingMsg, next.id(), pendingMsgs.msgs,
-                                        pendingMsgs.customDiscardId);
+                                        pendingMsgs.discardId, pendingMsgs.customDiscardId);
 
                                     addFailedNodes(pendingMsg, failedNodes);
 
@@ -3793,7 +3791,7 @@ else if (!spi.failureDetectionTimeoutEnabled() && (e instanceof
 
                             if (!(msg instanceof TcpDiscoveryConnectionCheckMessage))
                                 prepareNodeAddedMessage(msg, next.id(), pendingMsgs.msgs,
-                                    pendingMsgs.customDiscardId);
+                                    pendingMsgs.discardId, pendingMsgs.customDiscardId);
 
                             try {
                                 SecurityUtils.serializeVersion(1);
@@ -4007,7 +4005,7 @@ private void processPendingMessagesLocally(TcpDiscoveryAbstractMessage curMsg) {
 
             for (TcpDiscoveryAbstractMessage pendingMsg : pendingMsgs) {
                 prepareNodeAddedMessage(pendingMsg, locNodeId, pendingMsgs.msgs,
-                    pendingMsgs.customDiscardId);
+                    pendingMsgs.discardId, pendingMsgs.customDiscardId);
 
                 pendingMsg.senderNodeId(locNodeId);
 
@@ -4214,7 +4212,7 @@ private void processJoinRequestMessage(final TcpDiscoveryJoinRequestMessage msg)
             if (isLocalNodeCoordinator()) {
                 TcpDiscoveryNode existingNode = ring.node(node.id());
 
-                if (existingNode != null) {
+                if (existingNode != null && existingNode.visible()) {
                     if (!node.socketAddresses().equals(existingNode.socketAddresses())) {
                         if (!pingNode(existingNode)) {
                             U.warn(log, "Sending node failed message for existing node: " + node);
@@ -5213,12 +5211,6 @@ else if (spiState == CONNECTING)
 
                             pendingMsgs.reset(msg.messages(), msg.discardedMessageId(),
                                 msg.discardedCustomMessageId());
-
-                            // Clear data to minimize message size.
-                            msg.messages(null, null, null);
-                            msg.topology(null);
-                            msg.topologyHistory(null);
-                            msg.clearDiscoveryData();
                         }
                         else {
                             if (log.isDebugEnabled())
@@ -5285,7 +5277,7 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
             UUID locNodeId = getLocalNodeId();
 
             if (locNodeCoord) {
-                if (msg.verified()) {
+                if (msg.verified() && msg.verifierNodeId().equals(locNodeId)) {
                     addMessage(new TcpDiscoveryDiscardMessage(locNodeId, msg.id(), false));
 
                     return;

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/MdcAwareNodesComparator.java

@@ -0,0 +1,37 @@
+/*
+ * 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.spi.discovery.tcp.internal;
+
+import java.util.Comparator;
+
+/** Compares nodes using the Data Center id as a primary factor. */
+public class MdcAwareNodesComparator implements Comparator<TcpDiscoveryNode> {
+    /** */
+    @Override public int compare(TcpDiscoveryNode n1, TcpDiscoveryNode n2) {
+        String n1DcId = n1.dataCenterId() == null ? "" : n1.dataCenterId();
+        String n2DcId = n2.dataCenterId() == null ? "" : n2.dataCenterId();
+
+        int res = n1DcId.compareTo(n2DcId);
+
+        if (res == 0) {
+            res = n1.compareTo(n2);
+        }
+
+        return res;
+    }
+}

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java

@@ -506,7 +506,10 @@ public void clear() {
             if (filtered.size() < 2)
                 return null;
 
-            Iterator<TcpDiscoveryNode> iter = filtered.iterator();
+            NavigableSet<TcpDiscoveryNode> sorted = new TreeSet<>(new MdcAwareNodesComparator());
+            sorted.addAll(filtered);
+
+            Iterator<TcpDiscoveryNode> iter = sorted.iterator();
 
             while (iter.hasNext()) {
                 TcpDiscoveryNode node = iter.next();
@@ -515,7 +518,7 @@ public void clear() {
                     break;
             }
 
-            return iter.hasNext() ? iter.next() : F.first(filtered);
+            return iter.hasNext() ? iter.next() : F.first(sorted);
         }
         finally {
             rwLock.readLock().unlock();
@@ -541,10 +544,13 @@ public void clear() {
             if (filtered.size() < 2)
                 return null;
 
+            NavigableSet<TcpDiscoveryNode> sorted = new TreeSet<>(new MdcAwareNodesComparator());
+            sorted.addAll(filtered);
+
             TcpDiscoveryNode previous = null;
 
             // Get last node that is previous in a ring
-            for (TcpDiscoveryNode node : filtered) {
+            for (TcpDiscoveryNode node : sorted) {
                 if (locNode.equals(node) && previous != null)
                     break;
 
@@ -569,11 +575,14 @@ public TcpDiscoveryNode previousNodeOf(TcpDiscoveryNode ringNode) {
         try {
             TcpDiscoveryNode prev = null;
 
-            for (TcpDiscoveryNode node : nodes) {
+            NavigableSet<TcpDiscoveryNode> sorted = new TreeSet<>(new MdcAwareNodesComparator());
+            sorted.addAll(nodes);
+
+            for (TcpDiscoveryNode node : sorted) {
                 if (node.equals(ringNode)) {
                     if (prev == null)
                         // ringNode is the first node, return last node in the ring.
-                        return nodes.last();
+                        return sorted.last();
 
                     return prev;
                 }

modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheExchangeMergeMdcTest.java

@@ -0,0 +1,56 @@
+/*
+ * 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.internal.processors.cache.distributed;
+
+import java.util.concurrent.ThreadLocalRandom;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.configuration.IgniteConfiguration;
+
+/** */
+public class CacheExchangeMergeMdcTest extends CacheExchangeMergeTest {
+    /** */
+    protected static final String DC_ID_0 = "DC0";
+
+    /** */
+    protected static final String DC_ID_1 = "DC1";
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+        applyDC();
+
+        return cfg;
+    }
+
+    /** */
+    protected void applyDC() {
+        ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+        boolean mainDc = rnd.nextBoolean();
+
+        System.setProperty(IgniteSystemProperties.IGNITE_DATA_CENTER_ID, mainDc ? DC_ID_0 : DC_ID_1);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        super.afterTest();
+
+        System.clearProperty(IgniteSystemProperties.IGNITE_DATA_CENTER_ID);
+    }
+}