/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.ignite.internal.processors.cache;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteLogger;
import org.apache.ignite.cache.CacheExistsException;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.internal.GridCachePluginContext;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteNodeAttributes;
import org.apache.ignite.internal.managers.discovery.DiscoCache;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState;
import org.apache.ignite.internal.processors.query.QuerySchema;
import org.apache.ignite.internal.processors.query.QuerySchemaPatch;
import org.apache.ignite.internal.processors.query.QueryUtils;
import org.apache.ignite.internal.processors.query.schema.SchemaOperationException;
import org.apache.ignite.internal.util.GridConcurrentHashSet;
import org.apache.ignite.internal.util.lang.GridFunc;
import org.apache.ignite.internal.util.typedef.F;
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.U;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteProductVersion;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.marshaller.jdk.JdkMarshaller;
import org.apache.ignite.plugin.CachePluginContext;
import org.apache.ignite.plugin.CachePluginProvider;
import org.apache.ignite.plugin.PluginProvider;
import org.apache.ignite.spi.discovery.DiscoveryDataBag;
import org.jetbrains.annotations.Nullable;

import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT;
import static org.apache.ignite.cache.CacheMode.LOCAL;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
import static org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType.CACHE_PROC;

/**
 * Logic related to cache discovery data processing.
 */
class ClusterCachesInfo {
    /** Version since which merge of config is supports. */
    private static final IgniteProductVersion V_MERGE_CONFIG_SINCE = IgniteProductVersion.fromString("2.5.0");

    /** */
    private final GridKernalContext ctx;

    /** Dynamic caches. */
    private final ConcurrentMap<String, DynamicCacheDescriptor> registeredCaches = new ConcurrentHashMap<>();

    /** */
    private final ConcurrentMap<Integer, CacheGroupDescriptor> registeredCacheGrps = new ConcurrentHashMap<>();

    /** Cache templates. */
    private final ConcurrentMap<String, DynamicCacheDescriptor> registeredTemplates = new ConcurrentHashMap<>();

    /** Caches currently being restarted. */
    private final Collection<String> restartingCaches = new GridConcurrentHashSet<>();

    /** */
    private final IgniteLogger log;

    /** */
    private CachesOnDisconnect cachesOnDisconnect;

    /** Local cache info */
    private CacheJoinNodeDiscoveryData joinDiscoData;

    /** Cluster cache info */
    private GridData gridData;

    /** */
    private LocalJoinCachesContext locJoinCachesCtx;

    /** */
    private Map<String, T2<CacheConfiguration, NearCacheConfiguration>> locCfgsForActivation = Collections.emptyMap();

    /** */
    private Map<UUID, CacheClientReconnectDiscoveryData> clientReconnectReqs;

    /** {@code True} if joined cluster while cluster state change was in progress. */
    private boolean joinOnTransition;

    /**
     * @param ctx Context.
     */
    public ClusterCachesInfo(GridKernalContext ctx) {
        this.ctx = ctx;

        log = ctx.log(getClass());
    }

    /**
     * @param joinDiscoData Information about configured caches and templates.
     * @throws IgniteCheckedException If configuration validation failed.
     */
    public void onStart(CacheJoinNodeDiscoveryData joinDiscoData) throws IgniteCheckedException {
        this.joinDiscoData = joinDiscoData;

        Map<String, CacheConfiguration> grpCfgs = new HashMap<>();

        for (CacheJoinNodeDiscoveryData.CacheInfo info : joinDiscoData.caches().values()) {
            if (info.cacheData().config().getGroupName() == null)
                continue;

            CacheConfiguration ccfg = grpCfgs.get(info.cacheData().config().getGroupName());

            if (ccfg == null)
                grpCfgs.put(info.cacheData().config().getGroupName(), info.cacheData().config());
            else
                validateCacheGroupConfiguration(ccfg, info.cacheData().config());
        }

        String conflictErr = processJoiningNode(joinDiscoData, ctx.localNodeId(), true);

        if (conflictErr != null)
            throw new IgniteCheckedException("Failed to start configured cache. " + conflictErr);
    }

    /**
     * @param checkConsistency {@code True} if need check cache configurations consistency.
     * @throws IgniteCheckedException If failed.
     */
    public void onKernalStart(boolean checkConsistency) throws IgniteCheckedException {
        if (gridData != null && gridData.conflictErr != null)
            throw new IgniteCheckedException(gridData.conflictErr);

        if (gridData != null && gridData.joinDiscoData != null) {
            CacheJoinNodeDiscoveryData joinDiscoData = gridData.joinDiscoData;

            for (CacheJoinNodeDiscoveryData.CacheInfo locCacheInfo : joinDiscoData.caches().values()) {
                CacheConfiguration locCfg = locCacheInfo.cacheData().config();

                CacheData cacheData = gridData.gridData.caches().get(locCfg.getName());

                if (cacheData != null) {
                    if (!F.eq(cacheData.sql(), locCacheInfo.sql())) {
                        throw new IgniteCheckedException("Cache configuration mismatch (local cache was created " +
                            "via " + (locCacheInfo.sql() ? "CREATE TABLE" : "Ignite API") + ", while remote cache " +
                            "was created via " + (cacheData.sql() ? "CREATE TABLE" : "Ignite API") + "): " +
                            locCacheInfo.cacheData().config().getName());
                    }

                    if (checkConsistency) {
                        checkCache(locCacheInfo, cacheData, cacheData.receivedFrom());

                        ClusterNode rmt = ctx.discovery().node(cacheData.receivedFrom());

                        if (rmt == null) {
                            for (ClusterNode node : ctx.discovery().localJoin().discoCache().serverNodes()) {
                                if (!node.isLocal() && ctx.discovery().cacheAffinityNode(node, locCfg.getName())) {
                                    rmt = node;

                                    break;
                                }
                            }
                        }

                        if (rmt != null) {
                            for (PluginProvider p : ctx.plugins().allProviders()) {
                                CachePluginContext pluginCtx = new GridCachePluginContext(ctx, locCfg);

                                CachePluginProvider provider = p.createCacheProvider(pluginCtx);

                                if (provider != null)
                                    provider.validateRemote(locCfg, cacheData.cacheConfiguration(), rmt);
                            }
                        }
                    }
                }

                if (checkConsistency)
                    validateStartCacheConfiguration(locCfg);
            }
        }

        gridData = null;
    }

    /**
     * Checks that remote caches has configuration compatible with the local.
     *
     * @param locInfo Local configuration.
     * @param rmtData Remote configuration.
     * @param rmt Remote node.
     * @throws IgniteCheckedException If check failed.
     */
    @SuppressWarnings("unchecked")
    private void checkCache(CacheJoinNodeDiscoveryData.CacheInfo locInfo, CacheData rmtData, UUID rmt)
        throws IgniteCheckedException {
        GridCacheAttributes rmtAttr = new GridCacheAttributes(rmtData.cacheConfiguration());
        GridCacheAttributes locAttr = new GridCacheAttributes(locInfo.cacheData().config());

        CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "cacheMode", "Cache mode",
            locAttr.cacheMode(), rmtAttr.cacheMode(), true);

        CU.checkAttributeMismatch(log, rmtAttr.groupName(), rmt, "groupName", "Cache group name",
            locAttr.groupName(), rmtAttr.groupName(), true);

