diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java
index 3d9bfcc..29aa940 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/CustomMessageWrapper.java
@@ -50,9 +50,8 @@
     }
 
     /** {@inheritDoc} */
-    @Deprecated
     @Override public boolean stopProcess() {
-        return false;
+        return delegate.stopProcess();
     }
 
     /**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java
index 2309fc0..c3cd0ab 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryCustomMessage.java
@@ -20,6 +20,7 @@
 import java.io.Serializable;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddedMessage;
 import org.jetbrains.annotations.Nullable;
@@ -92,6 +93,15 @@
     public boolean isMutable();
 
     /**
+     * See {@link DiscoverySpiCustomMessage#stopProcess()}.
+     *
+     * @return {@code True} if message should not be sent to others nodes after it was processed on coordinator.
+     */
+    public default boolean stopProcess() {
+        return false;
+    }
+
+    /**
      * Creates new discovery cache if message caused topology version change.
      *
      * @param mgr Discovery manager.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinityChangeMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinityChangeMessage.java
index 708e3c1..1330e88 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinityChangeMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinityChangeMessage.java
@@ -28,6 +28,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.spi.discovery.DiscoverySpiMutableCustomMessageSupport;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -55,8 +56,18 @@
     /** */
     private GridDhtPartitionsFullMessage partsMsg;
 
-    /** */
-    private transient boolean exchangeNeeded;
+    /** If this flag is {@code true} then this message should lead to partition map exchnage. */
+    private boolean exchangeNeeded;
+
+    /**
+     * This flag indicates that this message should not be passed to other nodes except the coordinator.
+     * Instead of this message, the message which is returned by {@link #ackMessage()} will be sent.
+     * See {@link DiscoveryCustomMessage#stopProcess()}.
+     *
+     * This flag is used when discovery SPI does not support mutable custom messages.
+     * See {@link DiscoverySpiMutableCustomMessageSupport}.
+     */
+    private transient boolean stopProc;
 
     /**
      * Constructor used when message is created after cache rebalance finished.
@@ -76,7 +87,8 @@
      * @param partsMsg Partitions messages.
      * @param assignmentChange Assignment change.
      */
