ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From a.@apache.org
Subject [2/3] ignite git commit: IGNITE-4779 Missed discovery data snapshot during exchange processing (Backport from 2.0)
Date Mon, 03 Apr 2017 15:31:23 GMT
IGNITE-4779 Missed discovery data snapshot during exchange processing (Backport from 2.0)


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

Branch: refs/heads/master
Commit: 8273e6703944ea50b229a512398ac741eb713073
Parents: 2263283
Author: Anton Vinogradov <av@apache.org>
Authored: Tue Mar 21 18:04:01 2017 +0300
Committer: Anton Vinogradov <av@apache.org>
Committed: Tue Mar 21 18:04:01 2017 +0300

----------------------------------------------------------------------
 .../internal/managers/discovery/DiscoCache.java | 310 ++++++++
 .../discovery/GridDiscoveryManager.java         | 710 ++++++-------------
 .../eventstorage/DiscoveryEventListener.java    |  33 +
 .../eventstorage/GridEventStorageManager.java   | 162 ++++-
 .../affinity/GridAffinityAssignmentCache.java   |   7 +-
 .../cache/CacheAffinitySharedManager.java       |  35 +-
 .../cache/GridCacheAffinityManager.java         |   2 +-
 .../GridCachePartitionExchangeManager.java      |  64 +-
 .../dht/GridClientPartitionTopology.java        |  20 +-
 .../dht/GridDhtAssignmentFetchFuture.java       |   7 +-
 .../dht/GridDhtPartitionTopologyImpl.java       |  44 +-
 .../dht/atomic/GridDhtAtomicCache.java          |   4 +-
 .../GridDhtPartitionsExchangeFuture.java        |  33 +-
 .../service/GridServiceProcessor.java           |  21 +-
 .../GridDiscoveryManagerAliveCacheSelfTest.java |  64 +-
 .../GridDiscoveryManagerAttributesSelfTest.java |  14 +-
 .../discovery/GridDiscoveryManagerSelfTest.java | 214 ------
 .../IgniteTopologyPrintFormatSelfTest.java      |   8 +-
 .../testsuites/IgniteKernalSelfTestSuite.java   |   3 -
 19 files changed, 854 insertions(+), 901 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/8273e670/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
new file mode 100644
index 0000000..5247ac1
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoCache.java
@@ -0,0 +1,310 @@
+/*
+ * 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.managers.discovery;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.util.GridConcurrentHashSet;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.P1;
+import org.apache.ignite.internal.util.typedef.internal.CU;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ *
+ */
+public class DiscoCache {
+    /** Local node. */
+    private final ClusterNode loc;
+
+    /** Remote nodes. */
+    private final List<ClusterNode> rmtNodes;
+
+    /** All nodes. */
+    private final List<ClusterNode> allNodes;
+
+    /** All server nodes. */
+    private final List<ClusterNode> srvNodes;
+
+    /** Daemon nodes. */
+    private final List<ClusterNode> daemonNodes;
+
+    /** All server nodes. */
+    private final List<ClusterNode> srvNodesWithCaches;
+
+    /** All nodes with at least one cache configured. */
+    @GridToStringInclude
+    private final List<ClusterNode> allNodesWithCaches;
+
+    /** All remote nodes with at least one cache configured. */
+    @GridToStringInclude
+    private final List<ClusterNode> rmtNodesWithCaches;
+
+    /** Cache nodes by cache name. */
+    @GridToStringInclude
+    private final Map<Integer, List<ClusterNode>> allCacheNodes;
+
+    /** Affinity cache nodes by cache name. */
+    @GridToStringInclude
+    private final Map<Integer, List<ClusterNode>> affCacheNodes;
+
+    /** Node map. */
+    private final Map<UUID, ClusterNode> nodeMap;
+
+    /** Caches where at least one node has near cache enabled. */
+    @GridToStringInclude
+    private final Set<Integer> nearEnabledCaches;
+
+    /** Alive nodes. */
+    private final Set<UUID> alives = new GridConcurrentHashSet<>();
+
+    /**
+     * @param loc Local node.
+     * @param rmtNodes Remote nodes.
+     * @param allNodes All nodes.
+     * @param srvNodes Server nodes.
+     * @param daemonNodes Daemon nodes.
+     * @param srvNodesWithCaches Server nodes with at least one cache configured.
+     * @param allNodesWithCaches All nodes with at least one cache configured.
+     * @param rmtNodesWithCaches Remote nodes with at least one cache configured.
+     * @param allCacheNodes Cache nodes by cache name.
+     * @param affCacheNodes Affinity cache nodes by cache name.
+     * @param nodeMap Node map.
+     * @param nearEnabledCaches Caches where at least one node has near cache enabled.
+     * @param alives Alive nodes.
+     */
+    DiscoCache(ClusterNode loc,
+        List<ClusterNode> rmtNodes,
+        List<ClusterNode> allNodes,
+        List<ClusterNode> srvNodes,
+        List<ClusterNode> daemonNodes,
+        List<ClusterNode> srvNodesWithCaches,
+        List<ClusterNode> allNodesWithCaches,
+        List<ClusterNode> rmtNodesWithCaches,
+        Map<Integer, List<ClusterNode>> allCacheNodes,
+        Map<Integer, List<ClusterNode>> affCacheNodes,
+        Map<UUID, ClusterNode> nodeMap,
+        Set<Integer> nearEnabledCaches,
+        Set<UUID> alives) {
+        this.loc = loc;
+        this.rmtNodes = rmtNodes;
+        this.allNodes = allNodes;
+        this.srvNodes = srvNodes;
+        this.daemonNodes = daemonNodes;
+        this.srvNodesWithCaches = srvNodesWithCaches;
+        this.allNodesWithCaches = allNodesWithCaches;
+        this.rmtNodesWithCaches = rmtNodesWithCaches;
+        this.allCacheNodes = allCacheNodes;
+        this.affCacheNodes = affCacheNodes;
+        this.nodeMap = nodeMap;
+        this.nearEnabledCaches = nearEnabledCaches;
+        this.alives.addAll(alives);
+    }
+
+    /** @return Local node. */
+    public ClusterNode localNode() {
+        return loc;
+    }
+
+    /** @return Remote nodes. */
+    public List<ClusterNode> remoteNodes() {
+        return rmtNodes;
+    }
+
+    /** @return All nodes. */
+    public List<ClusterNode> allNodes() {
+        return allNodes;
+    }
+
+    /** @return Server nodes. */
+    public List<ClusterNode> serverNodes() {
+        return srvNodes;
+    }
+
+    /** @return Daemon nodes. */
+    public List<ClusterNode> daemonNodes() {
+        return daemonNodes;
+    }
+
+    /** @return Server nodes with at least one cache configured. */
+    public List<ClusterNode> serverNodesWithCaches() {
+        return srvNodesWithCaches;
+    }
+
+    /**
+     * Gets all remote nodes that have at least one cache configured.
+     *
+     * @return Collection of nodes.
+     */
+    public List<ClusterNode> remoteNodesWithCaches() {
+        return rmtNodesWithCaches;
+    }
+
+    /**
+     * Gets collection of nodes with at least one cache configured.
+     *
+     * @return Collection of nodes.
+     */
+    public List<ClusterNode> allNodesWithCaches() {
+        return allNodesWithCaches;
+    }
+
+    /**
+     * Gets collection of server nodes with at least one cache configured.
+     *
+     * @return Collection of nodes.
+     */
+    public Collection<ClusterNode> aliveServerNodes() {
+        return F.view(serverNodes(), new P1<ClusterNode>() {
+            @Override public boolean apply(ClusterNode node) {
+                return alives.contains(node.id());
+            }
+        });
+    }
+
+    /**
+     * Gets collection of server nodes with at least one cache configured.
+     *
+     * @return Collection of nodes.
+     */
+    public Collection<ClusterNode> aliveServerNodesWithCaches() {
+        return F.view(serverNodesWithCaches(), new P1<ClusterNode>() {
+            @Override public boolean apply(ClusterNode node) {
+                return alives.contains(node.id());
+            }
+        });
+    }
+
+    /**
+     * @return Oldest alive server node.
+     */
+    public @Nullable ClusterNode oldestAliveServerNode(){
+        Iterator<ClusterNode> it = aliveServerNodes().iterator();
+        return it.hasNext() ? it.next() : null;
+    }
+
+    /**
+     * @return Oldest alive server node with at least one cache configured.
+     */
+    public @Nullable ClusterNode oldestAliveServerNodeWithCache(){
+        Iterator<ClusterNode> it = aliveServerNodesWithCaches().iterator();
+        return it.hasNext() ? it.next() : null;
+    }
+
+    /**
+     * Gets all nodes that have cache with given name.
+     *
+     * @param cacheName Cache name.
+     * @return Collection of nodes.
+     */
+    public List<ClusterNode> cacheNodes(@Nullable String cacheName) {
+        return cacheNodes(CU.cacheId(cacheName));
+    }
+
+    /**
+     * Gets all nodes that have cache with given ID.
+     *
+     * @param cacheId Cache ID.
+     * @return Collection of nodes.
+     */
+    public List<ClusterNode> cacheNodes(Integer cacheId) {
+        return emptyIfNull(allCacheNodes.get(cacheId));
+    }
+
+    /**
+     * Gets all nodes that have cache with given ID and should participate in affinity calculation. With
+     * partitioned cache nodes with near-only cache do not participate in affinity node calculation.
+     *
+     * @param cacheName Cache name.
+     * @return Collection of nodes.
+     */
+    public List<ClusterNode> cacheAffinityNodes(@Nullable String cacheName) {
+        return cacheAffinityNodes(CU.cacheId(cacheName));
+    }
+
+    /**
+     * Gets all nodes that have cache with given ID and should participate in affinity calculation. With
+     * partitioned cache nodes with near-only cache do not participate in affinity node calculation.
+     *
+     * @param cacheId Cache ID.
+     * @return Collection of nodes.
+     */
+    public List<ClusterNode> cacheAffinityNodes(int cacheId) {
+        return emptyIfNull(affCacheNodes.get(cacheId));
+    }
+
+    /**
+     * Checks if cache with given ID has at least one node with near cache enabled.
+     *
+     * @param cacheId Cache ID.
+     * @return {@code True} if cache with given name has at least one node with near cache enabled.
+     */
+    public boolean hasNearCache(int cacheId) {
+        return nearEnabledCaches.contains(cacheId);
+    }
+
+    /**
+     * @param id Node ID.
+     * @return Node.
+     */
+    public @Nullable ClusterNode node(UUID id) {
+        return nodeMap.get(id);
+    }
+
+    /**
+     * Removes left node from alives lists.
+     *
+     * @param rmvd Removed node.
+     */
+    public void updateAlives(ClusterNode rmvd) {
+        alives.remove(rmvd.id());
+    }
+
+    /**
+     * Removes left nodes from cached alives lists.
+     *
+     * @param discovery Discovery manager.
+     */
+    public void updateAlives(GridDiscoveryManager discovery) {
+        for (UUID alive : alives) {
+            if (!discovery.alive(alive))
+                alives.remove(alive);
+        }
+    }
+
+    /**
+     * @param nodes Cluster nodes.
+     * @return Empty collection if nodes list is {@code null}
+     */
+    private List<ClusterNode> emptyIfNull(List<ClusterNode> nodes) {
+        return nodes == null ? Collections.<ClusterNode>emptyList() : nodes;
+    }
+
+    /** {@inheritDoc} */
+    @Override public String toString() {
+        return S.toString(DiscoCache.class, this);
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/8273e670/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
index 9aa4db1..80549dc 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java
@@ -35,14 +35,12 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.NavigableMap;
 import java.util.Set;
-import java.util.TreeMap;
+import java.util.TreeSet;
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ConcurrentSkipListMap;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -74,24 +72,23 @@ import org.apache.ignite.internal.managers.communication.GridIoManager;
 import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
 import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.cache.CacheAffinitySharedManager;
+import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
 import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
 import org.apache.ignite.internal.processors.jobmetrics.GridJobMetrics;
 import org.apache.ignite.internal.processors.security.SecurityContext;
 import org.apache.ignite.internal.processors.service.GridServiceProcessor;
 import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
-import org.apache.ignite.internal.util.F0;
 import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashMap;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.future.GridFinishedFuture;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.lang.GridTuple5;
+import org.apache.ignite.internal.util.lang.GridTuple6;
 import org.apache.ignite.internal.util.tostring.GridToStringExclude;
-import org.apache.ignite.internal.util.tostring.GridToStringInclude;
-import org.apache.ignite.internal.util.typedef.C1;
 import org.apache.ignite.internal.util.typedef.CI1;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.G;
 import org.apache.ignite.internal.util.typedef.P1;
+import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.internal.CU;
 import org.apache.ignite.internal.util.typedef.internal.LT;
 import org.apache.ignite.internal.util.typedef.internal.S;
@@ -99,7 +96,6 @@ import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.internal.util.worker.GridWorker;
 import org.apache.ignite.lang.IgniteFuture;
 import org.apache.ignite.lang.IgnitePredicate;
-import org.apache.ignite.lang.IgniteProductVersion;
 import org.apache.ignite.lang.IgniteUuid;
 import org.apache.ignite.plugin.security.SecurityCredentials;
 import org.apache.ignite.plugin.segmentation.SegmentationPolicy;
@@ -113,6 +109,7 @@ import org.apache.ignite.spi.discovery.DiscoverySpiListener;
 import org.apache.ignite.spi.discovery.DiscoverySpiNodeAuthenticator;
 import org.apache.ignite.spi.discovery.DiscoverySpiOrderSupport;
 import org.apache.ignite.thread.IgniteThread;
+import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.jsr166.ConcurrentHashMap8;
 
@@ -233,7 +230,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
     private long segChkFreq;
 
     /** Local node join to topology event. */
-    private GridFutureAdapter<DiscoveryEvent> locJoinEvt = new GridFutureAdapter<>();
+    private GridFutureAdapter<T2<DiscoveryEvent, DiscoCache>> locJoin = new GridFutureAdapter<>();
 
     /** GC CPU load. */
     private volatile double gcCpuLoad;
@@ -542,20 +539,25 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                     }
                 }
 
