ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [3/4] ignite git commit: ignite-5578
Date Thu, 03 Aug 2017 12:55:43 GMT
ignite-5578


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/e3ecc169
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/e3ecc169
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/e3ecc169

Branch: refs/heads/ignite-5578
Commit: e3ecc169ceab543748c9ba3ef409540697cf2529
Parents: b1579ea
Author: sboikov <sboikov@gridgain.com>
Authored: Thu Aug 3 15:45:23 2017 +0300
Committer: sboikov <sboikov@gridgain.com>
Committed: Thu Aug 3 15:45:23 2017 +0300

----------------------------------------------------------------------
 .../cache/CacheAffinitySharedManager.java       | 140 ++++++++-----------
 .../processors/cache/ExchangeContext.java       |   4 +-
 .../cache/ExchangeDiscoveryEvents.java          |   6 +-
 .../GridCachePartitionExchangeManager.java      |  26 ++--
 .../dht/GridClientPartitionTopology.java        |   2 +-
 .../dht/GridDhtPartitionTopologyImpl.java       |   4 +-
 .../GridDhtPartitionsExchangeFuture.java        | 139 +++++++++---------
 .../dht/preloader/InitNewCoordinatorFuture.java |   2 +-
 .../GridCacheDatabaseSharedManager.java         |   2 +-
 .../CacheLateAffinityAssignmentTest.java        |   2 +-
 10 files changed, 154 insertions(+), 173 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheAffinitySharedManager.java
----------------------------------------------------------------------
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 e6f5d6b..dc905a0 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
@@ -759,7 +759,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                         evts.topologyVersion());
 
                     if (fut.cacheAddedOnExchange(cacheDesc.cacheId(), cacheDesc.receivedFrom())) {
-                        if (fut.discoCache().cacheGroupAffinityNodes(cacheDesc.groupId()).isEmpty())
+                        if (fut.events().discoveryCache().cacheGroupAffinityNodes(cacheDesc.groupId()).isEmpty())
                             U.quietAndWarn(log, "No server nodes found for cache client: " + req.cacheName());
                     }
                 }
@@ -957,7 +957,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                 IgniteUuid deploymentId = desc.deploymentId();
 
                 if (!deploymentId.equals(deploymentIds.get(aff.groupId()))) {
-                    aff.clientEventTopologyChange(exchFut.discoveryEvent(), topVer);
+                    aff.clientEventTopologyChange(exchFut.firstEvent(), topVer);
 
                     return;
                 }
@@ -991,7 +991,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                     aff.initialize(topVer, cachedAssignment(aff, assignment, affCache));
                 }
                 else
-                    aff.clientEventTopologyChange(exchFut.discoveryEvent(), topVer);
+                    aff.clientEventTopologyChange(exchFut.firstEvent(), topVer);
             }
         });
     }
@@ -1004,14 +1004,14 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
      * @throws IgniteCheckedException If failed.
      */
     public void onClientEvent(final GridDhtPartitionsExchangeFuture fut, boolean crd) throws IgniteCheckedException {
-        boolean locJoin = fut.discoveryEvent().eventNode().isLocal();
+        boolean locJoin = fut.firstEvent().eventNode().isLocal();
 
         if (!locJoin) {
             forAllCacheGroups(crd, new IgniteInClosureX<GridAffinityAssignmentCache>() {
                 @Override public void applyx(GridAffinityAssignmentCache aff) throws IgniteCheckedException {
                     AffinityTopologyVersion topVer = fut.initialVersion();
 
-                    aff.clientEventTopologyChange(fut.discoveryEvent(), topVer);
+                    aff.clientEventTopologyChange(fut.firstEvent(), topVer);
                 }
             });
         }
@@ -1111,11 +1111,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
             assert old == null : old;
 
-            List<List<ClusterNode>> newAff = grpHolder.affinity().calculate(fut.initialVersion(),
-                fut.discoveryEvent(),
-                fut.discoCache());
-
-            grpHolder.affinity().initialize(fut.initialVersion(), newAff);
+            calculateAndInit(fut.events(), grpHolder.affinity(), fut.initialVersion());
         }
         else if (grpHolder.client() && grp != null) {
             assert grpHolder.affinity().idealAssignment() != null;
@@ -1149,12 +1145,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                 @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException {
                     CacheGroupHolder cache = groupHolder(fut.initialVersion(), desc);
 
-                    if (cache.affinity().lastVersion().equals(AffinityTopologyVersion.NONE)) {
-                        List<List<ClusterNode>> assignment =
-                            cache.affinity().calculate(fut.initialVersion(), fut.discoveryEvent(), fut.discoCache());
-
-                        cache.affinity().initialize(fut.initialVersion(), assignment);
-                    }
+                    if (cache.affinity().lastVersion().equals(AffinityTopologyVersion.NONE))
+                        calculateAndInit(fut.events(), cache.affinity(), fut.initialVersion());
                 }
             });
         }
@@ -1182,13 +1174,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
         ExchangeDiscoveryEvents evts = fut.context().events();
 