        if (rmtAttr.cacheMode() != LOCAL) {
            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "interceptor", "Cache Interceptor",
                locAttr.interceptorClassName(), rmtAttr.interceptorClassName(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "atomicityMode",
                "Cache atomicity mode", locAttr.atomicityMode(), rmtAttr.atomicityMode(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "cachePreloadMode",
                "Cache preload mode", locAttr.cacheRebalanceMode(), rmtAttr.cacheRebalanceMode(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "topologyValidator",
                "Cache topology validator", locAttr.topologyValidatorClassName(), rmtAttr.topologyValidatorClassName(), true);

            ClusterNode rmtNode = ctx.discovery().node(rmt);

            if (CU.affinityNode(ctx.discovery().localNode(), locInfo.cacheData().config().getNodeFilter())
                && rmtNode != null && CU.affinityNode(rmtNode, rmtData.cacheConfiguration().getNodeFilter())) {
                CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "storeFactory", "Store factory",
                    locAttr.storeFactoryClassName(), rmtAttr.storeFactoryClassName(), true);
            }

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "cacheAffinity", "Cache affinity",
                locAttr.cacheAffinityClassName(), rmtAttr.cacheAffinityClassName(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "cacheAffinityMapper",
                "Cache affinity mapper", locAttr.cacheAffinityMapperClassName(),
                rmtAttr.cacheAffinityMapperClassName(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "affinityPartitionsCount",
                "Affinity partitions count", locAttr.affinityPartitionsCount(),
                rmtAttr.affinityPartitionsCount(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "evictionFilter", "Eviction filter",
                locAttr.evictionFilterClassName(), rmtAttr.evictionFilterClassName(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "evictionPolicy", "Eviction policy",
                locAttr.evictionPolicyClassName(), rmtAttr.evictionPolicyClassName(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "evictionPolicyFactory", "Eviction policy factory",
                locAttr.evictionPolicyFactoryClassName(), rmtAttr.evictionPolicyFactoryClassName(), true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "transactionManagerLookup",
                "Transaction manager lookup", locAttr.transactionManagerLookupClassName(),
                rmtAttr.transactionManagerLookupClassName(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "defaultLockTimeout",
                "Default lock timeout", locAttr.defaultLockTimeout(), rmtAttr.defaultLockTimeout(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "preloadBatchSize",
                "Preload batch size", locAttr.rebalanceBatchSize(), rmtAttr.rebalanceBatchSize(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "rebalanceDelay",
                "Rebalance delay", locAttr.rebalanceDelay(), rmtAttr.rebalanceDelay(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "rebalanceBatchesPrefetchCount",
                "Rebalance batches prefetch count", locAttr.rebalanceBatchesPrefetchCount(),
                rmtAttr.rebalanceBatchesPrefetchCount(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "rebalanceOrder",
                "Rebalance order", locAttr.rebalanceOrder(), rmtAttr.rebalanceOrder(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "rebalanceThrottle",
                "Rebalance throttle", locAttr.rebalanceThrottle(), rmtAttr.rebalanceThrottle(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "rebalanceTimeout",
                "Rebalance timeout", locAttr.rebalanceTimeout(), rmtAttr.rebalanceTimeout(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "writeSynchronizationMode",
                "Write synchronization mode", locAttr.writeSynchronization(), rmtAttr.writeSynchronization(),
                true);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "writeBehindBatchSize",
                "Write behind batch size", locAttr.writeBehindBatchSize(), rmtAttr.writeBehindBatchSize(),
                false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "writeBehindCoalescing",
                "Write behind coalescing", locAttr.writeBehindCoalescing(), rmtAttr.writeBehindCoalescing(),
                false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "writeBehindEnabled",
                "Write behind enabled", locAttr.writeBehindEnabled(), rmtAttr.writeBehindEnabled(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "writeBehindFlushFrequency",
                "Write behind flush frequency", locAttr.writeBehindFlushFrequency(),
                rmtAttr.writeBehindFlushFrequency(), false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "writeBehindFlushSize",
                "Write behind flush size", locAttr.writeBehindFlushSize(), rmtAttr.writeBehindFlushSize(),
                false);

            CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "writeBehindFlushThreadCount",
                "Write behind flush thread count", locAttr.writeBehindFlushThreadCount(),
                rmtAttr.writeBehindFlushThreadCount(), false);

            if (locAttr.cacheMode() == PARTITIONED) {
                CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "nearEvictionPolicy",
                    "Near eviction policy", locAttr.nearEvictionPolicyClassName(),
                    rmtAttr.nearEvictionPolicyClassName(), false);

                CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "nearEvictionPolicyFactory",
                    "Near eviction policy factory", locAttr.nearEvictionPolicyFactoryClassName(),
                    rmtAttr.nearEvictionPolicyFactoryClassName(), false);

                CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "affinityIncludeNeighbors",
                    "Affinity include neighbors", locAttr.affinityIncludeNeighbors(),
                    rmtAttr.affinityIncludeNeighbors(), true);

                CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "affinityKeyBackups",
                    "Affinity key backups", locAttr.affinityKeyBackups(),
                    rmtAttr.affinityKeyBackups(), true);

                CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "qryParallelism",
                    "Query parallelism", locAttr.qryParallelism(), rmtAttr.qryParallelism(), true);
            }
        }

        CU.checkAttributeMismatch(log, rmtAttr.cacheName(), rmt, "isEncryptionEnabled",
            "Cache encrypted", locAttr.isEncryptionEnabled(), rmtAttr.isEncryptionEnabled(), true);
    }

    /**
     * @param msg Message.
     * @param node Node sent message.
     */
    public void onClientCacheChange(ClientCacheChangeDiscoveryMessage msg, ClusterNode node) {
        Map<Integer, Boolean> startedCaches = msg.startedCaches();

        if (startedCaches != null) {
            for (Map.Entry<Integer, Boolean> e : startedCaches.entrySet()) {
                for (DynamicCacheDescriptor desc : registeredCaches.values()) {
                    if (e.getKey().equals(desc.cacheId())) {
                        ctx.discovery().addClientNode(desc.cacheName(), node.id(), e.getValue());

                        break;
                    }
                }
            }
        }

        Set<Integer> closedCaches = msg.closedCaches();

        if (closedCaches != null) {
            for (Integer cacheId : closedCaches) {
                for (DynamicCacheDescriptor desc : registeredCaches.values()) {
                    if (cacheId.equals(desc.cacheId())) {
                        ctx.discovery().onClientCacheClose(desc.cacheName(), node.id());

                        break;
                    }
                }
            }
        }
    }

    /**
     * Creates exchanges actions. Forms a list of caches and cache groups to be stopped
     * due to dynamic cache start failure.
     *
     * @param failMsg Dynamic change request fail message.
     * @param topVer Topology version.
     */
    public void onCacheChangeRequested(DynamicCacheChangeFailureMessage failMsg, AffinityTopologyVersion topVer) {
        ExchangeActions exchangeActions = new ExchangeActions();

        List<DynamicCacheChangeRequest> requests = new ArrayList<>(failMsg.cacheNames().size());

        for (String cacheName : failMsg.cacheNames()) {
            DynamicCacheDescriptor cacheDescr = registeredCaches.get(cacheName);

            assert cacheDescr != null : "Dynamic cache descriptor is missing [cacheName=" + cacheName + "]";

            requests.add(DynamicCacheChangeRequest.stopRequest(ctx, cacheName, cacheDescr.sql(), true));
        }

        processCacheChangeRequests(exchangeActions, requests, topVer,false);

        failMsg.exchangeActions(exchangeActions);
    }

    /**
     * @param batch Cache change request.
     * @param topVer Topology version.
     * @return {@code True} if minor topology version should be increased.
     */
    public boolean onCacheChangeRequested(DynamicCacheChangeBatch batch, AffinityTopologyVersion topVer) {
        DiscoveryDataClusterState state = ctx.state().clusterState();

        if (state.active() && !state.transition()) {
            ExchangeActions exchangeActions = new ExchangeActions();

            CacheChangeProcessResult res = processCacheChangeRequests(exchangeActions,
                batch.requests(),
                topVer,
                false);

            if (res.needExchange) {
                assert !exchangeActions.empty() : exchangeActions;

                batch.exchangeActions(exchangeActions);
            }

            return res.needExchange;
        }
        else {
            IgniteCheckedException err = new IgniteCheckedException("Failed to start/stop cache, cluster state change " +
                "is in progress.");

            for (DynamicCacheChangeRequest req : batch.requests()) {
                if (req.template()) {
                    ctx.cache().completeTemplateAddFuture(req.startCacheConfiguration().getName(),
                        req.deploymentId());
                }
                else
                    ctx.cache().completeCacheStartFuture(req, false, err);
            }

            return false;
        }
    }

    /**
     * @param exchangeActions Exchange actions to update.
     * @param reqs Requests.
     * @param topVer Topology version.
     * @param persistedCfgs {@code True} if process start of persisted caches during cluster activation.
     * @return Process result.
     */
    private CacheChangeProcessResult processCacheChangeRequests(
        ExchangeActions exchangeActions,
        Collection<DynamicCacheChangeRequest> reqs,
        AffinityTopologyVersion topVer,
        boolean persistedCfgs) {
        CacheChangeProcessResult res = new CacheChangeProcessResult();

        final List<T2<DynamicCacheChangeRequest, AffinityTopologyVersion>> reqsToComplete = new ArrayList<>();

        for (DynamicCacheChangeRequest req : reqs) {
            if (req.template()) {
                CacheConfiguration ccfg = req.startCacheConfiguration();

                assert ccfg != null : req;

                DynamicCacheDescriptor desc = registeredTemplates.get(req.cacheName());

                if (desc == null) {
                    DynamicCacheDescriptor templateDesc = new DynamicCacheDescriptor(ctx,
                        ccfg,
                        req.cacheType(),
                        null,
                        true,
                        req.initiatingNodeId(),
                        false,
                        false,
                        req.deploymentId(),
                        req.schema());

                    DynamicCacheDescriptor old = registeredTemplates().put(ccfg.getName(), templateDesc);

                    assert old == null;

                    res.addedDescs.add(templateDesc);
                }

                if (!persistedCfgs)
                    ctx.cache().completeTemplateAddFuture(ccfg.getName(), req.deploymentId());

                continue;
            }

            assert !req.clientStartOnly() : req;

            DynamicCacheDescriptor desc = registeredCaches.get(req.cacheName());

            boolean needExchange = false;

            boolean clientCacheStart = false;

            AffinityTopologyVersion waitTopVer = null;

            if (req.start()) {
                // Starting a new cache.
                if (desc == null) {
                    String conflictErr = checkCacheConflict(req.startCacheConfiguration());

                    if (conflictErr != null) {
                        U.warn(log, "Ignore cache start request. " + conflictErr);

                        IgniteCheckedException err = new IgniteCheckedException("Failed to start " +
                            "cache. " + conflictErr);

                        if (persistedCfgs)
                            res.errs.add(err);
                        else
                            ctx.cache().completeCacheStartFuture(req, false, err);

                        continue;
                    }

                    if (req.clientStartOnly()) {
                        assert !persistedCfgs;

                        ctx.cache().completeCacheStartFuture(req, false, new IgniteCheckedException("Failed to start " +
                            "client cache (a cache with the given name is not started): " + req.cacheName()));
                    }
                    else {
                        SchemaOperationException err = QueryUtils.checkQueryEntityConflicts(
                            req.startCacheConfiguration(), registeredCaches.values());

                        if (err != null) {
                            if (persistedCfgs)
                                res.errs.add(err);
                            else
                                ctx.cache().completeCacheStartFuture(req, false, err);

                            continue;
                        }

                        CacheConfiguration<?, ?> ccfg = req.startCacheConfiguration();

                        assert req.cacheType() != null : req;
                        assert F.eq(ccfg.getName(), req.cacheName()) : req;

                        int cacheId = CU.cacheId(req.cacheName());

                        CacheGroupDescriptor grpDesc = registerCacheGroup(exchangeActions,
                            topVer,
                            ccfg,
                            cacheId,
                            req.initiatingNodeId(),
                            req.deploymentId(),
                            req.encryptionKey());

                        DynamicCacheDescriptor startDesc = new DynamicCacheDescriptor(ctx,
                            ccfg,
                            req.cacheType(),
                            grpDesc,
                            false,
                            req.initiatingNodeId(),
                            false,
                            req.sql(),
                            req.deploymentId(),
                            req.schema());

                        DynamicCacheDescriptor old = registeredCaches.put(ccfg.getName(), startDesc);

                        restartingCaches.remove(ccfg.getName());

                        assert old == null;

                        ctx.discovery().setCacheFilter(
                            startDesc.cacheId(),
                            grpDesc.groupId(),
                            ccfg.getName(),
                            ccfg.getNearConfiguration() != null);

                        if (!persistedCfgs) {
                            ctx.discovery().addClientNode(req.cacheName(),
                                req.initiatingNodeId(),
                                req.nearCacheConfiguration() != null);
                        }

                        res.addedDescs.add(startDesc);

                        exchangeActions.addCacheToStart(req, startDesc);

                        needExchange = true;
                    }
                }
                else {
                    assert !persistedCfgs;
                    assert req.initiatingNodeId() != null : req;

                    if (req.failIfExists()) {
                        ctx.cache().completeCacheStartFuture(req, false,
                            new CacheExistsException("Failed to start cache " +
                                "(a cache with the same name is already started): " + req.cacheName()));
                    }
                    else {
                        // Cache already exists, it is possible client cache is needed.
                        ClusterNode node = ctx.discovery().node(req.initiatingNodeId());

                        boolean clientReq = node != null &&
                            !ctx.discovery().cacheAffinityNode(node, req.cacheName());

                        if (clientReq) {
                            ctx.discovery().addClientNode(req.cacheName(),
                                req.initiatingNodeId(),
                                req.nearCacheConfiguration() != null);

                            if (node.id().equals(req.initiatingNodeId())) {
                                desc.clientCacheStartVersion(topVer);

                                clientCacheStart = true;

                                ctx.discovery().clientCacheStartEvent(req.requestId(), F.asMap(req.cacheName(), req), null);
                            }
                        }
                    }
                }

                if (!needExchange && !clientCacheStart && desc != null) {
                    if (desc.clientCacheStartVersion() != null)
                        waitTopVer = desc.clientCacheStartVersion();
                    else {
                        AffinityTopologyVersion nodeStartVer =
                            new AffinityTopologyVersion(ctx.discovery().localNode().order(), 0);

                        if (desc.startTopologyVersion() != null)
                            waitTopVer = desc.startTopologyVersion();
                        else
                            waitTopVer = desc.receivedFromStartVersion();

                        if (waitTopVer == null || nodeStartVer.compareTo(waitTopVer) > 0)
                            waitTopVer = nodeStartVer;
                    }
                }
            }
            else if (req.resetLostPartitions()) {
                if (desc != null) {
                    needExchange = true;

                    exchangeActions.addCacheToResetLostPartitions(req, desc);
                }
            }
            else if (req.stop()) {
                if (desc != null) {
                    if (req.sql() && !desc.sql()) {
                        ctx.cache().completeCacheStartFuture(req, false,
                            new IgniteCheckedException("Only cache created with CREATE TABLE may be removed with " +
                                "DROP TABLE [cacheName=" + req.cacheName() + ']'));

                        continue;
                    }

                    if (!req.sql() && desc.sql()) {
                        ctx.cache().completeCacheStartFuture(req, false,
                            new IgniteCheckedException("Only cache created with cache API may be removed with " +
                                "direct call to destroyCache [cacheName=" + req.cacheName() + ']'));

                        continue;
                    }

                    DynamicCacheDescriptor old = registeredCaches.remove(req.cacheName());

                    if (req.restart())
                        restartingCaches.add(req.cacheName());

                    assert old != null && old == desc : "Dynamic cache map was concurrently modified [req=" + req + ']';

                    ctx.discovery().removeCacheFilter(req.cacheName());

                    needExchange = true;

                    exchangeActions.addCacheToStop(req, desc);

                    CacheGroupDescriptor grpDesc = registeredCacheGrps.get(desc.groupId());

                    assert grpDesc != null && grpDesc.groupId() == desc.groupId() : desc;

                    grpDesc.onCacheStopped(desc.cacheName(), desc.cacheId());

                    if (!grpDesc.hasCaches()) {
                        registeredCacheGrps.remove(grpDesc.groupId());

                        ctx.discovery().removeCacheGroup(grpDesc);

                        exchangeActions.addCacheGroupToStop(grpDesc, req.destroy());

                        assert exchangeActions.checkStopRequestConsistency(grpDesc.groupId());

                        // If all caches in group will be destroyed it is not necessary to destroy single cache
                        // because group will be stopped anyway.
                        if (req.destroy()) {
                            for (ExchangeActions.CacheActionData action : exchangeActions.cacheStopRequests()) {
                                if (action.descriptor().groupId() == grpDesc.groupId())
                                    action.request().destroy(false);
                            }
                        }
                    }
                }
            }
            else
                assert false : req;

            if (!needExchange) {
                if (!clientCacheStart && ctx.localNodeId().equals(req.initiatingNodeId()))
                    reqsToComplete.add(new T2<>(req, waitTopVer));
            }
            else
                res.needExchange = true;
        }

        if (!F.isEmpty(res.addedDescs)) {
            AffinityTopologyVersion startTopVer = res.needExchange ? topVer.nextMinorVersion() : topVer;

            for (DynamicCacheDescriptor desc : res.addedDescs) {
                assert desc.template() || res.needExchange;

                desc.startTopologyVersion(startTopVer);
            }
        }

        if (!F.isEmpty(reqsToComplete)) {
            ctx.closure().callLocalSafe(new Callable<Void>() {
                @Override public Void call() throws Exception {
                    for (T2<DynamicCacheChangeRequest, AffinityTopologyVersion> t : reqsToComplete) {
                        final DynamicCacheChangeRequest req = t.get1();
                        AffinityTopologyVersion waitTopVer = t.get2();

                        IgniteInternalFuture<?> fut = waitTopVer != null ?
                            ctx.cache().context().exchange().affinityReadyFuture(waitTopVer) : null;

                        if (fut == null || fut.isDone())
                            ctx.cache().completeCacheStartFuture(req, false, null);
                        else {
                            fut.listen(new IgniteInClosure<IgniteInternalFuture<?>>() {
                                @Override public void apply(IgniteInternalFuture<?> fut) {
                                    ctx.cache().completeCacheStartFuture(req, false, null);
                                }
                            });
                        }
                    }

                    return null;
                }
            });
        }

        return res;
    }

    /**
     * @param dataBag Discovery data bag.
     */
    void collectJoiningNodeData(DiscoveryDataBag dataBag) {
        if (!ctx.isDaemon())
            dataBag.addJoiningNodeData(CACHE_PROC.ordinal(), joinDiscoveryData());
    }

    /**
     * @return {@code True} if there are currently restarting caches.
     */
    boolean hasRestartingCaches() {
        return !F.isEmpty(restartingCaches);
    }

    /**
     * @return Collection of currently restarting caches.
     */
    Collection<String> restartingCaches() {
        return restartingCaches;
    }

    /**
     * @return Discovery date sent on local node join.
     */
    private Serializable joinDiscoveryData() {
        if (cachesOnDisconnect != null) {
            Map<Integer, CacheClientReconnectDiscoveryData.CacheGroupInfo> cacheGrpsInfo = new HashMap<>();
            Map<String, CacheClientReconnectDiscoveryData.CacheInfo> cachesInfo = new HashMap<>();

            Map<Integer, CacheGroupDescriptor> grps = cachesOnDisconnect.cacheGrps;
            Map<String, DynamicCacheDescriptor> caches = cachesOnDisconnect.caches;

            for (CacheGroupContext grp : ctx.cache().cacheGroups()) {
                CacheGroupDescriptor desc = grps.get(grp.groupId());

                assert desc != null : grp.cacheOrGroupName();

                cacheGrpsInfo.put(grp.groupId(), new CacheClientReconnectDiscoveryData.CacheGroupInfo(desc.config(),
                    desc.deploymentId(),
                    0));
            }

            for (IgniteInternalCache cache : ctx.cache().caches()) {
                DynamicCacheDescriptor desc = caches.get(cache.name());

                assert desc != null : cache.name();

                cachesInfo.put(cache.name(), new CacheClientReconnectDiscoveryData.CacheInfo(desc.cacheConfiguration(),
                    desc.cacheType(),
                    desc.deploymentId(),
                    cache.context().isNear(),
                    0));
            }

            return new CacheClientReconnectDiscoveryData(cacheGrpsInfo, cachesInfo);
        }
        else {
            assert ctx.config().isDaemon() || joinDiscoData != null;

            return joinDiscoData;
        }
    }

    /**
     * Called from exchange worker.
     *
     * @return Caches to be started when this node starts.
     */
    @Nullable public LocalJoinCachesContext localJoinCachesContext() {
        if (ctx.isDaemon())
            return null;

        LocalJoinCachesContext result = locJoinCachesCtx;

        locJoinCachesCtx = null;

        return result;
    }

    /**
     * @param joinedNodeId Joined node ID.
     * @return {@code True} if there are new caches received from joined node.
     */
    boolean hasCachesReceivedFromJoin(UUID joinedNodeId) {
        for (DynamicCacheDescriptor desc : registeredCaches.values()) {
            if (desc.staticallyConfigured()) {
                assert desc.receivedFrom() != null : desc;

                if (joinedNodeId.equals(desc.receivedFrom()))
                    return true;
            }
        }

        return false;
    }

    /**
     * @param joinedNodeId Joined node ID.
     * @return New caches received from joined node.
     */
    List<DynamicCacheDescriptor> cachesReceivedFromJoin(UUID joinedNodeId) {
        assert joinedNodeId != null;

        List<DynamicCacheDescriptor> started = null;

        if (!ctx.isDaemon()) {
            for (DynamicCacheDescriptor desc : orderedCaches(CacheComparators.DIRECT)) {
                if (desc.staticallyConfigured()) {
                    assert desc.receivedFrom() != null : desc;

                    if (joinedNodeId.equals(desc.receivedFrom())) {
                        if (started == null)
                            started = new ArrayList<>();

                        started.add(desc);
                    }
                }
            }
        }

        return started != null ? started : Collections.<DynamicCacheDescriptor>emptyList();
    }

    /**
     * Discovery event callback, executed from discovery thread.
     *
     * @param type Event type.
     * @param node Event node.
     * @param topVer Topology version.
     */
    public void onDiscoveryEvent(int type, ClusterNode node, AffinityTopologyVersion topVer) {
        if (type == EVT_NODE_JOINED && !ctx.isDaemon()) {
            for (CacheGroupDescriptor desc : registeredCacheGrps.values()) {
                if (node.id().equals(desc.receivedFrom()))
                    desc.receivedFromStartVersion(topVer);
            }

            for (DynamicCacheDescriptor desc : registeredCaches.values()) {
                if (node.id().equals(desc.receivedFrom()))
                    desc.receivedFromStartVersion(topVer);
            }

            for (DynamicCacheDescriptor desc : registeredTemplates.values()) {
                if (node.id().equals(desc.receivedFrom()))
                    desc.receivedFromStartVersion(topVer);
            }

            if (node.id().equals(ctx.discovery().localNode().id())) {
                if (gridData == null) { // First node starts.
                    assert joinDiscoData != null;

                    initStartCachesForLocalJoin(true, false);
                }
            }
        }
    }

    /**
     * @param dataBag Discovery data bag.
     */
    public void collectGridNodeData(DiscoveryDataBag dataBag) {
        if (ctx.isDaemon())
            return;

        if (!dataBag.commonDataCollectedFor(CACHE_PROC.ordinal()))
            dataBag.addGridCommonData(CACHE_PROC.ordinal(), collectCommonDiscoveryData());
    }

    /**
     * @return Information about started caches.
     */
    private CacheNodeCommonDiscoveryData collectCommonDiscoveryData() {
        Map<Integer, CacheGroupData> cacheGrps = new HashMap<>();

        for (CacheGroupDescriptor grpDesc : registeredCacheGrps.values()) {
            CacheGroupData grpData = new CacheGroupData(grpDesc.config(),
                grpDesc.groupName(),
                grpDesc.groupId(),
                grpDesc.receivedFrom(),
                grpDesc.startTopologyVersion(),
                grpDesc.deploymentId(),
                grpDesc.caches(),
                0,
                grpDesc.persistenceEnabled(),
                grpDesc.walEnabled(),
                grpDesc.walChangeRequests());

            cacheGrps.put(grpDesc.groupId(), grpData);
        }

        Map<String, CacheData> caches = new HashMap<>();

        for (DynamicCacheDescriptor desc : registeredCaches.values()) {
            CacheData cacheData = new CacheData(desc.cacheConfiguration(),
                desc.cacheId(),
                desc.groupId(),
                desc.cacheType(),
                desc.deploymentId(),
                desc.schema(),
                desc.receivedFrom(),
                desc.staticallyConfigured(),
                desc.sql(),
                false,
                0);

            caches.put(desc.cacheName(), cacheData);
        }

        Map<String, CacheData> templates = new HashMap<>();

        for (DynamicCacheDescriptor desc : registeredTemplates.values()) {
            CacheData cacheData = new CacheData(desc.cacheConfiguration(),
                0,
                0,
                desc.cacheType(),
                desc.deploymentId(),
                desc.schema(),
                desc.receivedFrom(),
                desc.staticallyConfigured(),
                false,
                true,
                0);

            templates.put(desc.cacheName(), cacheData);
        }

        Collection<String> restarting = new HashSet<>(restartingCaches);

        return new CacheNodeCommonDiscoveryData(caches,
            templates,
            cacheGrps,
            ctx.discovery().clientNodesMap(),
            restarting);
    }

    /**
     * @param data Discovery data.
     */
    public void onGridDataReceived(DiscoveryDataBag.GridDiscoveryData data) {
        if (ctx.isDaemon() || data.commonData() == null)
            return;

        assert joinDiscoData != null || disconnectedState();
        assert data.commonData() instanceof CacheNodeCommonDiscoveryData : data;

        CacheNodeCommonDiscoveryData cachesData = (CacheNodeCommonDiscoveryData)data.commonData();

        // CacheGroup configurations that were created from local node configuration.
        Map<Integer, CacheGroupDescriptor> locCacheGrps = new HashMap<>(registeredCacheGroups());

        //Replace locally registered data with actual data received from cluster.
        cleanCachesAndGroups();

        registerReceivedCacheGroups(cachesData, locCacheGrps);

        registerReceivedCacheTemplates(cachesData);

        registerReceivedCaches(cachesData);

        addReceivedClientNodesToDiscovery(cachesData);

        String conflictErr = validateRegisteredCaches();

        gridData = new GridData(joinDiscoData, cachesData, conflictErr);

        if (cachesOnDisconnect == null || cachesOnDisconnect.clusterActive())
            initStartCachesForLocalJoin(false, disconnectedState());
    }

    /**
     * Validation {@link #registeredCaches} on conflicts.
     *
     * @return Error message if conflicts was found.
     */
    @Nullable private String validateRegisteredCaches() {
        String conflictErr = null;

        if (joinDiscoData != null) {
            for (Map.Entry<String, CacheJoinNodeDiscoveryData.CacheInfo> e : joinDiscoData.caches().entrySet()) {
                if (!registeredCaches.containsKey(e.getKey())) {
                    conflictErr = checkCacheConflict(e.getValue().cacheData().config());

                    if (conflictErr != null) {
                        conflictErr = "Failed to start configured cache due to conflict with started caches. " +
                            conflictErr;

                        break;
                    }
                }
            }
        }

        return conflictErr;
    }

    /**
     * Adding received client nodes to discovery if needed.
     *
     * @param cachesData Data received from cluster.
     */
    private void addReceivedClientNodesToDiscovery(CacheNodeCommonDiscoveryData cachesData) {
        if (!F.isEmpty(cachesData.clientNodesMap())) {
            for (Map.Entry<String, Map<UUID, Boolean>> entry : cachesData.clientNodesMap().entrySet()) {
                String cacheName = entry.getKey();

                for (Map.Entry<UUID, Boolean> tup : entry.getValue().entrySet())
                    ctx.discovery().addClientNode(cacheName, tup.getKey(), tup.getValue());
            }
        }
    }

    /**
     * Register caches received from cluster.
     *
     * @param cachesData Data received from cluster.
     */
    private void registerReceivedCaches(CacheNodeCommonDiscoveryData cachesData) {
        Map<DynamicCacheDescriptor, QuerySchemaPatch> patchesToApply = new HashMap<>();
        Collection<DynamicCacheDescriptor> cachesToSave = new HashSet<>();

        boolean hasSchemaPatchConflict = false;

        for (CacheData cacheData : cachesData.caches().values()) {
            CacheGroupDescriptor grpDesc = registeredCacheGrps.get(cacheData.groupId());

            assert grpDesc != null : cacheData.cacheConfiguration().getName();

            CacheConfiguration<?, ?> cfg = cacheData.cacheConfiguration();

            DynamicCacheDescriptor desc = new DynamicCacheDescriptor(
                ctx,
                cacheData.cacheConfiguration(),
                cacheData.cacheType(),
                grpDesc,
                false,
                cacheData.receivedFrom(),
                cacheData.staticallyConfigured(),
                cacheData.sql(),
                cacheData.deploymentId(),
                new QuerySchema(cacheData.schema().entities())
            );

            Collection<QueryEntity> localQueryEntities = getLocalQueryEntities(cfg.getName());

            QuerySchemaPatch schemaPatch = desc.makeSchemaPatch(localQueryEntities);

            if (schemaPatch.hasConflicts()) {
                hasSchemaPatchConflict = true;

                log.warning("Skipping apply patch because conflicts : " + schemaPatch.getConflictsMessage());
            }
            else if (!schemaPatch.isEmpty())
                patchesToApply.put(desc, schemaPatch);
            else if (!GridFunc.eqNotOrdered(desc.schema().entities(), localQueryEntities))
                cachesToSave.add(desc); //received config is different of local config - need to resave

            desc.receivedOnDiscovery(true);

            registeredCaches.put(cacheData.cacheConfiguration().getName(), desc);

            ctx.discovery().setCacheFilter(
                desc.cacheId(),
                grpDesc.groupId(),
                cfg.getName(),
                cfg.getNearConfiguration() != null);
        }

        updateRegisteredCachesIfNeeded(patchesToApply, cachesToSave, hasSchemaPatchConflict);
    }

    /**
     * Merging config or resaving it if it needed.
     *
     * @param patchesToApply Patches which need to apply.
     * @param cachesToSave Caches which need to resave.
     * @param hasSchemaPatchConflict {@code true} if we have conflict during making patch.
     */
    private void updateRegisteredCachesIfNeeded(Map<DynamicCacheDescriptor, QuerySchemaPatch> patchesToApply,
        Collection<DynamicCacheDescriptor> cachesToSave, boolean hasSchemaPatchConflict) {
        //Skip merge of config if least one conflict was found.
        if (!hasSchemaPatchConflict && isMergeConfigSupports(ctx.discovery().localNode())) {
            boolean isClusterActive = ctx.state().clusterState().active();

            //Merge of config for cluster only for inactive grid.
            if (!isClusterActive && !patchesToApply.isEmpty()) {
                for (Map.Entry<DynamicCacheDescriptor, QuerySchemaPatch> entry : patchesToApply.entrySet()) {
                    if (entry.getKey().applySchemaPatch(entry.getValue()))
                        saveCacheConfiguration(entry.getKey());
                }

                for (DynamicCacheDescriptor descriptor : cachesToSave) {
                    saveCacheConfiguration(descriptor);
                }
            }
            else if (patchesToApply.isEmpty()) {
                for (DynamicCacheDescriptor descriptor : cachesToSave) {
                    saveCacheConfiguration(descriptor);
                }
            }
        }
    }

    /**
     * Register cache templates received from cluster.
     *
     * @param cachesData Data received from cluster.
     */
    private void registerReceivedCacheTemplates(CacheNodeCommonDiscoveryData cachesData) {
        for (CacheData cacheData : cachesData.templates().values()) {
            DynamicCacheDescriptor desc = new DynamicCacheDescriptor(
                ctx,
                cacheData.cacheConfiguration(),
                cacheData.cacheType(),
                null,
                true,
                cacheData.receivedFrom(),
                cacheData.staticallyConfigured(),
                false,
                cacheData.deploymentId(),
                cacheData.schema());

            registeredTemplates.put(cacheData.cacheConfiguration().getName(), desc);
        }
    }

    /**
     * Register cache groups received from cluster.
     *
     * @param cachesData Data received from cluster.
     * @param locCacheGrps Current local cache groups.
     */
    private void registerReceivedCacheGroups(CacheNodeCommonDiscoveryData cachesData,
        Map<Integer, CacheGroupDescriptor> locCacheGrps) {
        for (CacheGroupData grpData : cachesData.cacheGroups().values()) {
            CacheGroupDescriptor grpDesc = new CacheGroupDescriptor(
                grpData.config(),
                grpData.groupName(),
                grpData.groupId(),
                grpData.receivedFrom(),
                grpData.startTopologyVersion(),
                grpData.deploymentId(),
                grpData.caches(),
                grpData.persistenceEnabled(),
                grpData.walEnabled(),
                grpData.walChangeRequests());

            if (locCacheGrps.containsKey(grpDesc.groupId())) {
                CacheGroupDescriptor locGrpCfg = locCacheGrps.get(grpDesc.groupId());

                grpDesc.mergeWith(locGrpCfg);
            }

            CacheGroupDescriptor old = registeredCacheGrps.put(grpDesc.groupId(), grpDesc);

            assert old == null : old;

            ctx.discovery().addCacheGroup(grpDesc,
                grpData.config().getNodeFilter(),
                grpData.config().getCacheMode());
        }
    }

    /**
     * Clean local registered caches and groups
     */
    private void cleanCachesAndGroups() {
        registeredCaches.clear();
        registeredCacheGrps.clear();
        ctx.discovery().cleanCachesAndGroups();
    }

    /**
     * Save dynamic cache descriptor on disk.
     *
     * @param desc Cache to save.
     */
    private void saveCacheConfiguration(DynamicCacheDescriptor desc) {
        try {
            ctx.cache().saveCacheConfiguration(desc);
        }
        catch (IgniteCheckedException e) {
            log.error("Error while saving cache configuration to disk, cfg = " + desc.cacheConfiguration(), e);
        }
    }

    /**
     * Get started node query entities by cacheName.
     *
     * @param cacheName Cache for which query entities will be returned.
     * @return Local query entities.
     */
    private Collection<QueryEntity> getLocalQueryEntities(String cacheName) {
        if (joinDiscoData == null)
            return Collections.emptyList();

        CacheJoinNodeDiscoveryData.CacheInfo cacheInfo = joinDiscoData.caches().get(cacheName);

        if (cacheInfo == null)
            return Collections.emptyList();

        return cacheInfo.cacheData().queryEntities();
    }

    /**
     * Initialize collection with caches to be start:
     * {@code locJoinStartCaches} or {@code locCfgsForActivation} if cluster is inactive.
     *
     * @param firstNode {@code True} if first node in cluster starts.
     */
    private void initStartCachesForLocalJoin(boolean firstNode, boolean reconnect) {
        assert locJoinCachesCtx == null : locJoinCachesCtx;

        if (ctx.state().clusterState().transition()) {
            joinOnTransition = true;

            return;
        }

        if (joinDiscoData != null) {
            List<T2<DynamicCacheDescriptor, NearCacheConfiguration>> locJoinStartCaches = new ArrayList<>();
            locCfgsForActivation = new HashMap<>();

            boolean active = ctx.state().clusterState().active();

            for (DynamicCacheDescriptor desc : orderedCaches(CacheComparators.DIRECT)) {
                if (firstNode && !joinDiscoData.caches().containsKey(desc.cacheName()))
                    continue;

                CacheConfiguration<?, ?> cfg = desc.cacheConfiguration();

                if (reconnect && surviveReconnect(cfg.getName()) && cachesOnDisconnect.state.active() && active)
                    continue;

                CacheJoinNodeDiscoveryData.CacheInfo locCfg = joinDiscoData.caches().get(cfg.getName());

                NearCacheConfiguration nearCfg = null;

                if (locCfg != null) {
                    nearCfg = locCfg.cacheData().config().getNearConfiguration();

                    DynamicCacheDescriptor desc0 = new DynamicCacheDescriptor(ctx,
                        locCfg.cacheData().config(),
                        desc.cacheType(),
                        desc.groupDescriptor(),
                        desc.template(),
                        desc.receivedFrom(),
                        desc.staticallyConfigured(),
                        desc.sql(),
                        desc.deploymentId(),
                        desc.schema().copy());

                    desc0.startTopologyVersion(desc.startTopologyVersion());
                    desc0.receivedFromStartVersion(desc.receivedFromStartVersion());
                    desc0.clientCacheStartVersion(desc.clientCacheStartVersion());

                    desc0.cacheConfiguration().setStatisticsEnabled(cfg.isStatisticsEnabled());

                    desc = desc0;
                }

                if (locCfg != null ||
                    joinDiscoData.startCaches() ||
                    CU.affinityNode(ctx.discovery().localNode(), desc.groupDescriptor().config().getNodeFilter())) {
                    if (active)
                        locJoinStartCaches.add(new T2<>(desc, nearCfg));
                    else
                        locCfgsForActivation.put(desc.cacheName(), new T2<>(desc.cacheConfiguration(), nearCfg));
                }
            }

            locJoinCachesCtx = new LocalJoinCachesContext(
                locJoinStartCaches,
                new HashMap<>(registeredCacheGrps),
                new HashMap<>(registeredCaches));
        }
    }

    /**
     * @param msg Message.
     */
    public void onStateChangeFinish(ChangeGlobalStateFinishMessage msg) {
        if (joinOnTransition) {
            initStartCachesForLocalJoin(false, false);

            joinOnTransition = false;
        }
    }

    /**
     * @param msg Message.
     * @param topVer Current topology version.
     * @param curState Current cluster state.
     * @return Exchange action.
     * @throws IgniteCheckedException If configuration validation failed.
     */
    public ExchangeActions onStateChangeRequest(ChangeGlobalStateMessage msg, AffinityTopologyVersion topVer, DiscoveryDataClusterState curState)
        throws IgniteCheckedException {
        ExchangeActions exchangeActions = new ExchangeActions();

        if (msg.activate() == curState.active())
            return exchangeActions;

        if (msg.activate()) {
            for (DynamicCacheDescriptor desc : orderedCaches(CacheComparators.DIRECT)) {
                desc.startTopologyVersion(topVer);

                DynamicCacheChangeRequest req = new DynamicCacheChangeRequest(msg.requestId(),
                    desc.cacheName(),
                    msg.initiatorNodeId());

                req.startCacheConfiguration(desc.cacheConfiguration());
                req.cacheType(desc.cacheType());

                T2<CacheConfiguration, NearCacheConfiguration> locCfg = locCfgsForActivation.get(desc.cacheName());

                if (locCfg != null) {
                    if (locCfg.get1() != null)
                        req.startCacheConfiguration(locCfg.get1());

                    req.nearCacheConfiguration(locCfg.get2());

                    req.locallyConfigured(true);
                }

                exchangeActions.addCacheToStart(req, desc);
            }

            for (CacheGroupDescriptor grpDesc : registeredCacheGroups().values())
                exchangeActions.addCacheGroupToStart(grpDesc);

            List<StoredCacheData> storedCfgs = msg.storedCacheConfigurations();

            if (storedCfgs != null) {
                List<DynamicCacheChangeRequest> reqs = new ArrayList<>();

                IgniteUuid deploymentId = msg.id();

                for (StoredCacheData storedCfg : storedCfgs) {
                    CacheConfiguration ccfg = storedCfg.config();

                    if (!registeredCaches.containsKey(ccfg.getName())) {
                        DynamicCacheChangeRequest req = new DynamicCacheChangeRequest(msg.requestId(),
                            ccfg.getName(),
                            msg.initiatorNodeId());

                        req.deploymentId(deploymentId);
                        req.startCacheConfiguration(ccfg);
                        req.cacheType(ctx.cache().cacheType(ccfg.getName()));
                        req.schema(new QuerySchema(storedCfg.queryEntities()));
                        req.sql(storedCfg.sql());

                        reqs.add(req);
                    }
                }

                CacheChangeProcessResult res = processCacheChangeRequests(exchangeActions, reqs, topVer, true);

                if (!res.errs.isEmpty()) {
                    IgniteCheckedException err = new IgniteCheckedException("Failed to activate cluster.");

                    for (IgniteCheckedException err0 : res.errs)
                        err.addSuppressed(err0);

                    throw err;
                }
            }
        }
        else {
            locCfgsForActivation = new HashMap<>();

            for (DynamicCacheDescriptor desc : orderedCaches(CacheComparators.REVERSE)) {
                DynamicCacheChangeRequest req = DynamicCacheChangeRequest.stopRequest(ctx,
                    desc.cacheName(),
                    desc.sql(),
                    false);

                exchangeActions.addCacheToStop(req, desc);

                if (ctx.discovery().cacheClientNode(ctx.discovery().localNode(), desc.cacheName()))
                    locCfgsForActivation.put(desc.cacheName(), new T2<>((CacheConfiguration)null, (NearCacheConfiguration)null));
            }

            for (CacheGroupDescriptor grpDesc : registeredCacheGroups().values())
                exchangeActions.addCacheGroupToStop(grpDesc, false);
        }

        return exchangeActions;
    }

    /**
     * @param data Joining node data.
     */
    public void onJoiningNodeDataReceived(DiscoveryDataBag.JoiningNodeDiscoveryData data) {
        if (data.hasJoiningNodeData()) {
            Serializable joiningNodeData = data.joiningNodeData();

            if (joiningNodeData instanceof CacheClientReconnectDiscoveryData) {
                if (disconnectedState()) {
                    if (clientReconnectReqs == null)
                        clientReconnectReqs = new LinkedHashMap<>();

                    clientReconnectReqs.put(data.joiningNodeId(), (CacheClientReconnectDiscoveryData)joiningNodeData);
                }
                else
                    processClientReconnectData((CacheClientReconnectDiscoveryData)joiningNodeData, data.joiningNodeId());
            }
            else if (joiningNodeData instanceof CacheJoinNodeDiscoveryData)
                processJoiningNode((CacheJoinNodeDiscoveryData)joiningNodeData, data.joiningNodeId(), false);
        }
    }

    /**
     * @param clientData Discovery data.
     * @param clientNodeId Client node ID.
     */
    private void processClientReconnectData(CacheClientReconnectDiscoveryData clientData, UUID clientNodeId) {
        DiscoveryDataClusterState state = ctx.state().clusterState();

        if (state.active() && !state.transition()) {
            for (CacheClientReconnectDiscoveryData.CacheInfo cacheInfo : clientData.clientCaches().values()) {
                String cacheName = cacheInfo.config().getName();

                if (surviveReconnect(cacheName))
                    ctx.discovery().addClientNode(cacheName, clientNodeId, false);
                else {
                    DynamicCacheDescriptor desc = registeredCaches.get(cacheName);

                    if (desc != null && desc.deploymentId().equals(cacheInfo.deploymentId()))
                        ctx.discovery().addClientNode(cacheName, clientNodeId, cacheInfo.nearCache());
                }
            }
        }
    }

    /**
     * Checks cache configuration on conflict with already registered caches and cache groups.
     *
     * @param cfg Cache configuration.
     * @return {@code null} if validation passed, error message in other case.
     */
    private String checkCacheConflict(CacheConfiguration<?, ?> cfg) {
        int cacheId = CU.cacheId(cfg.getName());

        if (cacheGroupByName(cfg.getName()) != null)
            return "Cache name conflict with existing cache group (change cache name) [cacheName=" + cfg.getName() + ']';

        if (cfg.getGroupName() != null) {
            DynamicCacheDescriptor desc = registeredCaches.get(cfg.getGroupName());

            if (desc != null)
                return "Cache group name conflict with existing cache (change group name) [cacheName=" + cfg.getName() +
                    ", conflictingCacheName=" + desc.cacheName() + ']';
        }

        for (DynamicCacheDescriptor desc : registeredCaches.values()) {
            if (desc.cacheId() == cacheId)
                return "Cache ID conflict (change cache name) [cacheName=" + cfg.getName() +
                    ", conflictingCacheName=" + desc.cacheName() + ']';
        }

        int grpId = CU.cacheGroupId(cfg.getName(), cfg.getGroupName());

        if (cfg.getGroupName() != null) {
            if (cacheGroupByName(cfg.getGroupName()) == null) {
                CacheGroupDescriptor desc = registeredCacheGrps.get(grpId);

                if (desc != null)
                    return "Cache group ID conflict (change cache group name) [cacheName=" + cfg.getName() +
                        ", groupName=" + cfg.getGroupName() +
                        (desc.sharedGroup() ? ", conflictingGroupName=" : ", conflictingCacheName=") + desc.cacheOrGroupName() + ']';
            }
        }
        else {
            CacheGroupDescriptor desc = registeredCacheGrps.get(grpId);

            if (desc != null)
                return "Cache group ID conflict (change cache name) [cacheName=" + cfg.getName() +
                    (desc.sharedGroup() ? ", conflictingGroupName=" : ", conflictingCacheName=") + desc.cacheOrGroupName() + ']';
        }

        return null;
    }

    /**
     * @param joinData Joined node discovery data.
     * @param nodeId Joined node ID.
     * @param locJoin {@code True} if called on local node join.
     * @return Configuration conflict error.
     */
    private String processJoiningNode(CacheJoinNodeDiscoveryData joinData, UUID nodeId, boolean locJoin) {
        registerNewCacheTemplates(joinData, nodeId);

        Map<DynamicCacheDescriptor, QuerySchemaPatch> patchesToApply = new HashMap<>();

        boolean hasSchemaPatchConflict = false;
        boolean active = ctx.state().clusterState().active();

        boolean isMergeConfigSupport = isMergeConfigSupports(null);

        for (CacheJoinNodeDiscoveryData.CacheInfo cacheInfo : joinData.caches().values()) {
            CacheConfiguration<?, ?> cfg = cacheInfo.cacheData().config();

            if (!registeredCaches.containsKey(cfg.getName())) {
                String conflictErr = checkCacheConflict(cfg);

                if (conflictErr != null) {
                    if (locJoin)
                        return conflictErr;

                    U.warn(log, "Ignore cache received from joining node. " + conflictErr);

                    continue;
                }

                registerNewCache(joinData, nodeId, cacheInfo);
            }
            else if (!active && isMergeConfigSupport) {
                DynamicCacheDescriptor desc = registeredCaches.get(cfg.getName());

                QuerySchemaPatch schemaPatch = desc.makeSchemaPatch(cacheInfo.cacheData().queryEntities());

                if (schemaPatch.hasConflicts()) {
                    hasSchemaPatchConflict = true;

                    log.error("Error during making schema patch : " + schemaPatch.getConflictsMessage());
                }
                else if (!schemaPatch.isEmpty() && !hasSchemaPatchConflict)
                    patchesToApply.put(desc, schemaPatch);
            }

            ctx.discovery().addClientNode(cfg.getName(), nodeId, cfg.getNearConfiguration() != null);
        }

        //If conflict was detected we don't merge config and we leave existed config.
        if (!hasSchemaPatchConflict && !patchesToApply.isEmpty())
            for(Map.Entry<DynamicCacheDescriptor, QuerySchemaPatch> entry: patchesToApply.entrySet()){
                if (entry.getKey().applySchemaPatch(entry.getValue()))
                    saveCacheConfiguration(entry.getKey());
            }

        if (joinData.startCaches()) {
            for (DynamicCacheDescriptor desc : registeredCaches.values()) {
                ctx.discovery().addClientNode(desc.cacheName(),
                    nodeId,
                    desc.cacheConfiguration().getNearConfiguration() != null);
            }
        }

        return null;
    }

    /**
     * Register new cache received from joining node.
     *
     * @param joinData Data from joining node.
     * @param nodeId Joining node id.
     * @param cacheInfo Cache info of new node.
     */
    private void registerNewCache(
        CacheJoinNodeDiscoveryData joinData,
        UUID nodeId,
        CacheJoinNodeDiscoveryData.CacheInfo cacheInfo) {
        CacheConfiguration<?, ?> cfg = cacheInfo.cacheData().config();

        int cacheId = CU.cacheId(cfg.getName());

        CacheGroupDescriptor grpDesc = registerCacheGroup(null,
            null,
            cfg,
            cacheId,
            nodeId,
            joinData.cacheDeploymentId(),
            null);

        ctx.discovery().setCacheFilter(
            cacheId,
            grpDesc.groupId(),
            cfg.getName(),
            cfg.getNearConfiguration() != null);

        DynamicCacheDescriptor desc = new DynamicCacheDescriptor(ctx,
            cfg,
            cacheInfo.cacheType(),
            grpDesc,
            false,
            nodeId,
            cacheInfo.isStaticallyConfigured(),
            cacheInfo.sql(),
            joinData.cacheDeploymentId(),
            new QuerySchema(cacheInfo.cacheData().queryEntities()));

        DynamicCacheDescriptor old = registeredCaches.put(cfg.getName(), desc);

        assert old == null : old;
    }

    /**
     * Register new cache templates received from joining node.
     *
     * @param joinData Data from joining node.
     * @param nodeId Joining node id.
     */
    private void registerNewCacheTemplates(CacheJoinNodeDiscoveryData joinData, UUID nodeId) {
        for (CacheJoinNodeDiscoveryData.CacheInfo cacheInfo : joinData.templates().values()) {
            CacheConfiguration<?, ?> cfg = cacheInfo.cacheData().config();

            if (!registeredTemplates.containsKey(cfg.getName())) {
                DynamicCacheDescriptor desc = new DynamicCacheDescriptor(ctx,
                    cfg,
                    cacheInfo.cacheType(),
                    null,
                    true,
                    nodeId,
                    true,
                    false,
                    joinData.cacheDeploymentId(),
                    new QuerySchema(cacheInfo.cacheData().queryEntities()));

                DynamicCacheDescriptor old = registeredTemplates.put(cfg.getName(), desc);

                assert old == null : old;
            }
        }
    }

    /**
     * @return {@code true} if grid supports merge of config and {@code False} otherwise.
     */
    public boolean isMergeConfigSupports(ClusterNode joiningNode) {
        DiscoCache discoCache = ctx.discovery().discoCache();

        if (discoCache == null)
            return true;

        if (joiningNode != null && joiningNode.version().compareToIgnoreTimestamp(V_MERGE_CONFIG_SINCE) < 0)
            return false;

        Collection<ClusterNode> nodes = discoCache.allNodes();

        for (ClusterNode node : nodes) {
            IgniteProductVersion version = node.version();

            if (version.compareToIgnoreTimestamp(V_MERGE_CONFIG_SINCE) < 0)
                return false;
        }

        return true;
    }

    /**
     * @param grpName Group name.
     * @return Group descriptor if group found.
     */
    @Nullable private CacheGroupDescriptor cacheGroupByName(String grpName) {
        assert grpName != null;

        for (CacheGroupDescriptor grpDesc : registeredCacheGrps.values()) {
            if (grpName.equals(grpDesc.groupName()))
                return grpDesc;
        }

        return null;
    }

    /**
     * @param cacheName Cache name.
     * @return Group descriptor.
     */
    @Nullable private CacheGroupDescriptor nonSharedCacheGroupByCacheName(String cacheName) {
        assert cacheName != null;

        for (CacheGroupDescriptor grpDesc : registeredCacheGrps.values()) {
            if (!grpDesc.sharedGroup() && grpDesc.caches().containsKey(cacheName))
                return grpDesc;
        }

        return null;
    }

    /**
     * @param exchActions Optional exchange actions to update if new group was added.
     * @param curTopVer Current topology version if dynamic cache started.
     * @param startedCacheCfg Cache configuration.
     * @param cacheId Cache ID.
     * @param rcvdFrom Node ID cache was recived from.
     * @param deploymentId Deployment ID.
     * @param encKey Encryption key.
     * @return Group descriptor.
     */
    private CacheGroupDescriptor registerCacheGroup(
        @Nullable ExchangeActions exchActions,
        @Nullable AffinityTopologyVersion curTopVer,
        CacheConfiguration<?, ?> startedCacheCfg,
        Integer cacheId,
        UUID rcvdFrom,
        IgniteUuid deploymentId,
        @Nullable byte[] encKey) {
        if (startedCacheCfg.getGroupName() != null) {
            CacheGroupDescriptor desc = cacheGroupByName(startedCacheCfg.getGroupName());

            if (desc != null) {
                desc.onCacheAdded(startedCacheCfg.getName(), cacheId);

                return desc;
            }
        }

        int grpId = CU.cacheGroupId(startedCacheCfg.getName(), startedCacheCfg.getGroupName());

        Map<String, Integer> caches = Collections.singletonMap(startedCacheCfg.getName(), cacheId);

        boolean persistent = resolvePersistentFlag(exchActions, startedCacheCfg);

        CacheGroupDescriptor grpDesc = new CacheGroupDescriptor(
            startedCacheCfg,
            startedCacheCfg.getGroupName(),
            grpId,
            rcvdFrom,
            curTopVer != null ? curTopVer.nextMinorVersion() : null,
            deploymentId,
            caches,
            persistent,
            persistent,
            null);

        if (startedCacheCfg.isEncryptionEnabled())
            ctx.encryption().beforeCacheGroupStart(grpId, encKey);

        if (ctx.cache().context().pageStore() != null)
            ctx.cache().context().pageStore().beforeCacheGroupStart(grpDesc);

        CacheGroupDescriptor old = registeredCacheGrps.put(grpId, grpDesc);

        assert old == null : old;

        ctx.discovery().addCacheGroup(grpDesc, grpDesc.config().getNodeFilter(), startedCacheCfg.getCacheMode());

        if (exchActions != null)
            exchActions.addCacheGroupToStart(grpDesc);

        return grpDesc;
    }

    /**
     * Resolves persistent flag for new cache group descriptor.
     *
     * @param exchActions Optional exchange actions to update if new group was added.
     * @param startedCacheCfg Started cache configuration.
     */
    private boolean resolvePersistentFlag(@Nullable ExchangeActions exchActions, CacheConfiguration<?, ?> startedCacheCfg) {
        if (!ctx.clientNode()) {
            // On server, we always can determine whether cache is persistent by local storage configuration.
            return CU.isPersistentCache(startedCacheCfg, ctx.config().getDataStorageConfiguration());
        }
        else if (exchActions == null) {
            // It's either client local join event or cache is statically configured on another node.
            // No need to resolve on client - we'll anyway receive group descriptor from server with correct flag.
            return false;
        }
        else {
            // Dynamic cache start. Initiator of the start may not have known whether cache should be persistent.
            // On client, we should peek attributes of any affinity server node to get data storage configuration.
            Collection<ClusterNode> aliveSrvNodes = ctx.discovery().aliveServerNodes();

            assert !aliveSrvNodes.isEmpty() : "No alive server nodes";

            for (ClusterNode srvNode : aliveSrvNodes) {
                if (CU.affinityNode(srvNode, startedCacheCfg.getNodeFilter())) {
                    Object dsCfgBytes = srvNode.attribute(IgniteNodeAttributes.ATTR_DATA_STORAGE_CONFIG);

                    if (dsCfgBytes instanceof byte[]) {
                        try {
                            DataStorageConfiguration crdDsCfg = new JdkMarshaller().unmarshal(
                                (byte[])dsCfgBytes, U.resolveClassLoader(ctx.config()));

                            return CU.isPersistentCache(startedCacheCfg, crdDsCfg);
                        }
                        catch (IgniteCheckedException e) {
                            U.error(log, "Failed to unmarshal remote data storage configuration [remoteNode=" +
                                srvNode + ", cacheName=" + startedCacheCfg.getName() + "]", e);
                        }
                    }
                    else {
                        U.error(log, "Remote marshalled data storage configuration is absent [remoteNode=" + srvNode +
                            ", cacheName=" + startedCacheCfg.getName() + ", dsCfg=" + dsCfgBytes + "]");
                    }
                }
            }

            U.error(log, "Failed to find affinity server node with data storage configuration for starting cache " +
                "[cacheName=" + startedCacheCfg.getName() + ", aliveSrvNodes=" + aliveSrvNodes + "]");

            return false;
        }
    }

    /**
     * @param ccfg Cache configuration to start.
     * @throws IgniteCheckedException If failed.
     */
    void validateStartCacheConfiguration(CacheConfiguration ccfg) throws IgniteCheckedException {
        if (ccfg.getGroupName() != null) {
            CacheGroupDescriptor grpDesc = cacheGroupByName(ccfg.getGroupName());

            if (grpDesc != null) {
                assert ccfg.getGroupName().equals(grpDesc.groupName());

                validateCacheGroupConfiguration(grpDesc.config(), ccfg);
            }
        }
    }

    /**
     * @param cfg Existing configuration.
     * @param startCfg Cache configuration to start.
     * @throws IgniteCheckedException If validation failed.
     */
    private void validateCacheGroupConfiguration(CacheConfiguration cfg, CacheConfiguration startCfg)
        throws IgniteCheckedException {
        GridCacheAttributes attr1 = new GridCacheAttributes(cfg);
        GridCacheAttributes attr2 = new GridCacheAttributes(startCfg);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "cacheMode", "Cache mode",
            cfg.getCacheMode(), startCfg.getCacheMode(), true);

        if (cfg.getAtomicityMode() == TRANSACTIONAL_SNAPSHOT || startCfg.getAtomicityMode() == TRANSACTIONAL_SNAPSHOT)
            CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "atomicityMode", "Atomicity mode",
                attr1.atomicityMode(), attr2.atomicityMode(), true);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "affinity", "Affinity function",
            attr1.cacheAffinityClassName(), attr2.cacheAffinityClassName(), true);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "affinityPartitionsCount",
            "Affinity partitions count", attr1.affinityPartitionsCount(), attr2.affinityPartitionsCount(), true);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "nodeFilter", "Node filter",
            attr1.nodeFilterClassName(), attr2.nodeFilterClassName(), true);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "dataRegionName", "Data region",
            cfg.getDataRegionName(), startCfg.getDataRegionName(), true);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "topologyValidator", "Topology validator",
            attr1.topologyValidatorClassName(), attr2.topologyValidatorClassName(), true);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "partitionLossPolicy", "Partition Loss Policy",
            cfg.getPartitionLossPolicy(), startCfg.getPartitionLossPolicy(), true);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "rebalanceMode", "Rebalance mode",
            cfg.getRebalanceMode(), startCfg.getRebalanceMode(), true);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "rebalanceDelay", "Rebalance delay",
            cfg.getRebalanceDelay(), startCfg.getRebalanceDelay(), false);

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "rebalanceOrder", "Rebalance order",
            cfg.getRebalanceOrder(), startCfg.getRebalanceOrder(), false);

        if (cfg.getCacheMode() == PARTITIONED) {
            CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "backups", "Backups",
                cfg.getBackups(), startCfg.getBackups(), true);
        }

        CU.validateCacheGroupsAttributesMismatch(log, cfg, startCfg, "encryptionEnabled", "Encrypted",
            cfg.isEncryptionEnabled(), startCfg.isEncryptionEnabled(), true);
    }

    /**
     * @return Registered caches.
     */
    ConcurrentMap<String, DynamicCacheDescriptor> registeredCaches() {
        return registeredCaches;
    }

    /**
     * @return Registered cache templates.
     */
    ConcurrentMap<String, DynamicCacheDescriptor> registeredTemplates() {
        return registeredTemplates;
    }

    /**
     * @return Registered cache groups.
     */
    ConcurrentMap<Integer, CacheGroupDescriptor> registeredCacheGroups() {
        return registeredCacheGrps;
    }

    /**
     * Returns registered cache descriptors ordered by {@code comparator}
     * @param comparator Comparator (DIRECT, REVERSE or custom) to order cache descriptors.
     * @return Ordered by comparator cache descriptors.
     */
    private Collection<DynamicCacheDescriptor> orderedCaches(Comparator<DynamicCacheDescriptor> comparator) {
        List<DynamicCacheDescriptor> ordered = new ArrayList<>();
        ordered.addAll(registeredCaches.values());

        Collections.sort(ordered, comparator);
        return ordered;
    }

    /**
     *
     */
    public void onDisconnected() {
        cachesOnDisconnect = new CachesOnDisconnect(
            ctx.state().clusterState(),
            new HashMap<>(registeredCacheGrps),
            new HashMap<>(registeredCaches));

        registeredCacheGrps.clear();
        registeredCaches.clear();
        registeredTemplates.clear();

        clientReconnectReqs = null;
    }

    /**
     * @param active {@code True} if reconnected to active cluster.
     * @param transition {@code True} if reconnected while state transition in progress.
     * @return Information about stopped caches and cache groups.
     */
    public ClusterCachesReconnectResult onReconnected(boolean active, boolean transition) {
        assert disconnectedState();

        Set<String> stoppedCaches = new HashSet<>();
        Set<Integer> stoppedCacheGrps = new HashSet<>();

        Set<String> survivedCaches = new HashSet<>();

        if (!active) {
            joinOnTransition = transition;

            if (F.isEmpty(locCfgsForActivation)) {
                locCfgsForActivation = new HashMap<>();

                for (IgniteInternalCache cache : ctx.cache().caches()) {
                    locCfgsForActivation.put(cache.name(),
                        new T2<>((CacheConfiguration)null, cache.configuration().getNearConfiguration()));
                }
            }

            for (Map.Entry<Integer, CacheGroupDescriptor> e : cachesOnDisconnect.cacheGrps.entrySet())
                stoppedCacheGrps.add(e.getValue().groupId());

            for (Map.Entry<String, DynamicCacheDescriptor> e : cachesOnDisconnect.caches.entrySet())
                stoppedCaches.add(e.getKey());
        }
        else {
            for (Map.Entry<Integer, CacheGroupDescriptor> e : cachesOnDisconnect.cacheGrps.entrySet()) {
                CacheGroupDescriptor locDesc = e.getValue();

                CacheGroupDescriptor desc;
                boolean stopped = true;

                if (locDesc.sharedGroup()) {
                    desc = cacheGroupByName(locDesc.groupName());

                    if (desc != null && desc.deploymentId().equals(locDesc.deploymentId()))
                        stopped = false;
                }
                else {
                    desc = nonSharedCacheGroupByCacheName(locDesc.config().getName());

                    if (desc != null &&
                        (surviveReconnect(locDesc.config().getName()) || desc.deploymentId().equals(locDesc.deploymentId())))
                        stopped = false;
                }

                if (stopped)
                    stoppedCacheGrps.add(locDesc.groupId());
                else
                    assert locDesc.groupId() == desc.groupId();
            }

            for (Map.Entry<String, DynamicCacheDescriptor> e : cachesOnDisconnect.caches.entrySet()) {
                DynamicCacheDescriptor desc = e.getValue();

                String cacheName = e.getKey();

                boolean stopped;

                if (!surviveReconnect(cacheName)) {
                    DynamicCacheDescriptor newDesc = registeredCaches.get(cacheName);

                    stopped = newDesc == null || !desc.deploymentId().equals(newDesc.deploymentId());
                }
                else
                    stopped = false;

                if (stopped)
                    stoppedCaches.add(cacheName);
                else
                    survivedCaches.add(cacheName);
            }

            if (locJoinCachesCtx != null) {
                locJoinCachesCtx.removeSurvivedCaches(survivedCaches);

                if (locJoinCachesCtx.isEmpty())
                    locJoinCachesCtx = null;
            }

            if (!cachesOnDisconnect.clusterActive())
                initStartCachesForLocalJoin(false, true);
        }

        if (clientReconnectReqs != null) {
            for (Map.Entry<UUID, CacheClientReconnectDiscoveryData> e : clientReconnectReqs.entrySet())
                processClientReconnectData(e.getValue(), e.getKey());

            clientReconnectReqs = null;
        }

        cachesOnDisconnect = null;

        return new ClusterCachesReconnectResult(stoppedCacheGrps, stoppedCaches);
    }

    /**
     * @return {@code True} if client node is currently in disconnected state.
     */
    private boolean disconnectedState() {
        return cachesOnDisconnect != null;
    }

    /**
     * @param cacheName Cache name.
     * @return {@code True} if cache with given name if system cache which should always survive client node disconnect.
     */
    private boolean surviveReconnect(String cacheName) {
        return CU.isUtilityCache(cacheName);
    }

    /**
     * Holds direct comparator (first system caches) and reverse comparator (first user caches).
     * Use DIRECT comparator for ordering cache start operations.
     * Use REVERSE comparator for ordering cache stop operations.
     */
    static class CacheComparators {
        /**
         * DIRECT comparator for cache descriptors (first system caches).
         */
        static Comparator<DynamicCacheDescriptor> DIRECT = new Comparator<DynamicCacheDescriptor>() {
            @Override public int compare(DynamicCacheDescriptor o1, DynamicCacheDescriptor o2) {
                if (o1.cacheType().userCache() ^ o2.cacheType().userCache())
                    return o2.cacheType().userCache() ? -1 : 1;

                return o1.cacheId().compareTo(o2.cacheId());
            }
        };

        /**
         * REVERSE comparator for cache descriptors (first user caches).
         */
        static Comparator<DynamicCacheDescriptor> REVERSE = new Comparator<DynamicCacheDescriptor>() {
            @Override
            public int compare(DynamicCacheDescriptor o1, DynamicCacheDescriptor o2) {
                return -DIRECT.compare(o1, o2);
            }
        };
    }

    /**
     *
     */
    private static class GridData {
        /** */
        private final CacheJoinNodeDiscoveryData joinDiscoData;

        /** */
        private final CacheNodeCommonDiscoveryData gridData;

        /** */
        private final String conflictErr;

        /**
         * @param joinDiscoData Discovery data collected for local node join.
         * @param gridData Grid data.
         * @param conflictErr Cache configuration conflict error.
         */
        GridData(CacheJoinNodeDiscoveryData joinDiscoData, CacheNodeCommonDiscoveryData gridData, String conflictErr) {
            this.joinDiscoData = joinDiscoData;
            this.gridData = gridData;
            this.conflictErr = conflictErr;
        }
    }

    /**
     *
     */
    private static class CachesOnDisconnect {
        /** */
        final DiscoveryDataClusterState state;

        /** */
        final Map<Integer, CacheGroupDescriptor> cacheGrps;

        /** */
        final Map<String, DynamicCacheDescriptor> caches;

        /**
         * @param state Cluster state.
         * @param cacheGrps Cache groups.
         * @param caches Caches.
         */
        CachesOnDisconnect(DiscoveryDataClusterState state,
            Map<Integer, CacheGroupDescriptor> cacheGrps,
            Map<String, DynamicCacheDescriptor> caches) {
            this.state = state;
            this.cacheGrps = cacheGrps;
            this.caches = caches;
        }

        /**
         * @return {@code True} if cluster was in active state.
         */
        boolean clusterActive() {
            return state.active() && !state.transition();
        }
    }

    /**
     *
     */
    private static class CacheChangeProcessResult {
        /** */
        private boolean needExchange;

        /** */
        private final List<DynamicCacheDescriptor> addedDescs = new ArrayList<>();

        /** */
        private final List<IgniteCheckedException> errs = new ArrayList<>();
    }
}