+                final DiscoCache discoCache;
+
                 // Put topology snapshot into discovery history.
                 // There is no race possible between history maintenance and concurrent discovery
                 // event notifications, since SPI notifies manager about all events from this listener.
                 if (verChanged) {
-                    DiscoCache cache = new DiscoCache(locNode, F.view(topSnapshot, F.remoteNodes(locNode.id())));
+                    discoCache = createDiscoCache(locNode, topSnapshot);
 
-                    discoCacheHist.put(nextTopVer, cache);
+                    discoCacheHist.put(nextTopVer, discoCache);
 
-                    boolean set = updateTopologyVersionIfGreater(nextTopVer, cache);
+                    boolean set = updateTopologyVersionIfGreater(nextTopVer, discoCache);
 
                     assert set || topVer == 0 : "Topology version has not been updated [this.topVer=" +
                         topSnap + ", topVer=" + topVer + ", node=" + node +
                         ", evt=" + U.gridEventName(type) + ']';
                 }
+                else
+                    // Current version.
+                    discoCache = discoCache();
 
                 // If this is a local join event, just save it and do not notify listeners.
                 if (type == EVT_NODE_JOINED && node.id().equals(locNode.id())) {
@@ -563,7 +565,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                         gridStartTime = getSpi().getGridStartTime();
 
                     updateTopologyVersionIfGreater(new AffinityTopologyVersion(locNode.order()),
-                        new DiscoCache(localNode(), F.view(topSnapshot, F.remoteNodes(locNode.id()))));
+                        discoCache);
 
                     startLatch.countDown();
 
@@ -573,14 +575,9 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                     discoEvt.eventNode(node);
                     discoEvt.type(EVT_NODE_JOINED);
 
-                    discoEvt.topologySnapshot(topVer, new ArrayList<>(
-                        F.viewReadOnly(topSnapshot, new C1<ClusterNode, ClusterNode>() {
-                            @Override public ClusterNode apply(ClusterNode e) {
-                                return e;
-                            }
-                        }, FILTER_DAEMON)));
+                    discoEvt.topologySnapshot(topVer, new ArrayList<>(F.view(topSnapshot, FILTER_DAEMON)));
 
-                    locJoinEvt.onDone(discoEvt);
+                    locJoin.onDone(new T2<>(discoEvt, discoCache));
 
                     return;
                 }
@@ -595,7 +592,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
 
                     ((IgniteKernal)ctx.grid()).onDisconnected();
 
-                    locJoinEvt = new GridFutureAdapter<>();
+                    locJoin = new GridFutureAdapter<>();
 
                     registeredCaches.clear();
 