-        if (canCalculateAffinity(desc, aff, fut)) {
-            List<List<ClusterNode>> assignment = aff.calculate(evts.topologyVersion(),
-                evts.lastEvent(),
-                evts.discoveryCache());
-
-            aff.initialize(evts.topologyVersion(), assignment);
-        }
+        if (canCalculateAffinity(desc, aff, fut))
+            calculateAndInit(evts, aff, evts.topologyVersion());
         else {
             GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
                 desc.groupId(),
@@ -1220,7 +1207,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
             return true;
 
         // If local node did not initiate exchange or local node is the only cache node in grid.
-        Collection<ClusterNode> affNodes = fut.discoCache().cacheGroupAffinityNodes(aff.groupId());
+        Collection<ClusterNode> affNodes = fut.events().discoveryCache().cacheGroupAffinityNodes(aff.groupId());
 
         return fut.cacheGroupAddedOnExchange(aff.groupId(), desc.receivedFrom()) ||
             !fut.exchangeId().nodeId().equals(cctx.localNodeId()) ||
@@ -1330,13 +1317,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
                     aff.initialize(evts.topologyVersion(), assignments);
                 }
-                else if (fut.cacheGroupAddedOnExchange(aff.groupId(), grp.receivedFrom())) {
-                    List<List<ClusterNode>> assignment = aff.calculate(evts.topologyVersion(),
-                        evts.lastEvent(),
-                        evts.discoveryCache());
-
-                    aff.initialize(evts.topologyVersion(), assignment);
-                }
+                else if (fut.cacheGroupAddedOnExchange(aff.groupId(), grp.receivedFrom()))
+                    calculateAndInit(evts, aff, evts.topologyVersion());
 
                 grp.topology().initPartitionsWhenAffinityReady(resTopVer, fut);
             }
@@ -1412,9 +1394,9 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
      * @throws IgniteCheckedException If failed.
      */
     public void onServerJoin(final GridDhtPartitionsExchangeFuture fut, boolean crd) throws IgniteCheckedException {
-        assert !fut.discoveryEvent().eventNode().isClient();
+        assert !fut.firstEvent().eventNode().isClient();
 
-        boolean locJoin = fut.discoveryEvent().eventNode().isLocal();
+        boolean locJoin = fut.firstEvent().eventNode().isLocal();
 
         WaitRebalanceInfo waitRebalanceInfo = null;
 
@@ -1424,13 +1406,9 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                     @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException {
                         AffinityTopologyVersion topVer = fut.initialVersion();
 
-                        CacheGroupHolder cache = groupHolder(topVer, desc);
-
-                        List<List<ClusterNode>> newAff = cache.affinity().calculate(topVer,
-                            fut.discoveryEvent(),
-                            fut.discoCache());
+                        CacheGroupHolder grpHolder = groupHolder(topVer, desc);
 
-                        cache.affinity().initialize(topVer, newAff);
+                        calculateAndInit(fut.events(), grpHolder.affinity(), topVer);
                     }
                 });
             }
@@ -1485,6 +1463,22 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
     }
 
     /**
+     * @param evts Discovery events.
+     * @param aff Affinity.
+     * @param topVer Topology version.
+     */
+    private void calculateAndInit(ExchangeDiscoveryEvents evts,
+        GridAffinityAssignmentCache aff,
+        AffinityTopologyVersion topVer)
+    {
+        List<List<ClusterNode>> assignment = aff.calculate(topVer,
+            evts.lastEvent(),
+            evts.discoveryCache());
+
+        aff.initialize(topVer, assignment);
+    }
+
+    /**
      * @param fut Exchange future.
      * @throws IgniteCheckedException If failed.
      */
@@ -1498,14 +1492,9 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                 continue;
 
             if (fut.cacheGroupAddedOnExchange(grp.groupId(), grp.receivedFrom())) {
-                // Do not calculate affinity since it can change in case of exchange merge.
-                if (!fut.context().mergeExchanges()) {
-                    List<List<ClusterNode>> assignment = grp.affinity().calculate(topVer,
-                            fut.discoveryEvent(),
-                            fut.discoCache());
-
-                    grp.affinity().initialize(topVer, assignment);
-                }
+                // In case if merge is allowed do not calculate affinity since it can change on exchange end.
+                if (!fut.context().mergeExchanges())
+                    calculateAndInit(fut.events(), grp.affinity(), topVer);
             }
             else {
                 if (fut.context().fetchAffinityOnJoin()) {
@@ -1516,22 +1505,17 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                     GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
                         grpDesc.groupId(),
                         topVer,
-                        fut.discoCache());
+                        fut.events().discoveryCache());
 
                     fetchFut.init(false);
 
                     fetchFuts.add(fetchFut);
                 }
                 else {
-                    if (fut.discoCache().serverNodes().size() > 0)
+                    if (fut.events().discoveryCache().serverNodes().size() > 0)
                         fut.context().addGroupAffinityRequestOnJoin(grp.groupId());
-                    else {
-                        List<List<ClusterNode>> aff = grp.affinity().calculate(topVer,
-                            fut.discoveryEvent(),
-                            fut.discoCache());
-
-                        grp.affinity().initialize(topVer, aff);
-                    }
+                    else
+                        calculateAndInit(fut.events(), grp.affinity(), topVer);
                 }
             }
         }