-    public CacheAffinityChangeMessage(GridDhtPartitionExchangeId exchId,
+    public CacheAffinityChangeMessage(
+        GridDhtPartitionExchangeId exchId,
         GridDhtPartitionsFullMessage partsMsg,
         Map<Integer, Map<Integer, List<UUID>>> assignmentChange) {
         this.exchId = exchId;
@@ -140,17 +152,44 @@
 
     /** {@inheritDoc} */
     @Nullable @Override public DiscoveryCustomMessage ackMessage() {
-        return null;
+        if (!stopProc)
+            return null;
+
+        // If stopProc is equal to true, then Discovery SPI does not support mutable custom messages.
+        // Let's return the same message, that was muted on the coordinator node. This message will be sent to all nodes
+        // instead of the original one.
+        return this;
     }
 
     /** {@inheritDoc} */
     @Override public boolean isMutable() {
-        return false;
+        return true;
     }
 
     /** {@inheritDoc} */
-    @Nullable @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr,
-        AffinityTopologyVersion topVer, DiscoCache discoCache) {
+    @Override public boolean stopProcess() {
+        return stopProc;
+    }
+
+    /**
+     * Sets stop processing flag. If this flag is {@code true} then this message is not passed to other nodes after
+     * the coordinator node notitied its own listner. If method {@link #ackMessage()} returns non-null ack message,
+     * it is sent to all nodes.
+     * This flag is used when discovery SPI does not support mutable custom messages.
+     * See {@link DiscoverySpiMutableCustomMessageSupport}.
+     *
+     * @param stopProc If {@code true} then this message is not passed to other nodes.
+     */
+    public void stopProcess(boolean stopProc) {
+        this.stopProc = stopProc;
+    }
+
+    /** {@inheritDoc} */
+    @Nullable @Override public DiscoCache createDiscoCache(
+        GridDiscoveryManager mgr,
+        AffinityTopologyVersion topVer,
+        DiscoCache discoCache
+    ) {
         return discoCache.copy(topVer, null);
     }
 
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
index f6f4a95..fda6598 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
@@ -77,6 +77,7 @@
 import org.apache.ignite.lang.IgniteFuture;
 import org.apache.ignite.lang.IgniteInClosure;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
 import org.jetbrains.annotations.Nullable;
 import static org.apache.ignite.cache.CacheRebalanceMode.NONE;
 import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
@@ -219,11 +220,21 @@
             return false;
         }
 
+        boolean isClient = cctx.discovery().localNode().isClient();
+
+        if (cctx.kernalContext().config().getDiscoverySpi() instanceof TcpDiscoverySpi)
+            isClient &= !((TcpDiscoverySpi)cctx.kernalContext().config().getDiscoverySpi()).isForceServerMode();
+
         // Skip message if affinity was already recalculated.
-        boolean exchangeNeeded = lastAffVer == null || lastAffVer.equals(msg.topologyVersion());
+        // Client node should just accept the flag from the mutated message.
+        boolean exchangeNeeded = (isClient) ? msg.exchangeNeeded() :
+            lastAffVer == null || lastAffVer.equals(msg.topologyVersion());
 
         msg.exchangeNeeded(exchangeNeeded);
 
+        if (!cctx.discovery().mutableCustomMessages() && !isClient)
+            msg.stopProcess(true);
+
         if (exchangeNeeded) {
             if (log.isDebugEnabled()) {
                 log.debug("Need process affinity change message [lastAffVer=" + lastAffVer +
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java
index afd7093..6e11673 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/DiscoverySpiCustomMessage.java
@@ -45,10 +45,9 @@
      * then message is not passed to others nodes, if after this method {@link #ackMessage()} returns non-null ack
      * message, it is sent to all nodes.
      *
-     * @return {@code True} if message should not be sent to all nodes.
+     * Note: this method is used then and only then the zookeeper discovery is configured.
      *
-     * @deprecated Is not used anymore and will be removed at 3.0.
+     * @return {@code True} if message should not be sent to all nodes.
      */
-    @Deprecated
     public boolean stopProcess();
 }
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java
index 0309447..d50c2c5 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java
@@ -105,7 +105,10 @@
 import static org.apache.ignite.cache.CacheRebalanceMode.ASYNC;
 import static org.apache.ignite.cache.CacheRebalanceMode.SYNC;
 import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
+import static org.apache.ignite.internal.TestRecordingCommunicationSpi.blockSingleExhangeMessage;
 import static org.apache.ignite.internal.processors.cache.ExchangeContext.IGNITE_EXCHANGE_COMPATIBILITY_VER_1;
+import static org.apache.ignite.testframework.GridTestUtils.runAsync;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
 
 /**
  *
@@ -1352,7 +1355,7 @@
             Ignite ignite = grids.get(i);
 
             if (!blocked.contains(ignite.name())) {
-                GridTestUtils.waitForCondition(new GridAbsPredicate() {
+                waitForCondition(new GridAbsPredicate() {
                     @Override public boolean apply() {
                         return fut.isDone();
                     }
@@ -1527,7 +1530,7 @@
             });
         }
 
-        IgniteInternalFuture<?> stopFut = GridTestUtils.runAsync(new Callable<Void>() {
+        IgniteInternalFuture<?> stopFut = runAsync(new Callable<Void>() {
             @Override public Void call() throws Exception {
                 for (int j = 1; j < NODES; j++) {
                     TestRecordingCommunicationSpi spi =
@@ -1561,6 +1564,106 @@
     }
 
     /**
+     * Wait for rebalance, send affinity change message, but affinity already changed
+     * (new nodes joined: server + client). Checks that tere is no race that could lead to
+     * unexpected partition map exchange on the client node.
+     *
+     * @throws Exception If failed.
+     */
+    @Test
+    public void testDelayAssignmentAffinityChangedUnexpectedPME() throws Exception {
+        Ignite ignite0 = startServer(0, 1);
+
+        for (int i = 0; i < 1024; i++)
+            ignite0.cache(CACHE_NAME1).put(i, i);
+
+        DiscoverySpiTestListener lsnr = new DiscoverySpiTestListener();
+
+        ((IgniteDiscoverySpi)ignite0.configuration().getDiscoverySpi()).setInternalListener(lsnr);
+
+        TestRecordingCommunicationSpi commSpi0 =
+            (TestRecordingCommunicationSpi)ignite0.configuration().getCommunicationSpi();
+
+        // Starting a new client node should not lead to a rebalance obviously.
+        // So, it is expected that data distribution is ideal (ideal assignment).
+        startClient(1, 2);
+
+        checkAffinity(2, topVer(2, 0), true);
+
+        // Block late affinity assignment. (*)
+        lsnr.blockCustomEvent(CacheAffinityChangeMessage.class);
+
+        // Starting a new server node triggers data rebalancing.
+        // [3, 0] - is not ideal (expected)
+        startServer(2, 3);
+
+        checkAffinity(3, topVer(3, 0), false);
+
+        // Wait for sending late affinity assignment message (1) from the coordinator node.
+        // This message will be blocked (*)
+        lsnr.waitCustomEvent();
+
+        // Block rebalance messages.
+        blockSupplySend(commSpi0, CACHE_NAME1);
+
+        // Starting a new server node means that the late affinity assignment message (1) should be skipped.
+        startServer(3, 4);
+
+        TestRecordingCommunicationSpi clientSpi = new TestRecordingCommunicationSpi();
+        clientSpi.blockMessages(blockSingleExhangeMessage());
+        spiC = igniteInstanceName -> clientSpi;
+
+        IgniteInternalFuture<?> startClientFut = runAsync(() -> {
+            startClient(4, 5);
+        });
+
+        clientSpi.waitForBlocked();
+
+        // Unblock the late affinity assignment message (1).
+        lsnr.stopBlockCustomEvents();
+
+        clientSpi.stopBlock();
+
+        startClientFut.get(15_000);
+
+        // [5, 0] - is not ideal (expected)
+        checkAffinity(5, topVer(5, 0), false);
+
+        // Rebalance is blocked at this moment, so [5, 1] is not ready.
+        checkNoExchange(5, topVer(5, 1));
+
+        // Unblock rebalancing.
+        // The late affinity assignments message (2) should be fired after all.
+        commSpi0.stopBlock();
+
+        // [5, 1] should be ideal
+        checkAffinity(5, topVer(5, 1), true);
+
+        // The following output demonstrates the issue.
+        // The coordinator node and client initiate PME on the same toplogy version,
+        // but it relies to different custom messages.
+        // client:
+        //      Started exchange init [
+        //          topVer=AffinityTopologyVersion [topVer=5, minorTopVer=1],
+        //          crd=false,
+        //          evt=DISCOVERY_CUSTOM_EVT, evtNode=00ac9434-fd34-4aae-95d3-ceb477700000,
+        //          customEvt=CacheAffinityChangeMessage [
+        //              id=3ccc8984181-ea41279c-71cb-4b8c-8b48-1dee1baa6fe0,                       <<< (1)
+        //              topVer=AffinityTopologyVersion [topVer=3, minorTopVer=0], ...]             <<< !!!
+        // coordinator:
+        //      Started exchange init
+        //          [topVer=AffinityTopologyVersion [topVer=5, minorTopVer=1],
+        //          crd=true,
+        //          evt=DISCOVERY_CUSTOM_EVT, evtNode=00ac9434-fd34-4aae-95d3-ceb477700000,
+        //          customEvt=CacheAffinityChangeMessage [
+        //              id=d2ec8984181-ea41279c-71cb-4b8c-8b48-1dee1baa6fe0,                        <<< (2)
+        //              topVer=AffinityTopologyVersion [topVer=4, minorTopVer=0], ...]              <<< !!!
+        awaitPartitionMapExchange(true, true, null, false);
+
+        assertPartitionsSame(idleVerify(grid(0), CACHE_NAME1));
+    }
+
+    /**
      * Wait for rebalance, send affinity change message, but affinity already changed (new node joined).
      *
      * @throws Exception If failed.
@@ -1657,7 +1760,7 @@
 
             lsnr.stopBlockCustomEvents();
 
-            boolean started = GridTestUtils.waitForCondition(new GridAbsPredicate() {
+            boolean started = waitForCondition(new GridAbsPredicate() {
                 @Override public boolean apply() {
                     return startedFuture.isDone();
                 }
@@ -2174,7 +2277,7 @@
             //Ensure exchanges merge.
             spiC = igniteInstanceName -> testSpis[getTestIgniteInstanceIndex(igniteInstanceName)];
 
-            GridTestUtils.runAsync(() -> {
+            runAsync(() -> {
                 try {
                     for (int j = 1; j < NODES; j++)
                         testSpis[j].waitForBlocked();
@@ -2321,7 +2424,7 @@
             }
         }, NODES, "update-thread");
 
-        IgniteInternalFuture<?> srvRestartFut = GridTestUtils.runAsync(new Callable<Void>() {
+        IgniteInternalFuture<?> srvRestartFut = runAsync(new Callable<Void>() {
             @Override public Void call() throws Exception {
                 while (!fail.get() && System.currentTimeMillis() < stopTime) {
                     Ignite node = startGrid(NODES);
@@ -2462,7 +2565,7 @@
 
             final ClusterNode srvcNode = affinity.get(part).get(0);
 
-            boolean wait = GridTestUtils.waitForCondition(new PA() {
+            boolean wait = waitForCondition(new PA() {
                 @Override public boolean apply() {
                     TestService srvc = grid(srvcNode).services().service(srvcName);
 
diff --git a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java
index 41a4761..9b7476c 100644
--- a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java
+++ b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveFinishMessage.java
@@ -58,7 +58,6 @@
     }
 
     /** {@inheritDoc} */
-    @Deprecated
     @Override public boolean stopProcess() {
         return false;
     }
diff --git a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveStartMessage.java b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveStartMessage.java
index 4c8d996..0c79c36 100644
--- a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveStartMessage.java
+++ b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkCommunicationErrorResolveStartMessage.java
@@ -50,7 +50,6 @@
     }
 
     /** {@inheritDoc} */
-    @Deprecated
     @Override public boolean stopProcess() {
         return false;
     }
diff --git a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkForceNodeFailMessage.java b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkForceNodeFailMessage.java
index bd6e1e8..de7291c 100644
--- a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkForceNodeFailMessage.java
+++ b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkForceNodeFailMessage.java
@@ -54,7 +54,6 @@
     }
 
     /** {@inheritDoc} */
-    @Deprecated
     @Override public boolean stopProcess() {
         return false;
     }
diff --git a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkNoServersMessage.java b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkNoServersMessage.java
index 957310d..626fe74 100644
--- a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkNoServersMessage.java
+++ b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZkNoServersMessage.java
@@ -39,7 +39,6 @@
     }
 
     /** {@inheritDoc} */
-    @Deprecated
     @Override public boolean stopProcess() {
         return false;
     }
diff --git a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
index 97f7b86..5225c9a 100644
--- a/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
+++ b/modules/zookeeper/src/main/java/org/apache/ignite/spi/discovery/zk/internal/ZookeeperDiscoveryImpl.java
@@ -2588,11 +2588,39 @@
         if (log.isDebugEnabled())
             log.debug("Generated CUSTOM event [evt=" + evtData + ", msg=" + msg + ']');
 
+        boolean fastStopProcess = false;
+
         if (msg instanceof ZkInternalMessage)
             processInternalMessage(evtData, (ZkInternalMessage)msg);
-        else
+        else {
             notifyCustomEvent(evtData, msg);
 
+            if (msg.stopProcess()) {
+                if (log.isDebugEnabled())
+                    log.debug("Fast stop process custom event [evt=" + evtData + ", msg=" + msg + ']');
+
+                fastStopProcess = true;
+
+                // No need to process this event on others nodes, skip this event.
+                evtsData.evts.remove(evtData.eventId());
+
+                evtsData.evtIdGen--;
+
+                DiscoverySpiCustomMessage ack = msg.ackMessage();
+
+                if (ack != null) {
+                    evtData = createAckEvent(ack, evtData);
+
+                    if (log.isDebugEnabled())
+                        log.debug("Generated CUSTOM event (ack for fast stop process) [evt=" + evtData + ", msg=" + msg + ']');
+
+                    notifyCustomEvent(evtData, ack);
+                }
+                else
+                    evtData = null;
+            }
+        }
+
         if (evtData != null) {
             evtsData.addEvent(rtState.top.nodesByOrder.values(), evtData);
 
@@ -2600,6 +2628,9 @@
 
             saveAndProcessNewEvents();
 
+            if (fastStopProcess)
+                deleteCustomEventDataAsync(zkClient, evtPath);
+
             if (failedNode != null) {
                 deleteAliveNode(failedNode.internalId());