@@ -608,7 +605,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                     topHist.clear();
 
                     topSnap.set(new Snapshot(AffinityTopologyVersion.ZERO,
-                        new DiscoCache(locNode, Collections.<ClusterNode>emptySet())));
+                        createDiscoCache(locNode, Collections.<ClusterNode>emptySet())));
                 }
                 else if (type == EVT_CLIENT_NODE_RECONNECTED) {
                     assert locNode.isClient() : locNode;
@@ -625,7 +622,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                             try {
                                 fut.get();
 
-                                discoWrk.addEvent(type, nextTopVer, node, topSnapshot, null);
+                                discoWrk.addEvent(type, nextTopVer, node, discoCache, topSnapshot, null);
                             }
                             catch (IgniteException ignore) {
                                 // No-op.
@@ -637,7 +634,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                 }
 
                 if (type == EVT_CLIENT_NODE_DISCONNECTED || type == EVT_NODE_SEGMENTED || !ctx.clientDisconnected())
-                    discoWrk.addEvent(type, nextTopVer, node, topSnapshot, customMsg);
+                    discoWrk.addEvent(type, nextTopVer, node, discoCache, topSnapshot, customMsg);
             }
         });
 
@@ -1333,8 +1330,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
             U.join(segChkThread, log);
         }
 
-        if (!locJoinEvt.isDone())
-            locJoinEvt.onDone(
+        if (!locJoin.isDone())
+            locJoin.onDone(
                 new IgniteCheckedException("Failed to wait for local node joined event (grid is stopping)."));
     }
 
@@ -1528,7 +1525,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
      *
      * @return Discovery collection cache.
      */