@@ -1542,8 +1526,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
             int grpId = fetchFut.groupId();
 
             fetchAffinity(topVer,
-                fut.discoveryEvent(),
-                fut.discoCache(),
+                fut.events().lastEvent(),
+                fut.events().discoveryCache(),
                 cctx.cache().cacheGroup(grpId).affinity(),
                 fetchFut);
         }
@@ -1603,7 +1587,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
      * @return {@code True} if affinity should be assigned by coordinator.
      */
     public boolean onServerLeft(final GridDhtPartitionsExchangeFuture fut, boolean crd) throws IgniteCheckedException {
-        ClusterNode leftNode = fut.discoveryEvent().eventNode();
+        ClusterNode leftNode = fut.firstEvent().eventNode();
 
         assert !leftNode.isClient() : leftNode;
 
@@ -1613,14 +1597,14 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                 @Override public void applyx(CacheGroupDescriptor desc) throws IgniteCheckedException {
                     CacheGroupHolder cache = groupHolder(fut.initialVersion(), desc);
 
-                    cache.aff.calculate(fut.initialVersion(), fut.discoveryEvent(), fut.discoCache());
+                    cache.aff.calculate(fut.initialVersion(), fut.firstEvent(), fut.firstEventCache());
                 }
             });
         }
         else {
             forAllCacheGroups(false, new IgniteInClosureX<GridAffinityAssignmentCache>() {
                 @Override public void applyx(GridAffinityAssignmentCache aff) throws IgniteCheckedException {
-                    aff.calculate(fut.initialVersion(), fut.discoveryEvent(), fut.discoCache());
+                    aff.calculate(fut.initialVersion(), fut.firstEvent(), fut.firstEventCache());
                 }
             });
         }
@@ -1670,12 +1654,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                     final GridAffinityAssignmentCache aff = grpHolder.affinity();
 
                     if (newAff) {
-                        if (!aff.lastVersion().equals(topVer)) {
-                            List<List<ClusterNode>> assign =
-                                aff.calculate(topVer, fut.discoveryEvent(), fut.discoCache());
-
-                            aff.initialize(topVer, assign);
-                        }
+                        if (!aff.lastVersion().equals(topVer))
+                            calculateAndInit(fut.events(), aff, topVer);
 
                         grpHolder.topology().beforeExchange(fut, true, false);
                     }
@@ -1700,7 +1680,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                         GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
                             desc.groupId(),
                             prev.topologyVersion(),
-                            prev.discoCache());
+                            prev.events().discoveryCache());
 
                         fetchFut.init(false);
 
@@ -1710,12 +1690,12 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                             @Override public void applyx(IgniteInternalFuture<GridDhtAffinityAssignmentResponse> fetchFut)
                                 throws IgniteCheckedException {
                                 fetchAffinity(prev.topologyVersion(),
-                                    prev.discoveryEvent(),
-                                    prev.discoCache(),
+                                    prev.events().lastEvent(),
+                                    prev.events().discoveryCache(),
                                     aff,
                                     (GridDhtAssignmentFetchFuture)fetchFut);
 
-                                aff.calculate(topVer, fut.discoveryEvent(), fut.discoCache());
+                                aff.calculate(topVer, fut.events().lastEvent(), fut.events().discoveryCache());
 
                                 affFut.onDone(topVer);
                             }
@@ -1730,13 +1710,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                     if (newAff) {
                         GridAffinityAssignmentCache aff = grpHolder.affinity();
 
-                        if (!aff.lastVersion().equals(topVer)) {
-                            List<List<ClusterNode>> assign = aff.calculate(topVer,
-                                fut.discoveryEvent(),
-                                fut.discoCache());
-
-                            aff.initialize(topVer, assign);
-                        }
+                        if (!aff.lastVersion().equals(topVer))
+                            calculateAndInit(fut.events(), aff, topVer);
 
                         grpHolder.topology().beforeExchange(fut, true, false);
                     }
@@ -1908,13 +1883,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
         throws IgniteCheckedException
     {
         if (addedOnExchnage) {
-            if (!aff.lastVersion().equals(evts.topologyVersion())) {
-                List<List<ClusterNode>> newAff = aff.calculate(evts.topologyVersion(),
-                    evts.lastEvent(),
-                    evts.discoveryCache());
-
-                aff.initialize(evts.topologyVersion(), newAff);
-            }
+            if (!aff.lastVersion().equals(evts.topologyVersion()))
+                calculateAndInit(evts, aff, evts.topologyVersion());
 
             return;
         }

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
index 94040ba..6442752 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeContext.java
@@ -56,7 +56,7 @@ public class ExchangeContext {
      * @param fut Exchange future.
      */
     public ExchangeContext(boolean crd, GridDhtPartitionsExchangeFuture fut) {
-        int protocolVer = exchangeProtocolVersion(fut.discoCache().minimumNodeVersion());
+        int protocolVer = exchangeProtocolVersion(fut.firstEventCache().minimumNodeVersion());
 
         if (compatibilityNode || (crd && fut.localJoinExchange())) {
             fetchAffOnJoin = true;
@@ -71,7 +71,7 @@ public class ExchangeContext {
 
             merge = !startCaches &&
                 protocolVer > 1 &&
-                fut.discoveryEvent().type() != EVT_DISCOVERY_CUSTOM_EVT;
+                fut.firstEvent().type() != EVT_DISCOVERY_CUSTOM_EVT;
         }
 
         evts = new ExchangeDiscoveryEvents(fut);

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java
index ff61701..d4fbe60 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java
@@ -71,7 +71,7 @@ public class ExchangeDiscoveryEvents {
      * @param fut Current exchange future.
      */
     ExchangeDiscoveryEvents(GridDhtPartitionsExchangeFuture fut) {
-        addEvent(fut.initialVersion(), fut.discoveryEvent(), fut.discoCache());
+        addEvent(fut.initialVersion(), fut.firstEvent(), fut.firstEventCache());
     }
 
     /**
@@ -106,7 +106,7 @@ public class ExchangeDiscoveryEvents {
      * @return Last server join/fail event version.
      */
     AffinityTopologyVersion lastServerEventVersion() {
-        assert srvEvtTopVer != null;
+        assert srvEvtTopVer != null : this;
 
         return srvEvtTopVer;
     }
@@ -169,7 +169,7 @@ public class ExchangeDiscoveryEvents {
     /**
      * @return Last event.
      */
-    DiscoveryEvent lastEvent() {
+    public DiscoveryEvent lastEvent() {
         return lastSrvEvt != null ? lastSrvEvt : lastEvt;
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 6efd19e..75ddacd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -1796,15 +1796,15 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
 
                 log.info("Merge exchange future on finish [curFut=" + curFut.initialVersion() +
                     ", mergedFut=" + fut.initialVersion() +
-                    ", evt=" + IgniteUtils.gridEventName(fut.discoveryEvent().type()) +
-                    ", evtNode=" + fut.discoveryEvent().eventNode().id()+
-                    ", evtNodeClient=" + CU.clientNode(fut.discoveryEvent().eventNode())+ ']');
+                    ", evt=" + IgniteUtils.gridEventName(fut.firstEvent().type()) +
+                    ", evtNode=" + fut.firstEvent().eventNode().id()+
+                    ", evtNodeClient=" + CU.clientNode(fut.firstEvent().eventNode())+ ']');
 
-                DiscoveryEvent evt = fut.discoveryEvent();
+                DiscoveryEvent evt = fut.firstEvent();
 
                 curFut.context().events().addEvent(fut.initialVersion(),
-                    fut.discoveryEvent(),
-                    fut.discoCache());
+                    fut.firstEvent(),
+                    fut.firstEventCache());
 
                 if (evt.type() == EVT_NODE_JOINED) {
                     final GridDhtPartitionsSingleMessage pendingMsg = fut.mergeJoinExchangeOnDone(curFut);
@@ -1888,7 +1888,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
                 if (task instanceof GridDhtPartitionsExchangeFuture) {
                     GridDhtPartitionsExchangeFuture fut = (GridDhtPartitionsExchangeFuture)task;
 
-                    DiscoveryEvent evt = fut.discoveryEvent();
+                    DiscoveryEvent evt = fut.firstEvent();
 
                     if (evt.type() == EVT_DISCOVERY_CUSTOM_EVT) {
                         log.info("Stop merge, custom event found: " + evt);
@@ -1911,13 +1911,13 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
 
                     log.info("Merge exchange future [curFut=" + curFut.initialVersion() +
                         ", mergedFut=" + fut.initialVersion() +
-                        ", evt=" + IgniteUtils.gridEventName(fut.discoveryEvent().type()) +
-                        ", evtNode=" + fut.discoveryEvent().eventNode().id() +
-                        ", evtNodeClient=" + CU.clientNode(fut.discoveryEvent().eventNode())+ ']');
+                        ", evt=" + IgniteUtils.gridEventName(fut.firstEvent().type()) +
+                        ", evtNode=" + fut.firstEvent().eventNode().id() +
+                        ", evtNodeClient=" + CU.clientNode(fut.firstEvent().eventNode())+ ']');
 
                     curFut.context().events().addEvent(fut.initialVersion(),
-                        fut.discoveryEvent(),
-                        fut.discoCache());
+                        fut.firstEvent(),
+                        fut.firstEventCache());
 
                     if (evt.type() == EVT_NODE_JOINED) {
                         if (fut.mergeJoinExchange(curFut))
@@ -2214,7 +2214,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
                             boolean newCrd = false;
 
                             if (!crd) {
-                                List<ClusterNode> srvNodes = exchFut.discoCache().serverNodes();
+                                List<ClusterNode> srvNodes = exchFut.firstEventCache().serverNodes();
 
                                 crd = newCrd = !srvNodes.isEmpty() && srvNodes.get(0).isLocal();
                             }

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
index 5a16bc9..12f3fb3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridClientPartitionTopology.java
@@ -258,7 +258,7 @@ public class GridClientPartitionTopology implements GridDhtPartitionTopology {
             if (stopping)
                 return;
 
-            discoCache = exchFut.discoCache();
+            discoCache = exchFut.events().discoveryCache();
 
             beforeExchange0(loc, exchFut);
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
index e99ad2c..5e58502 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtPartitionTopologyImpl.java
@@ -518,7 +518,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
 
                     if (grp.affinityNode()) {
                         if (grpStarted ||
-                            exchFut.discoveryEvent().type() == EVT_DISCOVERY_CUSTOM_EVT ||
+                            exchFut.firstEvent().type() == EVT_DISCOVERY_CUSTOM_EVT ||
                             exchFut.serverNodeDiscoveryEvent()) {
                             if (affReady) {
                                 assert grp.affinity().lastVersion().equals(evts.topologyVersion());
@@ -626,7 +626,7 @@ public class GridDhtPartitionTopologyImpl implements GridDhtPartitionTopology {
                                 changed = true;
 
                                 if (grp.eventRecordable(EVT_CACHE_REBALANCE_PART_DATA_LOST)) {
-                                    DiscoveryEvent discoEvt = exchFut.discoveryEvent();
+                                    DiscoveryEvent discoEvt = exchFut.events().lastEvent();
 
                                     grp.addRebalanceEvent(p,
                                         EVT_CACHE_REBALANCE_PART_DATA_LOST,

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index be6acf4..3d7749b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -125,10 +125,10 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
     /** */
     @GridToStringExclude
-    private volatile DiscoCache discoCache;
+    private volatile DiscoCache firstEvtDiscoCache;
 
-    /** Discovery event. */
-    private volatile DiscoveryEvent discoEvt;
+    /** Discovery event triggered this exchange. */
+    private volatile DiscoveryEvent firstDiscoEvt;
 
     /** */
     @GridToStringExclude
@@ -174,7 +174,10 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     /** Last committed cache version before next topology version use. */
     private AtomicReference<GridCacheVersion> lastVer = new AtomicReference<>();
 
-    /** */
+    /**
+     * Message received from node joining cluster, needed if this join exchange
+     * is merged with previous one.
+     */
     @GridToStringExclude
     private GridDhtPartitionsSingleMessage pendingJoinMsg;
 
@@ -222,11 +225,11 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     /** */
     private ConcurrentMap<UUID, GridDhtPartitionsSingleMessage> msgs = new ConcurrentHashMap8<>();
 
-    /** */
+    /** Single messages from merged 'node join' exchanges. */
     @GridToStringExclude
     private Map<UUID, GridDhtPartitionsSingleMessage> mergedJoinExchMsgs;
 
-    /** */
+    /** Number of awaited messages for merged 'node join' exchanges. */
     @GridToStringExclude
     private int awaitMergedMsgs;
 
@@ -255,7 +258,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     @GridToStringExclude
     private FinishState finishState;
 
-    /** */
+    /** Initialized when node becomes new coordinator. */
     @GridToStringExclude
     private InitNewCoordinatorFuture newCrdFut;
 
@@ -371,15 +374,6 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     }
 
     /**
-     * @return Discovery cache.
-     *
-     * TODO 5578 review usages, rename initialDiscoveryEvent
-     */
-    public DiscoCache discoCache() {
-        return discoCache;
-    }
-
-    /**
      * @param cacheId Cache ID.
      * @param rcvdFrom Node ID cache was received from.
      * @return {@code True} if cache was added during this exchange.
@@ -431,8 +425,8 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
         assert exchId.equals(this.exchId);
 
         this.exchId.discoveryEvent(discoEvt);
-        this.discoEvt = discoEvt;
-        this.discoCache = discoCache;
+        this.firstDiscoEvt= discoEvt;
+        this.firstEvtDiscoCache = discoCache;
 
         evtLatch.countDown();
     }
@@ -459,12 +453,25 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     }
 
     /**
-     * @return Discovery event.
+     * @return First event discovery event.
      *
-     * TODO 5578 review usages, rename initialDiscoveryEvent
      */
-    public DiscoveryEvent discoveryEvent() {
-        return discoEvt;
+    public DiscoveryEvent firstEvent() {
+        return firstDiscoEvt;
+    }
+
+    /**
+     * @return Discovery cache for first event.
+     */
+    public DiscoCache firstEventCache() {
+        return firstEvtDiscoCache;
+    }
+
+    /**
+     * @return Events processed in this exchange.
+     */
+    public ExchangeDiscoveryEvents events() {
+        return exchCtx.events();
     }
 
     /**
@@ -525,13 +532,13 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
         U.await(evtLatch);
 
-        assert discoEvt != null : this;
-        assert exchId.nodeId().equals(discoEvt.eventNode().id()) : this;
+        assert firstDiscoEvt != null : this;
+        assert exchId.nodeId().equals(firstDiscoEvt.eventNode().id()) : this;
 
         try {
             AffinityTopologyVersion topVer = initialVersion();
 
-            srvNodes = new ArrayList<>(discoCache.serverNodes());
+            srvNodes = new ArrayList<>(firstEvtDiscoCache.serverNodes());
 
             remaining.addAll(F.nodeIds(F.view(srvNodes, F.remoteNodes(cctx.localNodeId()))));
 
@@ -551,18 +558,18 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
             if (exchLog.isInfoEnabled()) {
                 exchLog.info("Started exchange init [topVer=" + topVer +
                     ", crd=" + crdNode +
-                    ", evt=" + IgniteUtils.gridEventName(discoEvt.type()) +
-                    ", evtNode=" + discoEvt.eventNode().id() +
-                    ", customEvt=" + (discoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT ? ((DiscoveryCustomEvent) discoEvt).customMessage() : null) +
+                    ", evt=" + IgniteUtils.gridEventName(firstDiscoEvt.type()) +
+                    ", evtNode=" + firstDiscoEvt.eventNode().id() +
+                    ", customEvt=" + (firstDiscoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT ? ((DiscoveryCustomEvent)firstDiscoEvt).customMessage() : null) +
                     ", allowMerge=" + exchCtx.mergeExchanges() + ']');
             }
 
             ExchangeType exchange;
 
-            if (discoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT) {
+            if (firstDiscoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT) {
                 assert !exchCtx.mergeExchanges();
 
-                DiscoveryCustomMessage msg = ((DiscoveryCustomEvent)discoEvt).customMessage();
+                DiscoveryCustomMessage msg = ((DiscoveryCustomEvent)firstDiscoEvt).customMessage();
 
                 if (msg instanceof ChangeGlobalStateMessage) {
                     assert exchActions != null && !exchActions.empty();
@@ -575,7 +582,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
                     exchange = onCacheChangeRequest(crdNode);
                 }
                 else if (msg instanceof SnapshotDiscoveryMessage) {
-                    exchange = CU.clientNode(discoEvt.eventNode()) ?
+                    exchange = CU.clientNode(firstDiscoEvt.eventNode()) ?
                         onClientNodeEvent(crdNode) :
                         onServerNodeEvent(crdNode);
                 }
@@ -588,10 +595,10 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
                 initCoordinatorCaches(newCrd);
             }
             else {
-                if (discoEvt.type() == EVT_NODE_JOINED) {
-                    if (!discoEvt.eventNode().isLocal()) {
+                if (firstDiscoEvt.type() == EVT_NODE_JOINED) {
+                    if (!firstDiscoEvt.eventNode().isLocal()) {
                         Collection<DynamicCacheDescriptor> receivedCaches = cctx.cache().startReceivedCaches(
-                            discoEvt.eventNode().id(),
+                            firstDiscoEvt.eventNode().id(),
                             topVer);
 
                         cctx.affinity().initStartedCaches(crdNode, this, receivedCaches);
@@ -616,7 +623,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
                         }
                     }
                     else {
-                        if (CU.clientNode(discoEvt.eventNode()))
+                        if (CU.clientNode(firstDiscoEvt.eventNode()))
                             exchange = onClientNodeEvent(crdNode);
                         else
                             exchange = cctx.kernalContext().clientNode() ? ExchangeType.CLIENT : ExchangeType.ALL;
@@ -626,7 +633,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
                         onLeft();
                 }
                 else {
-                    exchange = CU.clientNode(discoEvt.eventNode()) ? onClientNodeEvent(crdNode) :
+                    exchange = CU.clientNode(firstDiscoEvt.eventNode()) ? onClientNodeEvent(crdNode) :
                         onServerNodeEvent(crdNode);
                 }
             }
@@ -761,13 +768,13 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
             top.updateTopologyVersion(
                 this,
-                discoCache(),
+                events().discoveryCache(),
                 updSeq,
                 cacheGroupStopping(grp.groupId()));
         }
 
         for (GridClientPartitionTopology top : cctx.exchange().clientTopologies())
-            top.updateTopologyVersion(this, discoCache(), -1, cacheGroupStopping(top.groupId()));
+            top.updateTopologyVersion(this, events().discoveryCache(), -1, cacheGroupStopping(top.groupId()));
     }
 
     /**
@@ -892,19 +899,19 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
      * @return Exchange type.
      */
     private ExchangeType onClientNodeEvent(boolean crd) throws IgniteCheckedException {
-        assert CU.clientNode(discoEvt.eventNode()) : this;
+        assert CU.clientNode(firstDiscoEvt.eventNode()) : this;
 
-        if (discoEvt.type() == EVT_NODE_LEFT || discoEvt.type() == EVT_NODE_FAILED) {
+        if (firstDiscoEvt.type() == EVT_NODE_LEFT || firstDiscoEvt.type() == EVT_NODE_FAILED) {
             onLeft();
 
-            assert !discoEvt.eventNode().isLocal() : discoEvt;
+            assert !firstDiscoEvt.eventNode().isLocal() : firstDiscoEvt;
         }
         else
-            assert discoEvt.type() == EVT_NODE_JOINED || discoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT : discoEvt;
+            assert firstDiscoEvt.type() == EVT_NODE_JOINED || firstDiscoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT : firstDiscoEvt;
 
         cctx.affinity().onClientEvent(this, crd);
 
-        return discoEvt.eventNode().isLocal() ? ExchangeType.CLIENT : ExchangeType.NONE;
+        return firstDiscoEvt.eventNode().isLocal() ? ExchangeType.CLIENT : ExchangeType.NONE;
     }
 
     /**
@@ -913,9 +920,9 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
      * @return Exchange type.
      */
     private ExchangeType onServerNodeEvent(boolean crd) throws IgniteCheckedException {
-        assert !CU.clientNode(discoEvt.eventNode()) : this;
+        assert !CU.clientNode(firstDiscoEvt.eventNode()) : this;
 
-        if (discoEvt.type() == EVT_NODE_LEFT || discoEvt.type() == EVT_NODE_FAILED) {
+        if (firstDiscoEvt.type() == EVT_NODE_LEFT || firstDiscoEvt.type() == EVT_NODE_FAILED) {
             onLeft();
 
             exchCtx.events().warnNoAffinityNodes(cctx);
@@ -979,7 +986,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
         waitPartitionRelease();
 
-        boolean topChanged = discoEvt.type() != EVT_DISCOVERY_CUSTOM_EVT || affChangeMsg != null;
+        boolean topChanged = firstDiscoEvt.type() != EVT_DISCOVERY_CUSTOM_EVT || affChangeMsg != null;
 
         for (GridCacheContext cacheCtx : cctx.cacheContexts()) {
             if (cacheCtx.isLocal() || cacheStopping(cacheCtx.cacheId()))
@@ -1021,7 +1028,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
         try {
             long start = U.currentTimeMillis();
 
-            IgniteInternalFuture fut = cctx.snapshot().tryStartLocalSnapshotOperation(discoEvt);
+            IgniteInternalFuture fut = cctx.snapshot().tryStartLocalSnapshotOperation(events().lastEvent());
 
             if (fut != null) {
                 fut.get();
@@ -1190,7 +1197,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
      * @return {@code True} if exchange for local node join.
      */
     public boolean localJoinExchange() {
-        return discoEvt.type() == EVT_NODE_JOINED && discoEvt.eventNode().isLocal();
+        return firstDiscoEvt.type() == EVT_NODE_JOINED && firstDiscoEvt.eventNode().isLocal();
     }
 
     /**
@@ -1423,7 +1430,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
             Map<Integer, CacheValidation> m = U.newHashMap(cctx.cache().cacheGroups().size());
 
             for (CacheGroupContext grp : cctx.cache().cacheGroups())
-                m.put(grp.groupId(), validateCacheGroup(grp, discoEvt.topologyNodes()));
+                m.put(grp.groupId(), validateCacheGroup(grp, events().lastEvent().topologyNodes()));
 
             grpValidRes = m;
         }
@@ -1483,8 +1490,8 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
             exchActions = null;
 
-            if (discoEvt instanceof DiscoveryCustomEvent)
-                ((DiscoveryCustomEvent)discoEvt).customMessage(null);
+            if (firstDiscoEvt instanceof DiscoveryCustomEvent)
+                ((DiscoveryCustomEvent)firstDiscoEvt).customMessage(null);
 
             if (err == null)
                 cctx.exchange().lastFinishedFuture(this);
@@ -1528,9 +1535,11 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     }
 
     /**
+     * Records that this exchange if merged with another 'node join' exchange.
+     *
      * @param node Joined node.
      * @param msg Joined node message if already received.
-     * @return {@code True} if need wait for message from joined server node.
+     * @return {@code True} if need to wait for message from joined server node.
      */
     private boolean addMergedJoinExchange(ClusterNode node, @Nullable GridDhtPartitionsSingleMessage msg) {
         assert Thread.holdsLock(mux);
@@ -1582,6 +1591,8 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     }
 
     /**
+     * Merges this exchange with given one.
+     *
      * @param fut Current exchange to merge with.
      * @return {@code True} if need wait for message from joined server node.
      */
@@ -1596,7 +1607,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
             mergedWith = fut;
 
-            ClusterNode joinedNode = discoEvt.eventNode();
+            ClusterNode joinedNode = firstDiscoEvt.eventNode();
 
             wait = fut.addMergedJoinExchange(joinedNode, pendingJoinMsg);
         }
@@ -1784,7 +1795,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
                 }
 
                 if (finishState0 == null) {
-                    assert discoEvt.type() == EVT_NODE_JOINED && CU.clientNode(discoEvt.eventNode()) : this;
+                    assert firstDiscoEvt.type() == EVT_NODE_JOINED && CU.clientNode(firstDiscoEvt.eventNode()) : this;
 
                     finishState0 = new FinishState(cctx.localNodeId(),
                         initialVersion(),
@@ -2093,7 +2104,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
             for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
                 if (!grp.isLocal()) {
-                    boolean detectedOnGrp = grp.topology().detectLostPartitions(resTopVer, discoEvt);
+                    boolean detectedOnGrp = grp.topology().detectLostPartitions(resTopVer, events().lastEvent());
 
                     detected |= detectedOnGrp;
                 }
@@ -2130,7 +2141,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     }
 
     /**
-     *
+     * @param sndResNodes Additional nodes to send finish message to.
      */
     private void onAllReceived(@Nullable Collection<ClusterNode> sndResNodes) {
         try {
@@ -2138,7 +2149,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
             assert partHistSuppliers.isEmpty() : partHistSuppliers;
 
-            if (!crd.equals(discoCache.serverNodes().get(0)) && !exchCtx.mergeExchanges()) {
+            if (!exchCtx.mergeExchanges() && !crd.equals(events().discoveryCache().serverNodes().get(0))) {
                 for (CacheGroupContext grp : cctx.cache().cacheGroups()) {
                     if (!grp.isLocal())
                         grp.topology().beforeExchange(this, !centralizedAff, false);
@@ -2165,7 +2176,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     }
 
     /**
-     *
+     * @param sndResNodes Additional nodes to send finish message to.
      */
     private void finishExchangeOnCoordinator(@Nullable Collection<ClusterNode> sndResNodes) {
         try {
@@ -2239,13 +2250,13 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
                 }
             }
 
-            if (discoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT) {
-                assert discoEvt instanceof DiscoveryCustomEvent;
+            if (firstDiscoEvt.type() == EVT_DISCOVERY_CUSTOM_EVT) {
+                assert firstDiscoEvt instanceof DiscoveryCustomEvent;
 
                 if (activateCluster())
                     assignPartitionsStates();
 
-                if (((DiscoveryCustomEvent)discoEvt).customMessage() instanceof DynamicCacheChangeBatch) {
+                if (((DiscoveryCustomEvent)firstDiscoEvt).customMessage() instanceof DynamicCacheChangeBatch) {
                     if (exchActions != null) {
                         Set<String> caches = exchActions.cachesToResetLostPartitions();
 
@@ -2906,7 +2917,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
 
                         ClusterNode crd0;
 
-                        discoCache.updateAlives(node);
+                        events().discoveryCache().updateAlives(node);
 
                         InitNewCoordinatorFuture newCrdFut0;
 
@@ -3269,8 +3280,8 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
      */
     public String shortInfo() {
         return "GridDhtPartitionsExchangeFuture [topVer=" + initialVersion() +
-            ", evt=" + (discoEvt != null ? IgniteUtils.gridEventName(discoEvt.type()) : -1) +
-            ", evtNode=" + (discoEvt != null ? discoEvt.eventNode() : null) +
+            ", evt=" + (firstDiscoEvt != null ? IgniteUtils.gridEventName(firstDiscoEvt.type()) : -1) +
+            ", evtNode=" + (firstDiscoEvt != null ? firstDiscoEvt.eventNode() : null) +
             ", done=" + isDone() + ']';
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/InitNewCoordinatorFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/InitNewCoordinatorFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/InitNewCoordinatorFuture.java
index 71db8ad..42ce9b9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/InitNewCoordinatorFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/InitNewCoordinatorFuture.java
@@ -102,7 +102,7 @@ public class InitNewCoordinatorFuture extends GridCompoundFuture {
         if (restoreState) {
             DiscoCache curDiscoCache = cctx.discovery().discoCache();
 
-            DiscoCache discoCache = exchFut.discoCache();
+            DiscoCache discoCache = exchFut.events().discoveryCache();
 
             List<ClusterNode> nodes = new ArrayList<>();
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
index 7713caa5..0178bb6 100755
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/GridCacheDatabaseSharedManager.java
@@ -721,7 +721,7 @@ public class GridCacheDatabaseSharedManager extends IgniteCacheDatabaseSharedMan
 
     /** {@inheritDoc} */
     @Override public void beforeExchange(GridDhtPartitionsExchangeFuture fut) throws IgniteCheckedException {
-        DiscoveryEvent discoEvt = fut.discoveryEvent();
+        DiscoveryEvent discoEvt = fut.firstEvent();
 
         boolean joinEvt = discoEvt.type() == EventType.EVT_NODE_JOINED;
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/e3ecc169/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheLateAffinityAssignmentTest.java
----------------------------------------------------------------------
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 0d106a4..695d8a6 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
@@ -2809,7 +2809,7 @@ public class CacheLateAffinityAssignmentTest extends GridCommonAbstractTest {
                 GridDhtPartitionsExchangeFuture fut = futs.get(i);
 
                 if (fut.initialVersion().equals(topVer0)) {
-                    evt = fut.discoveryEvent();
+                    evt = fut.firstEvent();
 
                     break;
                 }


Mime
View raw message