-    private DiscoCache discoCache() {
+    public DiscoCache discoCache() {
         Snapshot cur = topSnap.get();
 
         assert cur != null;
@@ -1577,7 +1574,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
      * @return Collection of cache nodes.
      */
     public Collection<ClusterNode> nodes(AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(null, topVer).allNodes();
+        return resolveDiscoCache(CU.cacheId(null), topVer).allNodes();
     }
 
     /**
@@ -1585,7 +1582,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
      * @return All server nodes for given topology version.
      */
     public List<ClusterNode> serverNodes(AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(null, topVer).srvNodes;
+        return resolveDiscoCache(CU.cacheId(null), topVer).serverNodes();
     }
 
     /**
@@ -1596,7 +1593,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
      * @return Node.
      */
     public ClusterNode node(AffinityTopologyVersion topVer, UUID id) {
-        return resolveDiscoCache(null, topVer).node(id);
+        return resolveDiscoCache(CU.cacheId(null), topVer).node(id);
     }
 
     /**
@@ -1607,49 +1604,38 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
      * @return Collection of cache nodes.
      */
     public Collection<ClusterNode> cacheNodes(@Nullable String cacheName, AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(cacheName, topVer).cacheNodes(cacheName, topVer.topologyVersion());
+        return resolveDiscoCache(CU.cacheId(cacheName), topVer).cacheNodes(cacheName);
     }
 
     /**
-     * Gets all nodes with at least one cache configured.
+     * Gets cache nodes for cache with given ID.
      *
+     * @param cacheId Cache ID.
      * @param topVer Topology version.
      * @return Collection of cache nodes.
      */
-    public Collection<ClusterNode> cacheNodes(AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(null, topVer).allNodesWithCaches(topVer.topologyVersion());
+    public Collection<ClusterNode> cacheNodes(int cacheId, AffinityTopologyVersion topVer) {
+        return resolveDiscoCache(cacheId, topVer).cacheNodes(cacheId);
     }
 
     /**
-     * Gets cache remote nodes for cache with given name.
-     *
-     * @param topVer Topology version.
-     * @return Collection of cache nodes.
-     */
-    public Collection<ClusterNode> remoteCacheNodes(AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(null, topVer).remoteCacheNodes(topVer.topologyVersion());
-    }
-
-    /**
-     * Gets cache nodes for cache with given name.
+     * Gets all nodes with at least one cache configured.
      *
-     * @param cacheName Cache name.
      * @param topVer Topology version.
      * @return Collection of cache nodes.
      */
-    Collection<ClusterNode> aliveCacheNodes(@Nullable String cacheName, AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(cacheName, topVer).aliveCacheNodes(cacheName, topVer.topologyVersion());
+    public Collection<ClusterNode> cacheNodes(AffinityTopologyVersion topVer) {
+        return resolveDiscoCache(CU.cacheId(null), topVer).allNodesWithCaches();
     }
 
     /**
      * Gets cache remote nodes for cache with given name.
      *
-     * @param cacheName Cache name.
      * @param topVer Topology version.
      * @return Collection of cache nodes.
      */
-    Collection<ClusterNode> aliveRemoteCacheNodes(@Nullable String cacheName, AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(cacheName, topVer).aliveRemoteCacheNodes(cacheName, topVer.topologyVersion());
+    public Collection<ClusterNode> remoteCacheNodes(AffinityTopologyVersion topVer) {
+        return resolveDiscoCache(CU.cacheId(null), topVer).remoteNodesWithCaches();
     }
 
     /**
@@ -1657,11 +1643,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
      * @return Oldest alive server nodes with at least one cache configured.
      */
     @Nullable public ClusterNode oldestAliveCacheServerNode(AffinityTopologyVersion topVer) {
-        DiscoCache cache = resolveDiscoCache(null, topVer);
-
-        Map.Entry<ClusterNode, Boolean> e = cache.aliveSrvNodesWithCaches.firstEntry();
-
-        return e != null ? e.getKey() : null;
+        return resolveDiscoCache(CU.cacheId(null), topVer).oldestAliveServerNodeWithCache();
     }
 
     /**
@@ -1672,7 +1654,20 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
      * @return Collection of cache affinity nodes.
      */
     public Collection<ClusterNode> cacheAffinityNodes(@Nullable String cacheName, AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(cacheName, topVer).cacheAffinityNodes(cacheName, topVer.topologyVersion());
+        int cacheId = CU.cacheId(cacheName);
+
+        return resolveDiscoCache(cacheId, topVer).cacheAffinityNodes(cacheId);
+    }
+
+    /**
+     * Gets cache nodes for cache with given ID that participate in affinity calculation.
+     *
+     * @param cacheId Cache ID.
+     * @param topVer Topology version.
+     * @return Collection of cache affinity nodes.
+     */
+    public Collection<ClusterNode> cacheAffinityNodes(int cacheId, AffinityTopologyVersion topVer) {
+        return resolveDiscoCache(cacheId, topVer).cacheAffinityNodes(cacheId);
     }
 
     /**
@@ -1742,31 +1737,34 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
     }
 
     /**
-     * Checks if cache with given name has at least one node with near cache enabled.
+     * Checks if cache with given ID has at least one node with near cache enabled.
      *
-     * @param cacheName Cache name.
+     * @param cacheId Cache ID.
      * @param topVer Topology version.
      * @return {@code True} if cache with given name has at least one node with near cache enabled.
      */
-    public boolean hasNearCache(@Nullable String cacheName, AffinityTopologyVersion topVer) {
-        return resolveDiscoCache(cacheName, topVer).hasNearCache(cacheName);
+    public boolean hasNearCache(int cacheId, AffinityTopologyVersion topVer) {
+        return resolveDiscoCache(cacheId, topVer).hasNearCache(cacheId);
     }
 
     /**
      * Gets discovery cache for given topology version.
      *
-     * @param cacheName Cache name (participates in exception message).
+     * @param cacheId Cache ID (participates in exception message).
      * @param topVer Topology version.
      * @return Discovery cache.
      */
-    private DiscoCache resolveDiscoCache(@Nullable String cacheName, AffinityTopologyVersion topVer) {
+    private DiscoCache resolveDiscoCache(int cacheId, AffinityTopologyVersion topVer) {
         Snapshot snap = topSnap.get();
 
         DiscoCache cache = AffinityTopologyVersion.NONE.equals(topVer) || topVer.equals(snap.topVer) ?
             snap.discoCache : discoCacheHist.get(topVer);
 
         if (cache == null) {
-            throw new IgniteException("Failed to resolve nodes topology [cacheName=" + cacheName +
+            DynamicCacheDescriptor desc = ctx.cache().cacheDescriptor(cacheId);
+
+            throw new IgniteException("Failed to resolve nodes topology [" +
+                "cacheName=" + (desc != null ? desc.cacheConfiguration().getName() : "N/A") +
                 ", topVer=" + topVer +
                 ", history=" + discoCacheHist.keySet() +
                 ", snap=" + snap +
@@ -1817,7 +1815,19 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
     /** @return Event that represents a local node joined to topology. */
     public DiscoveryEvent localJoinEvent() {
         try {
-            return locJoinEvt.get();
+            return locJoin.get().get1();
+        }
+        catch (IgniteCheckedException e) {
+            throw new IgniteException(e);
+        }
+    }
+
+    /**
+     * @return Tuple that consists of a local join event and discovery cache at the join time.
+     */
+    public T2<DiscoveryEvent, DiscoCache> localJoin() {
+        try {
+            return locJoin.get();
         }
         catch (IgniteCheckedException e) {
             throw new IgniteException(e);
@@ -1891,6 +1901,114 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
     }
 
     /**
+     * @param loc Local node.
+     * @param topSnapshot Topology snapshot.
+     * @return Newly created discovery cache.
+     */
+    @NotNull private DiscoCache createDiscoCache(ClusterNode loc, Collection<ClusterNode> topSnapshot) {
+        HashSet<UUID> alives = U.newHashSet(topSnapshot.size());
+        HashMap<UUID, ClusterNode> nodeMap = U.newHashMap(topSnapshot.size());
+
+        ArrayList<ClusterNode> daemonNodes = new ArrayList<>(topSnapshot.size());
+        ArrayList<ClusterNode> srvNodes = new ArrayList<>(topSnapshot.size());
+        ArrayList<ClusterNode> rmtNodes = new ArrayList<>(topSnapshot.size());
+        ArrayList<ClusterNode> allNodes = new ArrayList<>(topSnapshot.size());
+
+        for (ClusterNode node : topSnapshot) {
+            if (alive(node))
+                alives.add(node.id());
+
+            if (node.isDaemon())
+                daemonNodes.add(node);
+            else {
+                allNodes.add(node);
+
+                if (!node.isLocal())
+                    rmtNodes.add(node);
+
+                if (!CU.clientNode(node))
+                    srvNodes.add(node);
+            }
+
+            nodeMap.put(node.id(), node);
+        }
+
+        assert !rmtNodes.contains(loc) : "Remote nodes collection shouldn't contain local node" +
+            " [rmtNodes=" + rmtNodes + ", loc=" + loc + ']';
+
+        Map<Integer, List<ClusterNode>> allCacheNodes = U.newHashMap(allNodes.size());
+        Map<Integer, List<ClusterNode>> affCacheNodes = U.newHashMap(allNodes.size());
+
+        Set<ClusterNode> allNodesWithCaches = new TreeSet<>(GridNodeOrderComparator.INSTANCE);
+        Set<ClusterNode> rmtNodesWithCaches = new TreeSet<>(GridNodeOrderComparator.INSTANCE);
+        Set<ClusterNode> srvNodesWithCaches = new TreeSet<>(GridNodeOrderComparator.INSTANCE);
+
+        Set<Integer> nearEnabledCaches = new HashSet<>();
+
+        for (ClusterNode node : allNodes) {
+            assert node.order() != 0 : "Invalid node order [locNode=" + loc + ", node=" + node + ']';
+            assert !node.isDaemon();
+
+            for (Map.Entry<String, CachePredicate> entry : registeredCaches.entrySet()) {
+                String cacheName = entry.getKey();
+                CachePredicate filter = entry.getValue();
+
+                if (filter.cacheNode(node)) {
+                    allNodesWithCaches.add(node);
+
+                    if(!CU.clientNode(node))
+                        srvNodesWithCaches.add(node);
+
+                    if (!node.isLocal())
+                        rmtNodesWithCaches.add(node);
+
+                    addToMap(allCacheNodes, cacheName, node);
+
+                    if (filter.dataNode(node))
+                        addToMap(affCacheNodes, cacheName, node);
+
+                    if (filter.nearNode(node))
+                        nearEnabledCaches.add(CU.cacheId(cacheName));
+                }
+            }
+        }
+
+        return new DiscoCache(
+            loc,
+            Collections.unmodifiableList(rmtNodes),
+            Collections.unmodifiableList(allNodes),
+            Collections.unmodifiableList(srvNodes),
+            Collections.unmodifiableList(daemonNodes),
+            U.sealList(srvNodesWithCaches),
+            U.sealList(allNodesWithCaches),
+            U.sealList(rmtNodesWithCaches),
+            Collections.unmodifiableMap(allCacheNodes),
+            Collections.unmodifiableMap(affCacheNodes),
+            Collections.unmodifiableMap(nodeMap),
+            Collections.unmodifiableSet(nearEnabledCaches),
+            alives);
+    }
+
+    /**
+     * Adds node to map.
+     *
+     * @param cacheMap Map to add to.
+     * @param cacheName Cache name.
+     * @param rich Node to add
+     */
+    private void addToMap(Map<Integer, List<ClusterNode>> cacheMap, String cacheName, ClusterNode rich) {
+        List<ClusterNode> cacheNodes = cacheMap.get(CU.cacheId(cacheName));
+
+        if (cacheNodes == null) {
+            cacheNodes = new ArrayList<>();
+
+            cacheMap.put(CU.cacheId(cacheName), cacheNodes);
+        }
+
+        cacheNodes.add(rich);
+    }
+
+    /**
      * Updates topology version if current version is smaller than updated.
      *
      * @param updated Updated topology version.
@@ -1991,8 +2109,16 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                     lastChk = now;
 
                     if (!segValid) {
-                        discoWrk.addEvent(EVT_NODE_SEGMENTED, AffinityTopologyVersion.NONE, getSpi().getLocalNode(),
-                            Collections.<ClusterNode>emptyList(), null);
+                        List<ClusterNode> empty = Collections.emptyList();
+
+                        ClusterNode node = getSpi().getLocalNode();
+
+                        discoWrk.addEvent(EVT_NODE_SEGMENTED,
+                            AffinityTopologyVersion.NONE,
+                            node,
+                            createDiscoCache(node, empty),
+                            empty,
+                            null);
 
                         lastSegChkRes.set(false);
                     }
@@ -2012,8 +2138,8 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
     /** Worker for discovery events. */
     private class DiscoveryWorker extends GridWorker {
         /** Event queue. */
-        private final BlockingQueue<GridTuple5<Integer, AffinityTopologyVersion, ClusterNode, Collection<ClusterNode>,
-            DiscoveryCustomMessage>> evts = new LinkedBlockingQueue<>();
+        private final BlockingQueue<GridTuple6<Integer, AffinityTopologyVersion, ClusterNode,
+            DiscoCache, Collection<ClusterNode>, DiscoveryCustomMessage>> evts = new LinkedBlockingQueue<>();
 
         /** Node segmented event fired flag. */
         private boolean nodeSegFired;
@@ -2031,10 +2157,11 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
          * @param type Discovery event type. See {@link DiscoveryEvent} for more details.
          * @param topVer Topology version.
          * @param node Remote node this event is connected with.
+         * @param discoCache Discovery cache.
          * @param topSnapshot Topology snapshot.
          */
         @SuppressWarnings("RedundantTypeArguments")
-        private void recordEvent(int type, long topVer, ClusterNode node, Collection<ClusterNode> topSnapshot) {
+        private void recordEvent(int type, long topVer, ClusterNode node, DiscoCache discoCache, Collection<ClusterNode> topSnapshot) {
             assert node != null;
 
             if (ctx.event().isRecordable(type)) {
@@ -2043,7 +2170,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                 evt.node(ctx.discovery().localNode());
                 evt.eventNode(node);
                 evt.type(type);
-
                 evt.topologySnapshot(topVer, U.<ClusterNode, ClusterNode>arrayList(topSnapshot, FILTER_DAEMON));
 
                 if (type == EVT_NODE_METRICS_UPDATED)
@@ -2070,7 +2196,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                 else
                     assert false;
 
-                ctx.event().record(evt);
+                ctx.event().record(evt, discoCache);
             }
         }
 
@@ -2078,6 +2204,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
          * @param type Event type.
          * @param topVer Topology version.
          * @param node Node.
+         * @param discoCache Discovery cache.
          * @param topSnapshot Topology snapshot.
          * @param data Custom message.
          */
@@ -2085,12 +2212,13 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
             int type,
             AffinityTopologyVersion topVer,
             ClusterNode node,
+            DiscoCache discoCache,
             Collection<ClusterNode> topSnapshot,
             @Nullable DiscoveryCustomMessage data
         ) {
             assert node != null : data;
 
-            evts.add(new GridTuple5<>(type, topVer, node, topSnapshot, data));
+            evts.add(new GridTuple6<>(type, topVer, node, discoCache, topSnapshot, data));
         }
 
         /**
@@ -2127,7 +2255,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
         /** @throws InterruptedException If interrupted. */
         @SuppressWarnings("DuplicateCondition")
         private void body0() throws InterruptedException {
-            GridTuple5<Integer, AffinityTopologyVersion, ClusterNode, Collection<ClusterNode>,
+            GridTuple6<Integer, AffinityTopologyVersion, ClusterNode, DiscoCache, Collection<ClusterNode>,
                 DiscoveryCustomMessage> evt = evts.take();
 
             int type = evt.get1();
@@ -2260,11 +2388,11 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                         customEvt.node(ctx.discovery().localNode());
                         customEvt.eventNode(node);
                         customEvt.type(type);
-                        customEvt.topologySnapshot(topVer.topologyVersion(), evt.get4());
+                        customEvt.topologySnapshot(topVer.topologyVersion(), evt.get5());
                         customEvt.affinityTopologyVersion(topVer);
-                        customEvt.customMessage(evt.get5());
+                        customEvt.customMessage(evt.get6());
 
-                        ctx.event().record(customEvt);
+                        ctx.event().record(customEvt, evt.get4());
                     }
 
                     return;
@@ -2278,7 +2406,7 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
                     assert false : "Invalid discovery event: " + type;
             }
 
-            recordEvent(type, topVer.topologyVersion(), node, evt.get4());
+            recordEvent(type, topVer.topologyVersion(), node, evt.get4(), evt.get5());
 
             if (segmented)
                 onSegmentation();
@@ -2488,432 +2616,6 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
         }
     }
 
-    /** Cache for discovery collections. */
-    private class DiscoCache {
-        /** Remote nodes. */
-        private final List<ClusterNode> rmtNodes;
-
-        /** All nodes. */
-        private final List<ClusterNode> allNodes;
-
-        /** All server nodes. */
-        private final List<ClusterNode> srvNodes;
-
-        /** All nodes with at least one cache configured. */
-        @GridToStringInclude
-        private final Collection<ClusterNode> allNodesWithCaches;
-
-        /** All nodes with at least one cache configured. */
-        @GridToStringInclude
-        private final Collection<ClusterNode> rmtNodesWithCaches;
-
-        /** Cache nodes by cache name. */
-        @GridToStringInclude
-        private final Map<String, Collection<ClusterNode>> allCacheNodes;
-
-        /** Remote cache nodes by cache name. */
-        @GridToStringInclude
-        private final Map<String, Collection<ClusterNode>> rmtCacheNodes;
-
-        /** Cache nodes by cache name. */
-        @GridToStringInclude
-        private final Map<String, Collection<ClusterNode>> affCacheNodes;
-
-        /** Caches where at least one node has near cache enabled. */
-        @GridToStringInclude
-        private final Set<String> nearEnabledCaches;
-
-        /** Nodes grouped by version. */
-        private final NavigableMap<IgniteProductVersion, Collection<ClusterNode>> nodesByVer;
-
-        /** Daemon nodes. */
-        private final List<ClusterNode> daemonNodes;
-
-        /** Node map. */
-        private final Map<UUID, ClusterNode> nodeMap;
-
-        /** Local node. */
-        private final ClusterNode loc;
-
-        /** Highest node order. */
-        private final long maxOrder;
-
-        /**
-         * Cached alive nodes list. As long as this collection doesn't accept {@code null}s use {@link
-         * #maskNull(String)} before passing raw cache names to it.
-         */
-        private final ConcurrentMap<String, Collection<ClusterNode>> aliveCacheNodes;
-
-        /**
-         * Cached alive remote nodes list. As long as this collection doesn't accept {@code null}s use {@link
-         * #maskNull(String)} before passing raw cache names to it.
-         */
-        private final ConcurrentMap<String, Collection<ClusterNode>> aliveRmtCacheNodes;
-
-        /**
-         * Cached alive server remote nodes with caches.
-         */
-        private final ConcurrentSkipListMap<ClusterNode, Boolean> aliveSrvNodesWithCaches;
-
-        /**
-         * @param loc Local node.
-         * @param rmts Remote nodes.
-         */
-        private DiscoCache(ClusterNode loc, Collection<ClusterNode> rmts) {
-            this.loc = loc;
-
-            rmtNodes = Collections.unmodifiableList(new ArrayList<>(F.view(rmts, FILTER_DAEMON)));
-
-            assert !rmtNodes.contains(loc) : "Remote nodes collection shouldn't contain local node" +
-                " [rmtNodes=" + rmtNodes + ", loc=" + loc + ']';
-
-            List<ClusterNode> all = new ArrayList<>(rmtNodes.size() + 1);
-
-            if (!loc.isDaemon())
-                all.add(loc);
-
-            all.addAll(rmtNodes);
-
-            Collections.sort(all, GridNodeOrderComparator.INSTANCE);
-
-            allNodes = Collections.unmodifiableList(all);
-
-            Map<String, Collection<ClusterNode>> cacheMap = new HashMap<>(allNodes.size(), 1.0f);
-            Map<String, Collection<ClusterNode>> rmtCacheMap = new HashMap<>(allNodes.size(), 1.0f);
-            Map<String, Collection<ClusterNode>> dhtNodesMap = new HashMap<>(allNodes.size(), 1.0f);
-            Collection<ClusterNode> nodesWithCaches = new HashSet<>(allNodes.size());
-            Collection<ClusterNode> rmtNodesWithCaches = new HashSet<>(allNodes.size());
-
-            aliveCacheNodes = new ConcurrentHashMap8<>(allNodes.size(), 1.0f);
-            aliveRmtCacheNodes = new ConcurrentHashMap8<>(allNodes.size(), 1.0f);
-            aliveSrvNodesWithCaches = new ConcurrentSkipListMap<>(GridNodeOrderComparator.INSTANCE);
-            nodesByVer = new TreeMap<>();
-
-            long maxOrder0 = 0;
-
-            Set<String> nearEnabledSet = new HashSet<>();
-
-            List<ClusterNode> srvNodes = new ArrayList<>();
-
-            for (ClusterNode node : allNodes) {
-                assert node.order() != 0 : "Invalid node order [locNode=" + loc + ", node=" + node + ']';
-                assert !node.isDaemon();
-
-                if (!CU.clientNode(node))
-                    srvNodes.add(node);
-
-                if (node.order() > maxOrder0)
-                    maxOrder0 = node.order();
-
-                boolean hasCaches = false;
-
-                for (Map.Entry<String, CachePredicate> entry : registeredCaches.entrySet()) {
-                    String cacheName = entry.getKey();
-
-                    CachePredicate filter = entry.getValue();
-
-                    if (filter.cacheNode(node)) {
-                        nodesWithCaches.add(node);
-
-                        if (!loc.id().equals(node.id()))
-                            rmtNodesWithCaches.add(node);
-
-                        addToMap(cacheMap, cacheName, node);
-
-                        if (alive(node.id()))
-                            addToMap(aliveCacheNodes, maskNull(cacheName), node);
-
-                        if (filter.dataNode(node))
-                            addToMap(dhtNodesMap, cacheName, node);
-
-                        if (filter.nearNode(node))
-                            nearEnabledSet.add(cacheName);
-
-                        if (!loc.id().equals(node.id())) {
-                            addToMap(rmtCacheMap, cacheName, node);
-
-                            if (alive(node.id()))
-                                addToMap(aliveRmtCacheNodes, maskNull(cacheName), node);
-                        }
-
-                        hasCaches = true;
-                    }
-                }
-
-                if (hasCaches && alive(node.id()) && !CU.clientNode(node))
-                    aliveSrvNodesWithCaches.put(node, Boolean.TRUE);
-
-                IgniteProductVersion nodeVer = U.productVersion(node);
-
-                // Create collection for this version if it does not exist.
-                Collection<ClusterNode> nodes = nodesByVer.get(nodeVer);
-
-                if (nodes == null) {
-                    nodes = new ArrayList<>(allNodes.size());
-
-                    nodesByVer.put(nodeVer, nodes);
-                }
-
-                nodes.add(node);
-            }
-
-            Collections.sort(srvNodes, CU.nodeComparator(true));
-
-            // Need second iteration to add this node to all previous node versions.
-            for (ClusterNode node : allNodes) {
-                IgniteProductVersion nodeVer = U.productVersion(node);
-
-                // Get all versions lower or equal node's version.
-                NavigableMap<IgniteProductVersion, Collection<ClusterNode>> updateView =
-                    nodesByVer.headMap(nodeVer, false);
-
-                for (Collection<ClusterNode> prevVersions : updateView.values())
-                    prevVersions.add(node);
-            }
-
-            maxOrder = maxOrder0;
-
-            allCacheNodes = Collections.unmodifiableMap(cacheMap);
-            rmtCacheNodes = Collections.unmodifiableMap(rmtCacheMap);
-            affCacheNodes = Collections.unmodifiableMap(dhtNodesMap);
-            allNodesWithCaches = Collections.unmodifiableCollection(nodesWithCaches);
-            this.rmtNodesWithCaches = Collections.unmodifiableCollection(rmtNodesWithCaches);
-            nearEnabledCaches = Collections.unmodifiableSet(nearEnabledSet);
-            this.srvNodes = Collections.unmodifiableList(srvNodes);
-
-            daemonNodes = Collections.unmodifiableList(new ArrayList<>(
-                F.view(F.concat(false, loc, rmts), F0.not(FILTER_DAEMON))));
-
-            Map<UUID, ClusterNode> nodeMap = new HashMap<>(allNodes().size() + daemonNodes.size(), 1.0f);
-
-            for (ClusterNode n : F.concat(false, allNodes(), daemonNodes()))
-                nodeMap.put(n.id(), n);
-
-            this.nodeMap = nodeMap;
-        }
-
-        /**
-         * Adds node to map.
-         *
-         * @param cacheMap Map to add to.
-         * @param cacheName Cache name.
-         * @param rich Node to add
-         */
-        private void addToMap(Map<String, Collection<ClusterNode>> cacheMap, String cacheName, ClusterNode rich) {
-            Collection<ClusterNode> cacheNodes = cacheMap.get(cacheName);
-
-            if (cacheNodes == null) {
-                cacheNodes = new ArrayList<>(allNodes.size());
-
-                cacheMap.put(cacheName, cacheNodes);
-            }
-
-            cacheNodes.add(rich);
-        }
-
-        /** @return Local node. */
-        ClusterNode localNode() {
-            return loc;
-        }
-
-        /** @return Remote nodes. */
-        Collection<ClusterNode> remoteNodes() {
-            return rmtNodes;
-        }
-
-        /** @return All nodes. */
-        Collection<ClusterNode> allNodes() {
-            return allNodes;
-        }
-
-        /**
-         * Gets collection of nodes which have version equal or greater than {@code ver}.
-         *
-         * @param ver Version to check.
-         * @return Collection of nodes with version equal or greater than {@code ver}.
-         */
-        Collection<ClusterNode> elderNodes(IgniteProductVersion ver) {
-            Map.Entry<IgniteProductVersion, Collection<ClusterNode>> entry = nodesByVer.ceilingEntry(ver);
-
-            if (entry == null)
-                return Collections.emptyList();
-
-            return entry.getValue();
-        }
-
-        /**
-         * @return Versions map.
-         */
-        NavigableMap<IgniteProductVersion, Collection<ClusterNode>> versionsMap() {
-            return nodesByVer;
-        }
-
-        /**
-         * Gets collection of nodes with at least one cache configured.
-         *
-         * @param topVer Topology version (maximum allowed node order).
-         * @return Collection of nodes.
-         */
-        Collection<ClusterNode> allNodesWithCaches(final long topVer) {
-            return filter(topVer, allNodesWithCaches);
-        }
-
-        /**
-         * Gets all nodes that have cache with given name.
-         *
-         * @param cacheName Cache name.
-         * @param topVer Topology version.
-         * @return Collection of nodes.
-         */
-        Collection<ClusterNode> cacheNodes(@Nullable String cacheName, final long topVer) {
-            return filter(topVer, allCacheNodes.get(cacheName));
-        }
-
-        /**
-         * Gets all remote nodes that have at least one cache configured.
-         *
-         * @param topVer Topology version.
-         * @return Collection of nodes.
-         */
-        Collection<ClusterNode> remoteCacheNodes(final long topVer) {
-            return filter(topVer, rmtNodesWithCaches);
-        }
-
-        /**
-         * Gets all nodes that have cache with given name and should participate in affinity calculation. With
-         * partitioned cache nodes with near-only cache do not participate in affinity node calculation.
-         *
-         * @param cacheName Cache name.
-         * @param topVer Topology version.
-         * @return Collection of nodes.
-         */
-        Collection<ClusterNode> cacheAffinityNodes(@Nullable String cacheName, final long topVer) {
-            return filter(topVer, affCacheNodes.get(cacheName));
-        }
-
-        /**
-         * Gets all alive nodes that have cache with given name.
-         *
-         * @param cacheName Cache name.
-         * @param topVer Topology version.
-         * @return Collection of nodes.
-         */
-        Collection<ClusterNode> aliveCacheNodes(@Nullable String cacheName, final long topVer) {
-            return filter(topVer, aliveCacheNodes.get(maskNull(cacheName)));
-        }
-
-        /**
-         * Gets all alive remote nodes that have cache with given name.
-         *
-         * @param cacheName Cache name.
-         * @param topVer Topology version.
-         * @return Collection of nodes.
-         */
-        Collection<ClusterNode> aliveRemoteCacheNodes(@Nullable String cacheName, final long topVer) {
-            return filter(topVer, aliveRmtCacheNodes.get(maskNull(cacheName)));
-        }
-
-        /**
-         * Checks if cache with given name has at least one node with near cache enabled.
-         *
-         * @param cacheName Cache name.
-         * @return {@code True} if cache with given name has at least one node with near cache enabled.
-         */
-        boolean hasNearCache(@Nullable String cacheName) {
-            return nearEnabledCaches.contains(cacheName);
-        }
-
-        /**
-         * Removes left node from cached alives lists.
-         *
-         * @param leftNode Left node.
-         */
-        void updateAlives(ClusterNode leftNode) {
-            if (leftNode.order() > maxOrder)
-                return;
-
-            filterNodeMap(aliveCacheNodes, leftNode);
-
-            filterNodeMap(aliveRmtCacheNodes, leftNode);
-
-            aliveSrvNodesWithCaches.remove(leftNode);
-        }
-
-        /**
-         * Creates a copy of nodes map without the given node.
-         *
-         * @param map Map to copy.
-         * @param exclNode Node to exclude.
-         */
-        private void filterNodeMap(ConcurrentMap<String, Collection<ClusterNode>> map, final ClusterNode exclNode) {
-            for (String cacheName : registeredCaches.keySet()) {
-                String maskedName = maskNull(cacheName);
-
-                while (true) {
-                    Collection<ClusterNode> oldNodes = map.get(maskedName);
-
-                    if (oldNodes == null || oldNodes.isEmpty())
-                        break;
-
-                    Collection<ClusterNode> newNodes = new ArrayList<>(oldNodes);
-
-                    if (!newNodes.remove(exclNode))
-                        break;
-
-                    if (map.replace(maskedName, oldNodes, newNodes))
-                        break;
-                }
-            }
-        }
-
-        /**
-         * Replaces {@code null} with {@code NULL_CACHE_NAME}.
-         *
-         * @param cacheName Cache name.
-         * @return Masked name.
-         */
-        private String maskNull(@Nullable String cacheName) {
-            return cacheName == null ? NULL_CACHE_NAME : cacheName;
-        }
-
-        /**
-         * @param topVer Topology version.
-         * @param nodes Nodes.
-         * @return Filtered collection (potentially empty, but never {@code null}).
-         */
-        private Collection<ClusterNode> filter(final long topVer, @Nullable Collection<ClusterNode> nodes) {
-            if (nodes == null)
-                return Collections.emptyList();
-
-            // If no filtering needed, return original collection.
-            return nodes.isEmpty() || topVer < 0 || topVer >= maxOrder ?
-                nodes :
-                F.view(nodes, new P1<ClusterNode>() {
-                    @Override public boolean apply(ClusterNode node) {
-                        return node.order() <= topVer;
-                    }
-                });
-        }
-
-        /** @return Daemon nodes. */
-        Collection<ClusterNode> daemonNodes() {
-            return daemonNodes;
-        }
-
-        /**
-         * @param id Node ID.
-         * @return Node.
-         */
-        @Nullable ClusterNode node(UUID id) {
-            return nodeMap.get(id);
-        }
-
-        /** {@inheritDoc} */
-        @Override public String toString() {
-            return S.toString(DiscoCache.class, this, "allNodesWithDaemons", U.toShortString(allNodes));
-        }
-    }
-
     /**
      * Cache predicate.
      */

http://git-wip-us.apache.org/repos/asf/ignite/blob/8273e670/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/DiscoveryEventListener.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/DiscoveryEventListener.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/DiscoveryEventListener.java
new file mode 100644
index 0000000..963d97e
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/DiscoveryEventListener.java
@@ -0,0 +1,33 @@
+/*
+ * 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.managers.eventstorage;
+
+import java.util.EventListener;
+import org.apache.ignite.events.DiscoveryEvent;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
+
+/**
+ * Internal listener for discovery events.
+ */
+public interface DiscoveryEventListener extends EventListener {
+    /**
+     * @param evt Discovery event.
+     * @param discoCache Discovery cache.
+     */
+    public void onEvent(DiscoveryEvent evt, DiscoCache discoCache);
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/8273e670/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java
index 607bb96..a2c64ba 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/eventstorage/GridEventStorageManager.java
@@ -47,6 +47,7 @@ import org.apache.ignite.internal.managers.GridManagerAdapter;
 import org.apache.ignite.internal.managers.communication.GridIoManager;
 import org.apache.ignite.internal.managers.communication.GridMessageListener;
 import org.apache.ignite.internal.managers.deployment.GridDeployment;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
 import org.apache.ignite.internal.processors.platform.PlatformEventFilterListener;
 import org.apache.ignite.internal.util.GridConcurrentLinkedHashSet;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
@@ -79,6 +80,9 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
     /** Local event listeners. */
     private final ConcurrentMap<Integer, Set<GridLocalEventListener>> lsnrs = new ConcurrentHashMap8<>();
 
+    /** Internal discovery listeners. */
+    private final ConcurrentMap<Integer, Set<DiscoveryEventListener>> discoLsnrs = new ConcurrentHashMap8<>();
+
     /** Busy lock to control activity of threads. */
     private final ReadWriteLock busyLock = new ReentrantReadWriteLock();
 
@@ -234,6 +238,7 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
         msgLsnr = null;
 
         lsnrs.clear();
+        discoLsnrs.clear();
     }
 
     /** {@inheritDoc} */
@@ -300,6 +305,30 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
     }
 
     /**
+     * Records discovery events.
+     *
+     * @param evt Event to record.
+     * @param discoCache Discovery cache.
+     */
+    public void record(DiscoveryEvent evt, DiscoCache discoCache) {
+        assert evt != null;
+
+        if (!enterBusy())
+            return;
+
+        try {
+            // Notify internal discovery listeners first.
+            notifyDiscoveryListeners(evt, discoCache);
+
+            // Notify all other registered listeners.
+            record(evt);
+        }
+        finally {
+            leaveBusy();
+        }
+    }
+
+    /**
      * Gets types of enabled user-recordable events.
      *
      * @return Array of types of enabled user-recordable events.
@@ -570,7 +599,7 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
 
         try {
             for (int t : types) {
-                getOrCreate(t).add(lsnr);
+                getOrCreate(lsnrs, t).add(lsnr);
 
                 if (!isRecordable(t))
                     U.warn(log, "Added listener for disabled event type: " + U.gridEventName(t));
@@ -595,14 +624,14 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
             return;
 
         try {
-            getOrCreate(type).add(lsnr);
+            getOrCreate(lsnrs, type).add(lsnr);
 
             if (!isRecordable(type))
                 U.warn(log, "Added listener for disabled event type: " + U.gridEventName(type));
 
             if (types != null) {
                 for (int t : types) {
-                    getOrCreate(t).add(lsnr);
+                    getOrCreate(lsnrs, t).add(lsnr);
 
                     if (!isRecordable(t))
                         U.warn(log, "Added listener for disabled event type: " + U.gridEventName(t));
@@ -615,16 +644,70 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
     }
 
     /**
+     * Adds discovery event listener. Note that this method specifically disallow an empty
+     * array of event type to prevent accidental subscription for all system event that
+     * may lead to a drastic performance decrease.
+     *
+     * @param lsnr Listener to add.
+     * @param types Event types to subscribe listener for.
+     */
+    public void addDiscoveryEventListener(DiscoveryEventListener lsnr, int[] types) {
+        assert lsnr != null;
+        assert types != null;
+        assert types.length > 0;
+
+        if (!enterBusy())
+            return;
+
+        try {
+            for (int t : types) {
+                getOrCreate(discoLsnrs, t).add(lsnr);
+            }
+        }
+        finally {
+            leaveBusy();
+        }
+    }
+
+    /**
+     * Adds discovery event listener.
+     *
+     * @param lsnr Listener to add.
+     * @param type Event type to subscribe listener for.
+     * @param types Additional event types to subscribe listener for.
+     */
+    public void addDiscoveryEventListener(DiscoveryEventListener lsnr, int type, @Nullable int... types) {
+        assert lsnr != null;
+
+        if (!enterBusy())
+            return;
+
+        try {
+            getOrCreate(discoLsnrs, type).add(lsnr);
+
+            if (types != null) {
+                for (int t : types) {
+                    getOrCreate(discoLsnrs, t).add(lsnr);
+                }
+            }
+        }
+        finally {
+            leaveBusy();
+        }
+    }
+
+    /**
+     * @param lsnrs Listeners map.
      * @param type Event type.
      * @return Listeners for given event type.
      */
-    private Collection<GridLocalEventListener> getOrCreate(Integer type) {
-        Set<GridLocalEventListener> set = lsnrs.get(type);
+    private <T> Collection<T> getOrCreate(ConcurrentMap<Integer, Set<T>> lsnrs, Integer type) {
+        Set<T> set = lsnrs.get(type);
 
         if (set == null) {
             set = new GridConcurrentLinkedHashSet<>();
 
-            Set<GridLocalEventListener> prev = lsnrs.putIfAbsent(type, set);
+            Set<T> prev = lsnrs.putIfAbsent(type, set);
 
             if (prev != null)
                 set = prev;
@@ -688,6 +771,38 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
     }
 
     /**
+     * Removes listener for specified events, if any. If no event types provided - it
+     * remove the listener for all its registered events.
+     *
+     * @param lsnr Listener.
+     * @param types Event types.
+     * @return Returns {@code true} if removed.
+     */
+    public boolean removeDiscoveryEventListener(DiscoveryEventListener lsnr, @Nullable int... types) {
+        assert lsnr != null;
+
+        boolean found = false;
+
+        if (F.isEmpty(types)) {
+            for (Set<DiscoveryEventListener> set : discoLsnrs.values())
+                if (set.remove(lsnr))
+                    found = true;
+        }
+        else {
+            assert types != null;
+
+            for (int type : types) {
+                Set<DiscoveryEventListener> set = discoLsnrs.get(type);
+
+                if (set != null && set.remove(lsnr))
+                    found = true;
+            }
+        }
+
+        return found;
+    }
+
+    /**
      *
      * @param p Optional predicate.
      * @param types Event types to wait for.
@@ -780,6 +895,41 @@ public class GridEventStorageManager extends GridManagerAdapter<EventStorageSpi>
     }
 
     /**
+     * @param evt Discovery event
+     * @param cache Discovery cache.
+     */
+    private void notifyDiscoveryListeners(DiscoveryEvent evt, DiscoCache cache) {
+        assert evt != null;
+
+        notifyDiscoveryListeners(discoLsnrs.get(evt.type()), evt, cache);
+    }
+
+    /**
+     * @param set Set of listeners.
+     * @param evt Discovery event.
+     * @param cache Discovery cache.
+     */
+    private void notifyDiscoveryListeners(@Nullable Collection<DiscoveryEventListener> set, DiscoveryEvent evt, DiscoCache cache) {
+        assert evt != null;
+
+        if (!F.isEmpty(set)) {
+            assert set != null;
+
+            for (DiscoveryEventListener lsnr : set) {
+                try {
+                    lsnr.onEvent(evt, cache);
+                }
+                catch (Throwable e) {
+                    U.error(log, "Unexpected exception in listener notification for event: " + evt, e);
+
+                    if (e instanceof Error)
+                        throw (Error)e;
+                }
+            }
+        }
+    }
+
+    /**
      * @param p Grid event predicate.
      * @return Collection of grid events.
      */

http://git-wip-us.apache.org/repos/asf/ignite/blob/8273e670/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
index a388c7a..5070462 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/affinity/GridAffinityAssignmentCache.java
@@ -39,6 +39,7 @@ import org.apache.ignite.events.DiscoveryEvent;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.GridNodeOrderComparator;
 import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
 import org.apache.ignite.internal.util.future.GridFutureAdapter;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.internal.CU;
@@ -252,10 +253,12 @@ public class GridAffinityAssignmentCache {
      *
      * @param topVer Topology version to calculate affinity cache for.
      * @param discoEvt Discovery event that caused this topology version change.
+     * @param discoCache Discovery cache.
      * @return Affinity assignments.
      */
     @SuppressWarnings("IfMayBeConditional")
-    public List<List<ClusterNode>> calculate(AffinityTopologyVersion topVer, DiscoveryEvent discoEvt) {
+    public List<List<ClusterNode>> calculate(AffinityTopologyVersion topVer, DiscoveryEvent discoEvt,
+        DiscoCache discoCache) {
         if (log.isDebugEnabled())
             log.debug("Calculating affinity [topVer=" + topVer + ", locNodeId=" + ctx.localNodeId() +
                 ", discoEvt=" + discoEvt + ']');
@@ -266,7 +269,7 @@ public class GridAffinityAssignmentCache {
         List<ClusterNode> sorted;
 
         if (!locCache) {
-            sorted = new ArrayList<>(ctx.discovery().cacheAffinityNodes(cacheName, topVer));
+            sorted = new ArrayList<>(discoCache.cacheAffinityNodes(cacheId()));
 
             Collections.sort(sorted, GridNodeOrderComparator.INSTANCE);
         }

http://git-wip-us.apache.org/repos/asf/ignite/blob/8273e670/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 2890887..2642d16 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
@@ -382,7 +382,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                 cctx.cache().prepareCacheStart(req, fut.topologyVersion());
 
                 if (fut.isCacheAdded(cacheId, fut.topologyVersion())) {
-                    if (cctx.discovery().cacheAffinityNodes(req.cacheName(), fut.topologyVersion()).isEmpty())
+                    if (fut.discoCache().cacheAffinityNodes(req.cacheName()).isEmpty())
                         U.quietAndWarn(log, "No server nodes found for cache client: " + req.cacheName());
                 }
 
@@ -403,7 +403,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                             assert aff.lastVersion().equals(AffinityTopologyVersion.NONE) : aff.lastVersion();
 
                             List<List<ClusterNode>> assignment = aff.calculate(fut.topologyVersion(),
-                                fut.discoveryEvent());
+                                fut.discoveryEvent(), fut.discoCache());
 
                             aff.initialize(fut.topologyVersion(), assignment);
                         }
@@ -753,7 +753,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
             assert old == null : old;
 
-            List<List<ClusterNode>> newAff = cache.affinity().calculate(fut.topologyVersion(), fut.discoveryEvent());
+            List<List<ClusterNode>> newAff = cache.affinity().calculate(fut.topologyVersion(), fut.discoveryEvent(), fut.discoCache());
 
             cache.affinity().initialize(fut.topologyVersion(), newAff);
         }
@@ -791,7 +791,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
                     if (cache.affinity().lastVersion().equals(AffinityTopologyVersion.NONE)) {
                         List<List<ClusterNode>> assignment =
-                            cache.affinity().calculate(fut.topologyVersion(), fut.discoveryEvent());
+                            cache.affinity().calculate(fut.topologyVersion(), fut.discoveryEvent(), fut.discoCache());
 
                         cache.affinity().initialize(fut.topologyVersion(), assignment);
                     }
@@ -817,14 +817,15 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
     private void initAffinity(GridAffinityAssignmentCache aff, GridDhtPartitionsExchangeFuture fut, boolean fetch)
         throws IgniteCheckedException {
         if (!fetch && canCalculateAffinity(aff, fut)) {
-            List<List<ClusterNode>> assignment = aff.calculate(fut.topologyVersion(), fut.discoveryEvent());
+            List<List<ClusterNode>> assignment = aff.calculate(fut.topologyVersion(), fut.discoveryEvent(), fut.discoCache());
 
             aff.initialize(fut.topologyVersion(), assignment);
         }
         else {
             GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
                 aff.cacheName(),
-                fut.topologyVersion());
+                fut.topologyVersion(),
+                fut.discoCache());
 
             fetchFut.init();
 
@@ -878,7 +879,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
                             CacheHolder cache = cache(fut, cacheDesc);
 
-                            List<List<ClusterNode>> newAff = cache.affinity().calculate(topVer, fut.discoveryEvent());
+                            List<List<ClusterNode>> newAff = cache.affinity().calculate(topVer, fut.discoveryEvent(), fut.discoCache());
 
                             cache.affinity().initialize(topVer, newAff);
                         }
@@ -945,14 +946,15 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
             if (cctx.localNodeId().equals(cacheDesc.receivedFrom())) {
                 List<List<ClusterNode>> assignment =
-                    cacheCtx.affinity().affinityCache().calculate(fut.topologyVersion(), fut.discoveryEvent());
+                    cacheCtx.affinity().affinityCache().calculate(fut.topologyVersion(), fut.discoveryEvent(), fut.discoCache());
 
                 cacheCtx.affinity().affinityCache().initialize(fut.topologyVersion(), assignment);
             }
             else {
                 GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
                     cacheCtx.name(),
-                    topVer);
+                    topVer,
+                    fut.discoCache());
 
                 fetchFut.init();
 
@@ -986,7 +988,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
         GridDhtAffinityAssignmentResponse res = fetchFut.get();
 
         if (res == null) {
-            List<List<ClusterNode>> aff = affCache.calculate(topVer, fut.discoveryEvent());
+            List<List<ClusterNode>> aff = affCache.calculate(topVer, fut.discoveryEvent(), fut.discoCache());
 
             affCache.initialize(topVer, aff);
         }
@@ -998,7 +1000,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
             else {
                 assert !affCache.centralizedAffinityFunction() || !lateAffAssign;
 
-                affCache.calculate(topVer, fut.discoveryEvent());
+                affCache.calculate(topVer, fut.discoveryEvent(), fut.discoCache());
             }
 
             List<List<ClusterNode>> aff = res.affinityAssignment(cctx.discovery());
@@ -1028,7 +1030,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                 if (cacheCtx.isLocal())
                     continue;
 
-                cacheCtx.affinity().affinityCache().calculate(fut.topologyVersion(), fut.discoveryEvent());
+                cacheCtx.affinity().affinityCache().calculate(fut.topologyVersion(), fut.discoveryEvent(), fut.discoCache());
             }
 
             centralizedAff = true;
@@ -1078,7 +1080,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
                 if (cache != null) {
                     if (cache.client())
-                        cache.affinity().calculate(fut.topologyVersion(), fut.discoveryEvent());
+                        cache.affinity().calculate(fut.topologyVersion(), fut.discoveryEvent(), fut.discoCache());
 
                     return;
                 }
@@ -1118,7 +1120,8 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
                     GridDhtAssignmentFetchFuture fetchFut = new GridDhtAssignmentFetchFuture(cctx,
                         aff.cacheName(),
-                        prev.topologyVersion());
+                        prev.topologyVersion(),
+                        prev.discoCache());
 
                     fetchFut.init();
 
@@ -1129,7 +1132,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
                             throws IgniteCheckedException {
                             fetchAffinity(prev, aff, (GridDhtAssignmentFetchFuture)fetchFut);
 
-                            aff.calculate(fut.topologyVersion(), fut.discoveryEvent());
+                            aff.calculate(fut.topologyVersion(), fut.discoveryEvent(), fut.discoCache());
 
                             affFut.onDone(fut.topologyVersion());
                         }
@@ -1269,7 +1272,7 @@ public class CacheAffinitySharedManager<K, V> extends GridCacheSharedManagerAdap
 
         assert aff.idealAssignment() != null : "Previous assignment is not available.";
 
-        List<List<ClusterNode>> idealAssignment = aff.calculate(topVer, fut.discoveryEvent());
+        List<List<ClusterNode>> idealAssignment = aff.calculate(topVer, fut.discoveryEvent(), fut.discoCache());
         List<List<ClusterNode>> newAssignment = null;
 
         if (latePrimary) {

http://git-wip-us.apache.org/repos/asf/ignite/blob/8273e670/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
index 8b7be1b..d9ff616 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityManager.java
@@ -77,7 +77,7 @@ public class GridCacheAffinityManager extends GridCacheManagerAdapter {
     @Override protected void onKernalStart0() throws IgniteCheckedException {
         if (cctx.isLocal())
             // No discovery event needed for local affinity.
-            aff.calculate(LOC_CACHE_TOP_VER, null);
+            aff.calculate(LOC_CACHE_TOP_VER, null, null);
     }
 
     /** {@inheritDoc} */


Mime
View raw message