/*
 * 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.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.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.SortedSet;
import java.util.StringJoiner;
import java.util.TreeMap;
import java.util.TreeSet;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import javax.cache.CacheException;
import javax.cache.expiry.EternalExpiryPolicy;
import javax.cache.expiry.ExpiryPolicy;
import javax.management.MBeanServer;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteInterruptedException;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.cache.CacheExistsException;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cache.store.CacheStoreSessionListener;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.cluster.ClusterState;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.DeploymentMode;
import org.apache.ignite.configuration.NearCacheConfiguration;
import org.apache.ignite.configuration.WarmUpConfiguration;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteFeatures;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteTransactionsEx;
import org.apache.ignite.internal.binary.BinaryContext;
import org.apache.ignite.internal.binary.BinaryMarshaller;
import org.apache.ignite.internal.binary.GridBinaryMarshaller;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.DetachedClusterNode;
import org.apache.ignite.internal.managers.communication.GridIoPolicy;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpi;
import org.apache.ignite.internal.managers.encryption.GroupKeyEncrypted;
import org.apache.ignite.internal.managers.systemview.walker.CacheGroupIoViewWalker;
import org.apache.ignite.internal.managers.systemview.walker.CachePagesListViewWalker;
import org.apache.ignite.internal.managers.systemview.walker.PartitionStateViewWalker;
import org.apache.ignite.internal.metric.IoStatisticsType;
import org.apache.ignite.internal.pagemem.store.IgnitePageStoreManager;
import org.apache.ignite.internal.pagemem.wal.IgniteWriteAheadLogManager;
import org.apache.ignite.internal.processors.GridProcessorAdapter;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.affinity.GridAffinityAssignmentCache;
import org.apache.ignite.internal.processors.cache.binary.CacheObjectBinaryProcessorImpl;
import org.apache.ignite.internal.processors.cache.datastructures.CacheDataStructuresManager;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtCacheAdapter;
import org.apache.ignite.internal.processors.cache.distributed.dht.IgniteClusterReadOnlyException;
import org.apache.ignite.internal.processors.cache.distributed.dht.atomic.GridDhtAtomicCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedCache;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.FinishPreloadingTask;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.StopCachesOnClientReconnectExchangeTask;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.PartitionDefferedDeleteQueueCleanupTask;
import org.apache.ignite.internal.processors.cache.distributed.dht.topology.PartitionsEvictManager;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearAtomicCache;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTransactionalCache;
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrManager;
import org.apache.ignite.internal.processors.cache.jta.CacheJtaManagerAdapter;
import org.apache.ignite.internal.processors.cache.mvcc.DeadlockDetectionManager;
import org.apache.ignite.internal.processors.cache.mvcc.MvccCachingManager;
import org.apache.ignite.internal.processors.cache.persistence.DataRegion;
import org.apache.ignite.internal.processors.cache.persistence.DatabaseLifecycleListener;
import org.apache.ignite.internal.processors.cache.persistence.GridCacheDatabaseSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.GridCacheOffheapManager;
import org.apache.ignite.internal.processors.cache.persistence.IgniteCacheDatabaseSharedManager;
import org.apache.ignite.internal.processors.cache.persistence.RowStore;
import org.apache.ignite.internal.processors.cache.persistence.checkpoint.CheckpointListener;
import org.apache.ignite.internal.processors.cache.persistence.file.FilePageStoreManager;
import org.apache.ignite.internal.processors.cache.persistence.freelist.FreeList;
import org.apache.ignite.internal.processors.cache.persistence.freelist.PagesList;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetastorageLifecycleListener;
import org.apache.ignite.internal.processors.cache.persistence.metastorage.ReadOnlyMetastorage;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteCacheSnapshotManager;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotDiscoveryMessage;
import org.apache.ignite.internal.processors.cache.persistence.tree.reuse.ReuseList;
import org.apache.ignite.internal.processors.cache.persistence.wal.FileWriteAheadLogManager;
import org.apache.ignite.internal.processors.cache.query.GridCacheDistributedQueryManager;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryManager;
import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryManager;
import org.apache.ignite.internal.processors.cache.store.CacheStoreManager;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTransactionsImpl;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxManager;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionManager;
import org.apache.ignite.internal.processors.cache.warmup.WarmUpStrategy;
import org.apache.ignite.internal.processors.cacheobject.IgniteCacheObjectProcessor;
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.datastructures.DataStructuresProcessor;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.processors.platform.cache.PlatformCacheManager;
import org.apache.ignite.internal.processors.plugin.CachePluginManager;
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.SchemaExchangeWorkerTask;
import org.apache.ignite.internal.processors.query.schema.SchemaNodeLeaveExchangeWorkerTask;
import org.apache.ignite.internal.processors.query.schema.message.SchemaAbstractDiscoveryMessage;
import org.apache.ignite.internal.processors.query.schema.message.SchemaProposeDiscoveryMessage;
import org.apache.ignite.internal.processors.security.IgniteSecurity;
import org.apache.ignite.internal.suggestions.GridPerformanceSuggestions;
import org.apache.ignite.internal.util.F0;
import org.apache.ignite.internal.util.IgniteCollectors;
import org.apache.ignite.internal.util.InitializationProtector;
import org.apache.ignite.internal.util.future.GridCompoundFuture;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.lang.GridPlainClosure2;
import org.apache.ignite.internal.util.lang.IgniteOutClosureX;
import org.apache.ignite.internal.util.lang.IgniteThrowableFunction;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.T3;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteClosure;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteOutClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.lifecycle.LifecycleAware;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.marshaller.MarshallerUtils;
import org.apache.ignite.mxbean.CacheGroupMetricsMXBean;
import org.apache.ignite.mxbean.IgniteMBeanAware;
import org.apache.ignite.plugin.security.SecurityException;
import org.apache.ignite.plugin.security.SecurityPermission;
import org.apache.ignite.spi.IgniteNodeValidationResult;
import org.apache.ignite.spi.discovery.DiscoveryDataBag;
import org.apache.ignite.spi.discovery.DiscoveryDataBag.GridDiscoveryData;
import org.apache.ignite.spi.discovery.DiscoveryDataBag.JoiningNodeDiscoveryData;
import org.apache.ignite.spi.systemview.view.CacheGroupIoView;
import org.apache.ignite.spi.systemview.view.CachePagesListView;
import org.apache.ignite.spi.systemview.view.PartitionStateView;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import static java.lang.String.format;
import static java.util.Arrays.asList;
import static java.util.Objects.isNull;
import static java.util.Objects.nonNull;
import static java.util.stream.Collectors.groupingBy;
import static java.util.stream.Collectors.mapping;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toSet;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_CACHE_REMOVED_ENTRIES_TTL;
import static org.apache.ignite.IgniteSystemProperties.IGNITE_SKIP_CONFIGURATION_CONSISTENCY_CHECK;
import static org.apache.ignite.IgniteSystemProperties.getBoolean;
import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL_SNAPSHOT;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
import static org.apache.ignite.cache.CacheMode.REPLICATED;
import static org.apache.ignite.cache.CacheRebalanceMode.SYNC;
import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static org.apache.ignite.configuration.DeploymentMode.CONTINUOUS;
import static org.apache.ignite.configuration.DeploymentMode.SHARED;
import static org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType.CACHE_PROC;
import static org.apache.ignite.internal.IgniteComponentType.JTA;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isNearEnabled;
import static org.apache.ignite.internal.processors.cache.GridCacheUtils.isPersistentCache;
import static org.apache.ignite.internal.processors.cache.ValidationOnNodeJoinUtils.validateHashIdResolvers;
import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition.DFLT_CACHE_REMOVE_ENTRIES_TTL;
import static org.apache.ignite.internal.processors.metric.impl.MetricUtils.metricName;
import static org.apache.ignite.internal.processors.security.SecurityUtils.remoteSecurityContext;
import static org.apache.ignite.internal.util.IgniteUtils.doInParallel;

/**
 * The main responsibility of this processor is orchestrating of starting, stopping caches and everything related
 * to that activity like validation of cache configuration, restoring configuration from persistence, etc.
 */
@SuppressWarnings({"unchecked", "TypeMayBeWeakened", "deprecation"})
public class GridCacheProcessor extends GridProcessorAdapter {
    /** */
    public static final String CLUSTER_READ_ONLY_MODE_ERROR_MSG_FORMAT =
        "Failed to perform %s operation (cluster is in read-only mode) [cacheGrp=%s, cache=%s]";

    /** */
    private static final String CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT =
        "Cannot start/stop cache within lock or transaction [cacheNames=%s, operation=%s]";

    /** System view name for page lists. */
    public static final String CACHE_GRP_PAGE_LIST_VIEW = "cacheGroupPageLists";

    /** System view description for page lists. */
    public static final String CACHE_GRP_PAGE_LIST_VIEW_DESC = "Cache group page lists";

    /** System view name for partition states. */
    public static final String PART_STATES_VIEW = "partitionStates";

    /** System view description for partition states. */
    public static final String PART_STATES_VIEW_DESC = "Distribution of cache group partitions across cluster nodes";

    /** System view name for cache group IO. */
    public static final String CACHE_GRP_IO_VIEW = metricName("local", "cache", "groups", "io");

    /** System view description for cache group IO. */
    public static final String CACHE_GRP_IO_VIEW_DESC = "Local node IO statistics for cache groups";

    /** @see IgniteSystemProperties#IGNITE_ALLOW_START_CACHES_IN_PARALLEL */
    public static final boolean DFLT_ALLOW_START_CACHES_IN_PARALLEL = true;

    /** Enables start caches in parallel. */
    private final boolean IGNITE_ALLOW_START_CACHES_IN_PARALLEL =
        IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_ALLOW_START_CACHES_IN_PARALLEL,
            DFLT_ALLOW_START_CACHES_IN_PARALLEL);

    /** */
    private final boolean keepStaticCacheConfiguration = IgniteSystemProperties.getBoolean(
        IgniteSystemProperties.IGNITE_KEEP_STATIC_CACHE_CONFIGURATION);

    /** MBean group for cache group metrics */
    private static final String CACHE_GRP_METRICS_MBEAN_GRP = "Cache groups";

    /**
     * Initial timeout (in milliseconds) for output the progress of restoring partitions status.
     * After the first output, the next ones will be output after value/5.
     * It is recommended to change this property only in tests.
     */
    static long TIMEOUT_OUTPUT_RESTORE_PARTITION_STATE_PROGRESS = TimeUnit.MINUTES.toMillis(5);

    /** Shared cache context. */
    private GridCacheSharedContext<?, ?> sharedCtx;

    /** */
    private final ConcurrentMap<Integer, CacheGroupContext> cacheGrps = new ConcurrentHashMap<>();

    /** */
    private final Map<String, GridCacheAdapter<?, ?>> caches;

    /** Caches stopped from onKernalStop callback. */
    private final Map<String, GridCacheAdapter> stoppedCaches = new ConcurrentHashMap<>();

    /** Map of proxies. */
    private final ConcurrentHashMap<String, IgniteCacheProxyImpl<?, ?>> jCacheProxies;

    /** Transaction interface implementation. */
    private IgniteTransactionsImpl transactions;

    /** Pending cache operations. */
    private ConcurrentMap<UUID, IgniteInternalFuture> pendingFuts = new ConcurrentHashMap<>();

    /** Template configuration add futures. */
    private ConcurrentMap<String, IgniteInternalFuture> pendingTemplateFuts = new ConcurrentHashMap<>();

    /** Enable/disable cache statistics futures. */
    private ConcurrentMap<UUID, EnableStatisticsFuture> manageStatisticsFuts = new ConcurrentHashMap<>();

    /** */
    private ClusterCachesInfo cachesInfo;

    /** */
    private GridLocalConfigManager locCfgMgr;

    /** */
    private IdentityHashMap<CacheStore, ThreadLocal> sesHolders = new IdentityHashMap<>();

    /** Must use JDK marsh since it is used by discovery to fire custom events. */
    private final Marshaller marsh;

    /** Count down latch for caches. */
    private final CountDownLatch cacheStartedLatch = new CountDownLatch(1);

    /** Internal cache names. */
    private final Set<String> internalCaches;

    /** Protector of initialization of specific value. */
    private final InitializationProtector initializationProtector = new InitializationProtector();

    /** Cache recovery lifecycle state and actions. */
    private final CacheRecoveryLifecycle recovery = new CacheRecoveryLifecycle();

    /** Cache configuration splitter. */
    private CacheConfigurationSplitter splitter;

    /** Cache configuration enricher. */
    private CacheConfigurationEnricher enricher;

    /** */
    public static final String EXPRITY_POLICY_MSG = "expiryPolicy=[type=%s, isEagerTtl=%s]";

    /**
     * @param ctx Kernal context.
     */
    public GridCacheProcessor(GridKernalContext ctx) {
        super(ctx);

        caches = new ConcurrentHashMap<>();
        jCacheProxies = new ConcurrentHashMap<>();
        internalCaches = new HashSet<>();

        marsh = MarshallerUtils.jdkMarshaller(ctx.igniteInstanceName());
        splitter = new CacheConfigurationSplitterImpl(ctx, marsh);
        enricher = new CacheConfigurationEnricher(ctx, marsh, U.resolveClassLoader(ctx.config()));
    }

    /**
     * @param cfg Configuration to check for possible performance issues.
     * @param hasStore {@code True} if store is configured.
     */
    private void suggestOptimizations(CacheConfiguration cfg, boolean hasStore) {
        GridPerformanceSuggestions perf = ctx.performance();

        String msg = "Disable eviction policy (remove from configuration)";

        if (cfg.getEvictionPolicyFactory() != null || cfg.getEvictionPolicy() != null)
            perf.add(msg, false);
        else
            perf.add(msg, true);

        if (cfg.getCacheMode() == PARTITIONED)
            perf.add("Disable near cache (set 'nearConfiguration' to null)", cfg.getNearConfiguration() == null);

        // Suppress warning if at least one ATOMIC cache found.
        perf.add("Enable ATOMIC mode if not using transactions (set 'atomicityMode' to ATOMIC)",
            cfg.getAtomicityMode() == ATOMIC);

        // Suppress warning if at least one non-FULL_SYNC mode found.
        perf.add("Disable fully synchronous writes (set 'writeSynchronizationMode' to PRIMARY_SYNC or FULL_ASYNC)",
            cfg.getWriteSynchronizationMode() != FULL_SYNC);

        if (hasStore && cfg.isWriteThrough())
            perf.add("Enable write-behind to persistent store (set 'writeBehindEnabled' to true)",
                cfg.isWriteBehindEnabled());
    }

    /**
     * Create exchange worker task for custom discovery message.
     *
     * @param msg Custom discovery message.
     * @return Task or {@code null} if message doesn't require any special processing.
     */
    public CachePartitionExchangeWorkerTask exchangeTaskForCustomDiscoveryMessage(DiscoveryCustomMessage msg) {
        if (msg instanceof SchemaAbstractDiscoveryMessage) {
            SchemaAbstractDiscoveryMessage msg0 = (SchemaAbstractDiscoveryMessage)msg;

            if (msg0.exchange())
                return new SchemaExchangeWorkerTask(remoteSecurityContext(ctx), msg0);
        }
        else if (msg instanceof ClientCacheChangeDummyDiscoveryMessage) {
            ClientCacheChangeDummyDiscoveryMessage msg0 = (ClientCacheChangeDummyDiscoveryMessage)msg;

            return msg0;
        }
        else if (msg instanceof CacheStatisticsModeChangeMessage) {
            CacheStatisticsModeChangeMessage msg0 = (CacheStatisticsModeChangeMessage)msg;

            if (msg0.initial())
                return new CacheStatisticsModeChangeTask(remoteSecurityContext(ctx), msg0);
        }

        return null;
    }

    /**
     * Process custom exchange task.
     *
     * @param task Task.
     */
    void processCustomExchangeTask(CachePartitionExchangeWorkerTask task) {
        if (task instanceof SchemaExchangeWorkerTask) {
            SchemaAbstractDiscoveryMessage msg = ((SchemaExchangeWorkerTask)task).message();

            if (msg instanceof SchemaProposeDiscoveryMessage) {
                SchemaProposeDiscoveryMessage msg0 = (SchemaProposeDiscoveryMessage)msg;

                ctx.query().onSchemaPropose(msg0);
            }
            else
                U.warn(log, "Unsupported schema discovery message: " + msg);
        }
        else if (task instanceof SchemaNodeLeaveExchangeWorkerTask) {
            SchemaNodeLeaveExchangeWorkerTask task0 = (SchemaNodeLeaveExchangeWorkerTask)task;

            ctx.query().onNodeLeave(task0.node());
        }
        else if (task instanceof ClientCacheChangeDummyDiscoveryMessage) {
            ClientCacheChangeDummyDiscoveryMessage task0 = (ClientCacheChangeDummyDiscoveryMessage)task;

            sharedCtx.affinity().processClientCachesRequests(task0);
        }
        else if (task instanceof ClientCacheUpdateTimeout) {
            ClientCacheUpdateTimeout task0 = (ClientCacheUpdateTimeout)task;

            sharedCtx.affinity().sendClientCacheChangesMessage(task0);
        }
        else if (task instanceof CacheStatisticsModeChangeTask) {
            CacheStatisticsModeChangeTask task0 = (CacheStatisticsModeChangeTask)task;

            processStatisticsModeChange(task0.message());
        }
        else if (task instanceof TxTimeoutOnPartitionMapExchangeChangeTask) {
            TxTimeoutOnPartitionMapExchangeChangeTask task0 = (TxTimeoutOnPartitionMapExchangeChangeTask)task;

            sharedCtx.tm().processTxTimeoutOnPartitionMapExchangeChange(task0.message());
        }
        else if (task instanceof StopCachesOnClientReconnectExchangeTask) {
            StopCachesOnClientReconnectExchangeTask task0 = (StopCachesOnClientReconnectExchangeTask)task;

            stopCachesOnClientReconnect(task0.stoppedCaches());

            task0.onDone();
        }
        else if (task instanceof WalStateNodeLeaveExchangeTask) {
            WalStateNodeLeaveExchangeTask task0 = (WalStateNodeLeaveExchangeTask)task;

            sharedCtx.walState().onNodeLeft(task0.node().id());
        }
        else if (task instanceof FinishPreloadingTask) {
            FinishPreloadingTask task0 = (FinishPreloadingTask)task;

            CacheGroupContext grp = cacheGroup(task0.groupId());

            if (grp != null)
                grp.preloader().finishPreloading(task0.topologyVersion(), task0.rebalanceId());
        }
        else
            U.warn(log, "Unsupported custom exchange task: " + task);
    }

    /**
     * @param ctx Context.
     * @return DHT managers.
     */
    private List<GridCacheManager> dhtManagers(GridCacheContext ctx) {
        return F.asList(ctx.store(), ctx.events(), ctx.evicts(), ctx.queries(), ctx.continuousQueries(),
            ctx.dr());
    }

    /**
     * @param ctx Context.
     * @return Managers present in both, DHT and Near caches.
     */
    @SuppressWarnings("IfMayBeConditional")
    private Collection<GridCacheManager> dhtExcludes(GridCacheContext ctx) {
        if (!isNearEnabled(ctx))
            return Collections.emptyList();
        else
            return F.asList(ctx.queries(), ctx.continuousQueries(), ctx.store());
    }

    /**
     * @param cfg Configuration.
     * @param objs Extra components.
     * @throws IgniteCheckedException If failed to inject.
     */
    private void prepare(CacheConfiguration cfg, Collection<Object> objs) throws IgniteCheckedException {
        prepare(cfg, cfg.getAffinity(), false);
        prepare(cfg, cfg.getAffinityMapper(), false);
        prepare(cfg, cfg.getEvictionFilter(), false);
        prepare(cfg, cfg.getInterceptor(), false);

        for (Object obj : objs)
            prepare(cfg, obj, false);
    }

    /**
     * @param cfg Cache configuration.
     * @param rsrc Resource.
     * @param near Near flag.
     * @throws IgniteCheckedException If failed.
     */
    private void prepare(CacheConfiguration cfg, @Nullable Object rsrc, boolean near) throws IgniteCheckedException {
        if (rsrc != null) {
            ctx.resource().injectGeneric(rsrc);

            ctx.resource().injectCacheName(rsrc, cfg.getName());

            registerMbean(rsrc, cfg.getName(), near);
        }
    }

    /**
     * @param cctx Cache context.
     */
    private void cleanup(GridCacheContext cctx) {
        CacheConfiguration cfg = cctx.config();

        cleanup(cfg, cfg.getAffinity(), false);
        cleanup(cfg, cfg.getAffinityMapper(), false);
        cleanup(cfg, cfg.getEvictionFilter(), false);
        cleanup(cfg, cfg.getInterceptor(), false);
        cleanup(cfg, cctx.store().configuredStore(), false);

        if (!CU.isUtilityCache(cfg.getName()) && !CU.isSystemCache(cfg.getName())) {
            unregisterMbean(cctx.cache().localMxBean(), cfg.getName(), false);
            unregisterMbean(cctx.cache().clusterMxBean(), cfg.getName(), false);
        }

        cctx.cleanup();
    }

    /**
     * @param grp Cache group.
     * @param destroy Group destroy flag.
     */
    private void cleanup(CacheGroupContext grp, boolean destroy) {
        CacheConfiguration cfg = grp.config();

        for (Object obj : grp.configuredUserObjects())
            cleanup(cfg, obj, false);

        if (!grp.systemCache() && !U.IGNITE_MBEANS_DISABLED) {
            try {
                ctx.config().getMBeanServer().unregisterMBean(U.makeMBeanName(ctx.igniteInstanceName(),
                    CACHE_GRP_METRICS_MBEAN_GRP, grp.cacheOrGroupName()));
            }
            catch (Throwable e) {
                U.error(log, "Failed to unregister MBean for cache group: " + grp.name(), e);
            }
        }

        grp.metrics().remove(destroy);

        grp.removeIOStatistic(destroy);

        sharedCtx.evict().cleanupRemovedGroup(grp.groupId());
    }

    /**
     * @param cfg Cache configuration.
     * @param rsrc Resource.
     * @param near Near flag.
     */
    private void cleanup(CacheConfiguration cfg, @Nullable Object rsrc, boolean near) {
        if (rsrc != null) {
            unregisterMbean(rsrc, cfg.getName(), near);

            try {
                ctx.resource().cleanupGeneric(rsrc);
            }
            catch (IgniteCheckedException e) {
                U.error(log, "Failed to cleanup resource: " + rsrc, e);
            }
        }
    }

    /** {@inheritDoc} */
    @SuppressWarnings({"unchecked"})
    @Override public void start() throws IgniteCheckedException {
        ctx.internalSubscriptionProcessor().registerMetastorageListener(recovery);
        ctx.internalSubscriptionProcessor().registerDatabaseListener(recovery);

        cachesInfo = new ClusterCachesInfo(ctx);

        DeploymentMode depMode = ctx.config().getDeploymentMode();

        if (!F.isEmpty(ctx.config().getCacheConfiguration())) {
            if (depMode != CONTINUOUS && depMode != SHARED)
                U.warn(log, "Deployment mode for cache is not CONTINUOUS or SHARED " +
                    "(it is recommended that you change deployment mode and restart): " + depMode);
        }

        Collection<CacheStoreSessionListener> sessionListeners =
            CU.startStoreSessionListeners(ctx, ctx.config().getCacheStoreSessionListenerFactories());

        sharedCtx = createSharedContext(ctx, sessionListeners);

        locCfgMgr = new GridLocalConfigManager(this, ctx);

        transactions = new IgniteTransactionsImpl(sharedCtx, null, false);

        // Start shared managers.
        for (GridCacheSharedManager mgr : sharedCtx.managers())
            mgr.start(sharedCtx);

        if (!ctx.isDaemon() && (!CU.isPersistenceEnabled(ctx.config())) || ctx.config().isClientMode()) {
            CacheJoinNodeDiscoveryData data = locCfgMgr.restoreCacheConfigurations();

            if (data != null)
                cachesInfo.onStart(data);
        }

        if (log.isDebugEnabled())
            log.debug("Started cache processor.");

        ctx.state().cacheProcessorStarted();

        ctx.systemView().registerFiltrableView(
            CACHE_GRP_PAGE_LIST_VIEW,
            CACHE_GRP_PAGE_LIST_VIEW_DESC,
            new CachePagesListViewWalker(),
            this::pagesListViewSupplier,
            Function.identity()
        );

        ctx.systemView().registerFiltrableView(
            PART_STATES_VIEW,
            PART_STATES_VIEW_DESC,
            new PartitionStateViewWalker(),
            this::partStatesViewSupplier,
            Function.identity()
        );

        ctx.systemView().registerView(
            CACHE_GRP_IO_VIEW,
            CACHE_GRP_IO_VIEW_DESC,
            new CacheGroupIoViewWalker(),
            () -> F.view(cacheGrps.values(), grp -> !grp.systemCache()),
            grpCtx -> {
                MetricRegistry mreg = ctx.metric().registry(metricName(IoStatisticsType.CACHE_GROUP.metricGroupName(),
                    grpCtx.cacheOrGroupName()));

                return new CacheGroupIoView(grpCtx, mreg);
            }
        );
    }

    /**
     * @param cfg Initializes cache configuration with proper defaults.
     * @param cacheObjCtx Cache object context.
     * @throws IgniteCheckedException If configuration is not valid.
     */
    void initialize(CacheConfiguration cfg, CacheObjectContext cacheObjCtx) throws IgniteCheckedException {
        CU.initializeConfigDefaults(log, cfg, cacheObjCtx);

        ctx.coordinators().preProcessCacheConfiguration(cfg);
    }

    /**
     * @param grpId Group ID.
     * @return Cache group.
     */
    @Nullable public CacheGroupContext cacheGroup(int grpId) {
        return cacheGrps.get(grpId);
    }

    /**
     * @return Cache groups.
     */
    public Collection<CacheGroupContext> cacheGroups() {
        return cacheGrps.values();
    }

    /** {@inheritDoc} */
    @Override public void onKernalStart(boolean active) throws IgniteCheckedException {
        if (ctx.isDaemon())
            return;

        AffinityTopologyVersion joinVer;

        try {
            boolean checkConsistency = !getBoolean(IGNITE_SKIP_CONFIGURATION_CONSISTENCY_CHECK);

            if (checkConsistency)
                ValidationOnNodeJoinUtils.checkConsistency(ctx, log);

            cachesInfo.onKernalStart(checkConsistency);

            sharedCtx.walState().onKernalStart();

            ctx.query().onCacheKernalStart();

            joinVer = sharedCtx.exchange().onKernalStart(active, false);
        }
        finally {
            cacheStartedLatch.countDown();
        }

        if (!ctx.clientNode())
            sharedCtx.time().addTimeoutObject(new PartitionDefferedDeleteQueueCleanupTask(
                sharedCtx, Long.getLong(IGNITE_CACHE_REMOVED_ENTRIES_TTL, DFLT_CACHE_REMOVE_ENTRIES_TTL)));

        // Notify shared managers.
        for (GridCacheSharedManager mgr : sharedCtx.managers())
            mgr.onKernalStart(active);

        // Escape if cluster inactive.
        if (!active)
            return;

        awaitRebalance(joinVer).get();
    }

    /**
     * Await for local join finish and caches start.
     *
     * @throws IgniteCheckedException If failed to wait.
     */
    public void awaitStarted() throws IgniteCheckedException {
        U.await(cacheStartedLatch);
    }

    /**
     * Await rebalance for caches with SYNC rebalance mode started on local join exchange.
     *
     * @param joinVer Topology version of local join.
     * @return Future indicates that rebalance for SYNC rebalance mode caches has completed.
     */
    private GridCompoundFuture<?, ?> awaitRebalance(AffinityTopologyVersion joinVer) {
        return internalCaches().stream()
            .map(GridCacheAdapter::context)
            .filter(GridCacheContext::affinityNode) // Only affinity caches.
            .filter(ctx -> ctx.config().getRebalanceMode() == SYNC) // Only caches with SYNC mode.
            .filter(ctx -> ctx.startTopologyVersion().equals(joinVer)) // Only caches started on local join.
            .filter(ctx -> ctx.config().getCacheMode() == REPLICATED // Caches without manual rebalance.
                || ctx.config().getCacheMode() == PARTITIONED && ctx.config().getRebalanceDelay() >= 0)
            .map(ctx -> ctx.preloader().syncFuture())
            .collect(IgniteCollectors.toCompoundFuture());
    }

    /** {@inheritDoc} */
    @Override public void stop(boolean cancel) throws IgniteCheckedException {
        stopCaches(cancel);

        List<? extends GridCacheSharedManager<?, ?>> mgrs = sharedCtx.managers();

        for (ListIterator<? extends GridCacheSharedManager<?, ?>> it = mgrs.listIterator(mgrs.size()); it.hasPrevious(); ) {
            GridCacheSharedManager<?, ?> mgr = it.previous();

            mgr.stop(cancel);
        }

        CU.stopStoreSessionListeners(ctx, sharedCtx.storeSessionListeners());

        sharedCtx.cleanup();

        if (log.isDebugEnabled())
            log.debug("Stopped cache processor.");
    }

    /**
     * @param cancel Cancel.
     */
    public void stopCaches(boolean cancel) {
        for (String cacheName : locCfgMgr.stopSequence()) {
            GridCacheAdapter<?, ?> cache = stoppedCaches.remove(cacheName);

            if (cache != null)
                stopCache(cache, cancel, false, false);
        }

        for (GridCacheAdapter<?, ?> cache : stoppedCaches.values()) {
            if (cache == stoppedCaches.remove(cache.name()))
                stopCache(cache, cancel, false, false);
        }

        for (CacheGroupContext grp : cacheGrps.values())
            stopCacheGroup(grp.groupId(), false);
    }

    /**
     * Blocks all available gateways
     */
    public void blockGateways() {
        for (IgniteCacheProxyImpl<?, ?> proxy : jCacheProxies.values())
            proxy.context0().gate().onStopped();
    }

    /**
     * Blocks (stops) cache gateway for caches according to given {@code cacheGroupIds}.
     *
     * @param cacheGrpIds Cache group ids for which cache gateway should be stopped.
     * @return Caches for which cache gateway is blocked (stopped).
     */
    public List<GridCacheAdapter> blockGateways(Collection<Integer> cacheGrpIds) {
        List<GridCacheAdapter> affectedCaches = internalCaches().stream()
            .filter(cache -> cacheGrpIds.contains(cache.context().groupId()))
            .collect(toList());

        affectedCaches.forEach(cache -> {
            // Add proxy if it's not initialized.
            addjCacheProxy(cache.context().name(), new IgniteCacheProxyImpl(cache.context(), cache, false));

            // Stop proxy.
            blockGateway(cache.context().name(), true, false);
        });

        return affectedCaches;
    }

    /** {@inheritDoc} */
    @Override public void onKernalStop(boolean cancel) {
        cacheStartedLatch.countDown();

        GridCachePartitionExchangeManager<Object, Object> exch = context().exchange();

        // Stop exchange manager first so that we call onKernalStop on all caches.
        // No new caches should be added after this point.
        exch.onKernalStop(cancel);

        sharedCtx.mvcc().onStop();

        for (CacheGroupContext grp : cacheGrps.values())
            grp.onKernalStop();

        onKernalStopCaches(cancel);

        cancelFutures();

        List<? extends GridCacheSharedManager<?, ?>> sharedMgrs = sharedCtx.managers();

        for (ListIterator<? extends GridCacheSharedManager<?, ?>> it = sharedMgrs.listIterator(sharedMgrs.size());
            it.hasPrevious(); ) {
            GridCacheSharedManager<?, ?> mgr = it.previous();

            if (mgr != exch)
                mgr.onKernalStop(cancel);
        }
    }

    /**
     * @param cancel Cancel.
     */
    public void onKernalStopCaches(boolean cancel) {
        IgniteCheckedException affErr =
            new IgniteCheckedException("Failed to wait for topology update, node is stopping.");

        for (CacheGroupContext grp : cacheGrps.values()) {
            GridAffinityAssignmentCache aff = grp.affinity();

            aff.cancelFutures(affErr);
        }

        for (String cacheName : locCfgMgr.stopSequence()) {
            GridCacheAdapter<?, ?> cache = caches.remove(cacheName);

            if (cache != null) {
                stoppedCaches.put(cacheName, cache);

                onKernalStop(cache, cancel);
            }
        }

        for (Map.Entry<String, GridCacheAdapter<?, ?>> entry : caches.entrySet()) {
            GridCacheAdapter<?, ?> cache = entry.getValue();

            if (cache == caches.remove(entry.getKey())) {
                stoppedCaches.put(entry.getKey(), cache);

                onKernalStop(entry.getValue(), cancel);
            }
        }
    }

    /** {@inheritDoc} */
    @Override public void onDisconnected(IgniteFuture<?> reconnectFut) throws IgniteCheckedException {
        IgniteClientDisconnectedCheckedException err = new IgniteClientDisconnectedCheckedException(
            ctx.cluster().clientReconnectFuture(),
            "Failed to execute dynamic cache change request, client node disconnected.");

        for (IgniteInternalFuture fut : pendingFuts.values())
            ((GridFutureAdapter)fut).onDone(err);

        for (IgniteInternalFuture fut : pendingTemplateFuts.values())
            ((GridFutureAdapter)fut).onDone(err);

        for (EnableStatisticsFuture fut : manageStatisticsFuts.values())
            fut.onDone(err);

        for (CacheGroupContext grp : cacheGrps.values())
            grp.onDisconnected(reconnectFut);

        for (GridCacheAdapter cache : caches.values()) {
            GridCacheContext cctx = cache.context();

            cctx.gate().onDisconnected(reconnectFut);

            List<GridCacheManager> mgrs = cache.context().managers();

            for (ListIterator<GridCacheManager> it = mgrs.listIterator(mgrs.size()); it.hasPrevious(); ) {
                GridCacheManager mgr = it.previous();

                mgr.onDisconnected(reconnectFut);
            }
        }

        sharedCtx.onDisconnected(reconnectFut);

        cachesInfo.onDisconnected();
    }

    /**
     * @param cctx Cache context.
     * @param stoppedCaches List where stopped cache should be added.
     */
    private void stopCacheOnReconnect(GridCacheContext cctx, List<GridCacheAdapter> stoppedCaches) {
        cctx.gate().reconnected(true);

        sharedCtx.removeCacheContext(cctx);

        caches.remove(cctx.name());

        completeProxyInitialize(cctx.name());

        jCacheProxies.remove(cctx.name());

        stoppedCaches.add(cctx.cache());
    }

    /** {@inheritDoc} */
    @Override public IgniteInternalFuture<?> onReconnected(boolean clusterRestarted) throws IgniteCheckedException {
        List<GridCacheAdapter> reconnected = new ArrayList<>(caches.size());

        DiscoveryDataClusterState state = ctx.state().clusterState();

        boolean active = state.active() && !state.transition();

        ClusterCachesReconnectResult reconnectRes = cachesInfo.onReconnected(active, state.transition());

        final List<GridCacheAdapter> stoppedCaches = new ArrayList<>();

        // Close SQL caches which were not started by client.
        for (String cacheToStop: reconnectRes.stoppedCaches()) {
            if (!caches.keySet().contains(cacheToStop))
                ctx.query().onCacheStop(cacheToStop);
        }

        for (final GridCacheAdapter cache : caches.values()) {
            boolean stopped = reconnectRes.stoppedCacheGroups().contains(cache.context().groupId())
                || reconnectRes.stoppedCaches().contains(cache.name());

            if (stopped)
                stopCacheOnReconnect(cache.context(), stoppedCaches);
            else {
                cache.onReconnected();

                reconnected.add(cache);

                if (cache.context().userCache()) {
                    DynamicCacheDescriptor desc = cacheDescriptor(cache.name());

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

                    if (!QueryUtils.isEnabled(cache.context().config())
                            && QueryUtils.isEnabled(desc.cacheConfiguration())) {
                        CacheConfiguration newCfg = desc.cacheConfiguration();

                        cache.context().onSchemaAddQueryEntity(newCfg.getQueryEntities(), newCfg.getSqlSchema(),
                                newCfg.isSqlEscapeAll(), newCfg.getQueryParallelism());
                    }

                    boolean rmvIdx = !cache.context().group().persistenceEnabled();

                    // Re-create cache structures inside indexing in order to apply recent schema changes.
                    GridCacheContextInfo cacheInfo = new GridCacheContextInfo(cache.context(), false);

                    ctx.query().onCacheStop0(cacheInfo, rmvIdx, rmvIdx);
                    ctx.query().onCacheStart0(cacheInfo, desc.schema(), desc.sql());
                }
            }
        }

        final Set<Integer> stoppedGrps = reconnectRes.stoppedCacheGroups();

        for (CacheGroupContext grp : cacheGrps.values()) {
            if (stoppedGrps.contains(grp.groupId()))
                cacheGrps.remove(grp.groupId());
            else
                grp.onReconnected();
        }

        sharedCtx.onReconnected(active);

        for (GridCacheAdapter cache : reconnected)
            cache.context().gate().reconnected(false);

        if (!stoppedCaches.isEmpty())
            return sharedCtx.exchange().deferStopCachesOnClientReconnect(stoppedCaches);

        return null;
    }

    /**
     * @param cache Cache to stop.
     * @param cancel Cancel flag.
     * @param callDestroy Cache destroy flag. This flag is passed to the {@link GridCacheManager#stop} method.
     * @param clearCache Cache data clear flag. Setting to {@code true} will remove all cache data.
     */
    private void stopCache(GridCacheAdapter<?, ?> cache, boolean cancel, boolean callDestroy, boolean clearCache) {
        stopCache(cache, cancel, callDestroy, clearCache, true);
    }

    /**
     * @param cache Cache to stop.
     * @param cancel Cancel flag.
     * @param callDestroy Cache destroy flag. This flag is passed to the {@link GridCacheManager#stop} method.
     * @param clearCache Cache data clear flag. Setting to {@code true} will remove all cache data.
     * @param clearDbObjects If {@code false} DB objects don't removed (used for cache.close() on client node).
     */
    @SuppressWarnings({"unchecked"})
    private void stopCache(GridCacheAdapter<?, ?> cache, boolean cancel, boolean callDestroy, boolean clearCache, boolean clearDbObjects) {
        GridCacheContext ctx = cache.context();

        try {
            if (!cache.isNear() && ctx.shared().wal() != null) {
                try {
                    ctx.shared().wal().flush(null, false);
                }
                catch (IgniteCheckedException e) {
                    U.error(log, "Failed to flush write-ahead log on cache stop " +
                        "[cache=" + ctx.name() + "]", e);
                }
            }

            sharedCtx.removeCacheContext(ctx);

            cache.stop();

            cache.removeMetrics(callDestroy);

            GridCacheContextInfo cacheInfo = new GridCacheContextInfo(ctx, false);

            if (clearDbObjects) {
                boolean rmvIdx = !cache.context().group().persistenceEnabled() || callDestroy;
                boolean clearIdx = !cache.context().group().persistenceEnabled() || clearCache;
                ctx.kernalContext().query().onCacheStop(cacheInfo, rmvIdx, clearIdx);
            }
            else
                ctx.kernalContext().query().getIndexing().closeCacheOnClient(ctx.name());

            if (isNearEnabled(ctx)) {
                GridDhtCacheAdapter dht = ctx.near().dht();

                // Check whether dht cache has been started.
                if (dht != null) {
                    dht.stop();

                    dht.removeMetrics(callDestroy);

                    GridCacheContext<?, ?> dhtCtx = dht.context();

                    List<GridCacheManager> dhtMgrs = dhtManagers(dhtCtx);

                    for (ListIterator<GridCacheManager> it = dhtMgrs.listIterator(dhtMgrs.size()); it.hasPrevious(); ) {
                        GridCacheManager mgr = it.previous();

                        mgr.stop(cancel, callDestroy);
                    }
                }
            }

            List<GridCacheManager> mgrs = ctx.managers();

            Collection<GridCacheManager> excludes = dhtExcludes(ctx);

            // Reverse order.
            for (ListIterator<GridCacheManager> it = mgrs.listIterator(mgrs.size()); it.hasPrevious(); ) {
                GridCacheManager mgr = it.previous();

                if (!excludes.contains(mgr))
                    mgr.stop(cancel, callDestroy);
            }

            ctx.kernalContext().continuous().onCacheStop(ctx);

            ctx.kernalContext().cache().context().snapshot().onCacheStop(ctx, callDestroy);

            ctx.kernalContext().coordinators().onCacheStop(ctx);

            ctx.group().stopCache(ctx, clearCache);

            U.stopLifecycleAware(log, lifecycleAwares(ctx.group(), cache.configuration(), ctx.store().configuredStore()));

            if (callDestroy && CU.storeCacheConfig(sharedCtx, ctx.config())) {
                try {
                    locCfgMgr.removeCacheData(new StoredCacheData(ctx.config()));
                }
                catch (IgniteCheckedException e) {
                    U.error(log, "Failed to delete cache configuration data while destroying cache" +
                        "[cache=" + ctx.name() + "]", e);
                }
            }

            if (log.isInfoEnabled()) {
                if (ctx.group().sharedGroup())
                    log.info("Stopped cache [cacheName=" + cache.name() + ", group=" + ctx.group().name() + ']');
                else
                    log.info("Stopped cache [cacheName=" + cache.name() + ']');
            }
        }
        finally {
            cleanup(ctx);
        }
    }

    /**
     * @param cache Cache.
     * @throws IgniteCheckedException If failed.
     */
    private void onKernalStart(GridCacheAdapter<?, ?> cache) throws IgniteCheckedException {
        GridCacheContext<?, ?> ctx = cache.context();

        // Start DHT cache as well.
        if (isNearEnabled(ctx)) {
            GridDhtCacheAdapter dht = ctx.near().dht();

            GridCacheContext<?, ?> dhtCtx = dht.context();

            for (GridCacheManager mgr : dhtManagers(dhtCtx))
                mgr.onKernalStart();

            dht.onKernalStart();

            if (log.isDebugEnabled())
                log.debug("Executed onKernalStart() callback for DHT cache: " + dht.name());
        }

        for (GridCacheManager mgr : F.view(ctx.managers(), F0.notContains(dhtExcludes(ctx))))
            mgr.onKernalStart();

        cache.onKernalStart();

        if (ctx.events().isRecordable(EventType.EVT_CACHE_STARTED))
            ctx.events().addEvent(EventType.EVT_CACHE_STARTED);

        if (log.isDebugEnabled())
            log.debug("Executed onKernalStart() callback for cache [name=" + cache.name() + ", mode=" +
                cache.configuration().getCacheMode() + ']');
    }

    /**
     * @param cache Cache to stop.
     * @param cancel Cancel flag.
     */
    @SuppressWarnings("unchecked")
    private void onKernalStop(GridCacheAdapter<?, ?> cache, boolean cancel) {
        GridCacheContext ctx = cache.context();

        if (isNearEnabled(ctx)) {
            GridDhtCacheAdapter dht = ctx.near().dht();

            if (dht != null) {
                GridCacheContext<?, ?> dhtCtx = dht.context();

                for (GridCacheManager mgr : dhtManagers(dhtCtx))
                    mgr.onKernalStop(cancel);

                dht.onKernalStop();
            }
        }

        List<GridCacheManager> mgrs = ctx.managers();

        Collection<GridCacheManager> excludes = dhtExcludes(ctx);

        // Reverse order.
        for (ListIterator<GridCacheManager> it = mgrs.listIterator(mgrs.size()); it.hasPrevious(); ) {
            GridCacheManager mgr = it.previous();

            if (!excludes.contains(mgr))
                mgr.onKernalStop(cancel);
        }

        cache.onKernalStop();

        if (!ctx.isRecoveryMode() && ctx.events().isRecordable(EventType.EVT_CACHE_STOPPED))
            ctx.events().addEvent(EventType.EVT_CACHE_STOPPED);
    }

    /**
     * @param cfg Cache configuration to use to create cache.
     * @param grp Cache group.
     * @param pluginMgr Cache plugin manager.
     * @param desc Cache descriptor.
     * @param locStartTopVer Current topology version.
     * @param cacheObjCtx Cache object context.
     * @param affNode {@code True} if local node affinity node.
     * @param updatesAllowed Updates allowed flag.
     * @param disabledAfterStart If true, then we will discard restarting state from proxies. If false then we will
     * change state of proxies to restarting
     * @return Cache context.
     * @throws IgniteCheckedException If failed to create cache.
     */
    private GridCacheContext<?, ?> createCacheContext(
        CacheConfiguration<?, ?> cfg,
        CacheGroupContext grp,
        @Nullable CachePluginManager pluginMgr,
        DynamicCacheDescriptor desc,
        AffinityTopologyVersion locStartTopVer,
        CacheObjectContext cacheObjCtx,
        boolean affNode,
        boolean updatesAllowed,
        boolean disabledAfterStart,
        boolean recoveryMode
    ) throws IgniteCheckedException {
        assert cfg != null;

        if (cfg.getCacheStoreFactory() instanceof GridCacheLoaderWriterStoreFactory) {
            GridCacheLoaderWriterStoreFactory factory = (GridCacheLoaderWriterStoreFactory)cfg.getCacheStoreFactory();

            prepare(cfg, factory.loaderFactory(), false);
            prepare(cfg, factory.writerFactory(), false);
        }
        else
            prepare(cfg, cfg.getCacheStoreFactory(), false);

        CacheStore cfgStore = cfg.getCacheStoreFactory() != null ? cfg.getCacheStoreFactory().create() : null;

        ValidationOnNodeJoinUtils.validate(ctx.config(), cfg, desc.cacheType(), cfgStore, ctx, log, (x, y) -> {
            try {
                assertParameter(x, y);
            }
            catch (IgniteCheckedException ex) {
                return ex;
            }

            return null;
        });

        if (pluginMgr == null)
            pluginMgr = new CachePluginManager(ctx, cfg);

        pluginMgr.validate();

        if (!recoveryMode && cfg.getAtomicityMode() == TRANSACTIONAL_SNAPSHOT && grp.affinityNode())
            sharedCtx.coordinators().ensureStarted();

        sharedCtx.jta().registerCache(cfg);

        // Skip suggestions for internal caches.
        if (desc.cacheType().userCache())
            suggestOptimizations(cfg, cfgStore != null);

        Collection<Object> toPrepare = new ArrayList<>();

        if (cfgStore instanceof GridCacheLoaderWriterStore) {
            toPrepare.add(((GridCacheLoaderWriterStore)cfgStore).loader());
            toPrepare.add(((GridCacheLoaderWriterStore)cfgStore).writer());
        }
        else
            toPrepare.add(cfgStore);

        prepare(cfg, toPrepare);

        U.startLifecycleAware(lifecycleAwares(grp, cfg, cfgStore));

        boolean nearEnabled = GridCacheUtils.isNearEnabled(cfg);

        CacheCompressionManager compressMgr = new CacheCompressionManager();
        GridCacheAffinityManager affMgr = new GridCacheAffinityManager();
        GridCacheEventManager evtMgr = new GridCacheEventManager();
        CacheEvictionManager evictMgr = (nearEnabled || cfg.isOnheapCacheEnabled())
            ? new GridCacheEvictionManager()
            : new CacheOffheapEvictionManager();
        GridCacheQueryManager qryMgr = new GridCacheDistributedQueryManager();
        CacheContinuousQueryManager contQryMgr = new CacheContinuousQueryManager();
        CacheDataStructuresManager dataStructuresMgr = new CacheDataStructuresManager();
        GridCacheTtlManager ttlMgr = new GridCacheTtlManager();

        CacheConflictResolutionManager rslvrMgr = pluginMgr.createComponent(CacheConflictResolutionManager.class);
        GridCacheDrManager drMgr = pluginMgr.createComponent(GridCacheDrManager.class);
        CacheStoreManager storeMgr = pluginMgr.createComponent(CacheStoreManager.class);
        PlatformCacheManager platformMgr = ctx.platform().cacheManager();

        if (cfgStore == null)
            storeMgr.initialize(cfgStore, sesHolders);
        else
            initializationProtector.protect(
                cfgStore,
                () -> storeMgr.initialize(cfgStore, sesHolders)
            );

        GridCacheContext<?, ?> cacheCtx = new GridCacheContext(
            ctx,
            sharedCtx,
            cfg,
            grp,
            desc.cacheType(),
            locStartTopVer,
            desc.deploymentId(),
            affNode,
            updatesAllowed,
            desc.cacheConfiguration().isStatisticsEnabled(),
            recoveryMode,
            /*
             * Managers in starting order!
             * ===========================
             */
            compressMgr,
            evtMgr,
            storeMgr,
            evictMgr,
            qryMgr,
            contQryMgr,
            dataStructuresMgr,
            ttlMgr,
            drMgr,
            rslvrMgr,
            pluginMgr,
            affMgr,
            platformMgr
        );

        cacheCtx.cacheObjectContext(cacheObjCtx);

        GridCacheAdapter cache = null;

        switch (cfg.getCacheMode()) {
            case PARTITIONED:
            case REPLICATED: {
                if (nearEnabled) {
                    switch (cfg.getAtomicityMode()) {
                        case TRANSACTIONAL:
                        case TRANSACTIONAL_SNAPSHOT: {
                            cache = new GridNearTransactionalCache(cacheCtx);

                            break;
                        }
                        case ATOMIC: {
                            cache = new GridNearAtomicCache(cacheCtx);

                            break;
                        }

                        default: {
                            assert false : "Invalid cache atomicity mode: " + cfg.getAtomicityMode();
                        }
                    }
                }
                else {
                    switch (cfg.getAtomicityMode()) {
                        case TRANSACTIONAL:
                        case TRANSACTIONAL_SNAPSHOT: {
                            cache = cacheCtx.affinityNode() ?
                                new GridDhtColocatedCache(cacheCtx) :
                                new GridDhtColocatedCache(cacheCtx, new GridNoStorageCacheMap());

                            break;
                        }
                        case ATOMIC: {
                            cache = cacheCtx.affinityNode() ?
                                new GridDhtAtomicCache(cacheCtx) :
                                new GridDhtAtomicCache(cacheCtx, new GridNoStorageCacheMap());

                            break;
                        }

                        default: {
                            assert false : "Invalid cache atomicity mode: " + cfg.getAtomicityMode();
                        }
                    }
                }

                break;
            }

            default: {
                assert false : "Invalid cache mode: " + cfg.getCacheMode();
            }
        }

        cache.active(!disabledAfterStart);

        cacheCtx.cache(cache);

        GridCacheContext<?, ?> ret = cacheCtx;

        /*
         * Create DHT cache.
         * ================
         */
        if (nearEnabled) {
            /*
             * Specifically don't create the following managers
             * here and reuse the one from Near cache:
             * 1. GridCacheVersionManager
             * 2. GridCacheIoManager
             * 3. GridCacheDeploymentManager
             * 4. GridCacheQueryManager (note, that we start it for DHT cache though).
             * 5. CacheContinuousQueryManager (note, that we start it for DHT cache though).
             * 6. GridCacheDgcManager
             * 7. GridCacheTtlManager.
             * 8. PlatformCacheManager.
             * ===============================================
             */
            evictMgr = cfg.isOnheapCacheEnabled() ? new GridCacheEvictionManager() : new CacheOffheapEvictionManager();
            evtMgr = new GridCacheEventManager();
            pluginMgr = new CachePluginManager(ctx, cfg);
            drMgr = pluginMgr.createComponent(GridCacheDrManager.class);

            cacheCtx = new GridCacheContext(
                ctx,
                sharedCtx,
                cfg,
                grp,
                desc.cacheType(),
                locStartTopVer,
                desc.deploymentId(),
                affNode,
                true,
                desc.cacheConfiguration().isStatisticsEnabled(),
                recoveryMode,
                /*
                 * Managers in starting order!
                 * ===========================
                 */
                compressMgr,
                evtMgr,
                storeMgr,
                evictMgr,
                qryMgr,
                contQryMgr,
                dataStructuresMgr,
                ttlMgr,
                drMgr,
                rslvrMgr,
                pluginMgr,
                affMgr,
                platformMgr
            );

            cacheCtx.cacheObjectContext(cacheObjCtx);

            GridDhtCacheAdapter dht = null;

            switch (cfg.getAtomicityMode()) {
                case TRANSACTIONAL:
                case TRANSACTIONAL_SNAPSHOT: {
                    assert cache instanceof GridNearTransactionalCache;

                    GridNearTransactionalCache near = (GridNearTransactionalCache)cache;

                    GridDhtCache dhtCache = cacheCtx.affinityNode() ?
                        new GridDhtCache(cacheCtx) :
                        new GridDhtCache(cacheCtx, new GridNoStorageCacheMap());

                    dhtCache.near(near);

                    near.dht(dhtCache);

                    dht = dhtCache;

                    break;
                }
                case ATOMIC: {
                    assert cache instanceof GridNearAtomicCache;

                    GridNearAtomicCache near = (GridNearAtomicCache)cache;

                    GridDhtAtomicCache dhtCache = cacheCtx.affinityNode() ?
                        new GridDhtAtomicCache(cacheCtx) :
                        new GridDhtAtomicCache(cacheCtx, new GridNoStorageCacheMap());

                    dhtCache.near(near);

                    near.dht(dhtCache);

                    dht = dhtCache;

                    break;
                }

                default: {
                    assert false : "Invalid cache atomicity mode: " + cfg.getAtomicityMode();
                }
            }

            cacheCtx.cache(dht);
        }

        if (!CU.isUtilityCache(cache.name()) && !CU.isSystemCache(cache.name())) {
            registerMbean(cache.localMxBean(), cache.name(), false);
            registerMbean(cache.clusterMxBean(), cache.name(), false);
        }

        return ret;
    }

    /**
     * @param reqs Cache requests to start.
     * @param fut Completable future.
     */
    public void registrateProxyRestart(Map<String, DynamicCacheChangeRequest> reqs, GridFutureAdapter<?> fut) {
        for (IgniteCacheProxyImpl<?, ?> proxy : jCacheProxies.values()) {
            if (reqs.containsKey(proxy.getName()) &&
                proxy.isRestarting() &&
                !reqs.get(proxy.getName()).disabledAfterStart()
            )
                proxy.registrateFutureRestart(fut);
        }
    }

    /**
     * @param reqs Cache requests to start.
     * @param initVer Init exchange version.
     * @param doneVer Finish excahnge vertison.
     */
    public void completeProxyRestart(
        Map<String, DynamicCacheChangeRequest> reqs,
        AffinityTopologyVersion initVer,
        AffinityTopologyVersion doneVer
    ) {
        if (initVer == null || doneVer == null)
            return;

        for (GridCacheAdapter<?, ?> cache : caches.values()) {
            GridCacheContext<?, ?> cacheCtx = cache.context();

            if (reqs.containsKey(cache.name()) ||
                (cacheCtx.startTopologyVersion().compareTo(initVer) <= 0 ||
                    cacheCtx.startTopologyVersion().compareTo(doneVer) <= 0))
                completeProxyInitialize(cache.name());

            if (
                cacheCtx.startTopologyVersion().compareTo(initVer) >= 0 &&
                    cacheCtx.startTopologyVersion().compareTo(doneVer) <= 0
            ) {
                IgniteCacheProxyImpl<?, ?> proxy = jCacheProxies.get(cache.name());

                boolean canRestart = Optional.ofNullable(reqs.get(cache.name()))
                    .map(req -> !req.disabledAfterStart())
                    .orElse(true);

                if (proxy != null && proxy.isRestarting() && canRestart) {
                    proxy.onRestarted(cacheCtx, cache);

                    if (cacheCtx.dataStructuresCache())
                        ctx.dataStructures().restart(cache.name(), proxy.internalProxy());
                }
            }
        }
    }

    /**
     * Gets a collection of currently started caches.
     *
     * @return Collection of started cache names.
     */
    public Collection<String> cacheNames() {
        return F.viewReadOnly(cacheDescriptors().values(),
            new IgniteClosure<DynamicCacheDescriptor, String>() {
                @Override public String apply(DynamicCacheDescriptor desc) {
                    return desc.cacheConfiguration().getName();
                }
            });
    }

    /**
     * Gets public cache that can be used for query execution. If cache isn't created on current node it will be
     * started.
     *
     * @param start Start cache.
     * @return Cache or {@code null} if there is no suitable cache.
     */
    public IgniteCacheProxy<?, ?> getOrStartPublicCache(boolean start) throws IgniteCheckedException {
        // Try to find started cache first.
        for (Map.Entry<String, GridCacheAdapter<?, ?>> e : caches.entrySet()) {
            if (!e.getValue().context().userCache())
                continue;

            CacheConfiguration ccfg = e.getValue().configuration();

            String cacheName = ccfg.getName();

            return publicJCache(cacheName);
        }

        if (start) {
            for (Map.Entry<String, DynamicCacheDescriptor> e : cachesInfo.registeredCaches().entrySet()) {
                DynamicCacheDescriptor desc = e.getValue();

                if (!desc.cacheType().userCache())
                    continue;

                CacheConfiguration ccfg = desc.cacheConfiguration();

                dynamicStartCache(null, ccfg.getName(), null, false, true, true).get();

                return publicJCache(ccfg.getName());
            }
        }

        return null;
    }

    /**
     * Gets a collection of currently started public cache names.
     *
     * @return Collection of currently started public cache names
     */
    public Collection<String> publicCacheNames() {
        return F.viewReadOnly(cacheDescriptors().values(),
            new IgniteClosure<DynamicCacheDescriptor, String>() {
                @Override public String apply(DynamicCacheDescriptor desc) {
                    return desc.cacheConfiguration().getName();
                }
            },
            new IgnitePredicate<DynamicCacheDescriptor>() {
                @Override public boolean apply(DynamicCacheDescriptor desc) {
                    return desc.cacheType().userCache();
                }
            }
        );
    }

    /**
     * Gets a collection of currently started public cache names.
     *
     * @return Collection of currently started public cache names
     */
    public Collection<String> publicAndDsCacheNames() {
        return F.viewReadOnly(cacheDescriptors().values(),
            new IgniteClosure<DynamicCacheDescriptor, String>() {
                @Override public String apply(DynamicCacheDescriptor desc) {
                    return desc.cacheConfiguration().getName();
                }
            },
            new IgnitePredicate<DynamicCacheDescriptor>() {
                @Override public boolean apply(DynamicCacheDescriptor desc) {
                    return desc.cacheType().userCache() || desc.cacheType() == CacheType.DATA_STRUCTURES;
                }
            }
        );
    }

    /**
     * Gets cache mode.
     *
     * @param cacheName Cache name to check.
     * @return Cache mode.
     */
    public CacheMode cacheMode(String cacheName) {
        assert cacheName != null;

        DynamicCacheDescriptor desc = cacheDescriptor(cacheName);

        return desc != null ? desc.cacheConfiguration().getCacheMode() : null;
    }

    /**
     * @return Caches to be started when this node starts.
     */
    @Nullable public LocalJoinCachesContext localJoinCachesContext() {
        if (ctx.discovery().localNode().order() == 1)
            cachesInfo.filterDynamicCacheDescriptors(locCfgMgr.localCachesOnStart());

        return cachesInfo.localJoinCachesContext();
    }

    /**
     * @param exchTopVer Local join exchange future version.
     * @param locJoinCtx Local join cache context.
     * @throws IgniteCheckedException If failed.
     */
    public IgniteInternalFuture<?> startCachesOnLocalJoin(
        AffinityTopologyVersion exchTopVer,
        LocalJoinCachesContext locJoinCtx
    ) throws IgniteCheckedException {
        long time = U.currentTimeMillis();

        if (locJoinCtx == null)
            return new GridFinishedFuture<>();

        IgniteInternalFuture<?> res = sharedCtx.affinity().initCachesOnLocalJoin(
            locJoinCtx.cacheGroupDescriptors(), locJoinCtx.cacheDescriptors());

        for (DynamicCacheDescriptor d: locJoinCtx.cacheDescriptors().values())
            ctx.coordinators().validateCacheConfiguration(d.cacheConfiguration());

        List<StartCacheInfo> startCacheInfos = locJoinCtx.caches().stream()
            .map(cacheInfo -> new StartCacheInfo(cacheInfo.get1(), cacheInfo.get2(), exchTopVer, false))
            .collect(toList());

        locJoinCtx.initCaches()
            .forEach(cacheDesc -> {
                try {
                    ctx.query().initQueryStructuresForNotStartedCache(cacheDesc);
                }
                catch (Exception e) {
                    log.error("Can't initialize query structures for not started cache [cacheName=" +
                        cacheDesc.cacheName() + "]", e);
                }
            });

        prepareStartCaches(startCacheInfos);

        context().exchange().exchangerUpdateHeartbeat();

        if (log.isInfoEnabled())
            log.info("Starting caches on local join performed in " + (U.currentTimeMillis() - time) + " ms.");

        return res;
    }

    /**
     * @param node Joined node.
     * @return {@code True} if there are new caches received from joined node.
     */
    public boolean hasCachesReceivedFromJoin(ClusterNode node) {
        return cachesInfo.hasCachesReceivedFromJoin(node.id());
    }

    /**
     * Starts statically configured caches received from remote nodes during exchange.
     *
     * @param nodeId Joining node ID.
     * @param exchTopVer Current exchange version.
     * @return Started caches descriptors.
     * @throws IgniteCheckedException If failed.
     */
    public Collection<DynamicCacheDescriptor> startReceivedCaches(UUID nodeId, AffinityTopologyVersion exchTopVer)
        throws IgniteCheckedException {
        List<DynamicCacheDescriptor> receivedCaches = cachesInfo.cachesReceivedFromJoin(nodeId);

        List<StartCacheInfo> startCacheInfos = receivedCaches.stream()
            .filter(desc -> isLocalAffinity(desc.groupDescriptor().config()))
            .map(desc -> new StartCacheInfo(desc, null, exchTopVer, false))
            .collect(toList());

        prepareStartCaches(startCacheInfos);

        return receivedCaches;
    }

    /**
     * @param cacheConfiguration Checked configuration.
     * @return {@code true} if local node is affinity node for cache.
     */
    private boolean isLocalAffinity(CacheConfiguration cacheConfiguration) {
        return CU.affinityNode(ctx.discovery().localNode(), cacheConfiguration.getNodeFilter());
    }

    /**
     * Start all input caches in parallel.
     *
     * @param startCacheInfos All caches information for start.
     */
    void prepareStartCaches(Collection<StartCacheInfo> startCacheInfos) throws IgniteCheckedException {
        prepareStartCaches(startCacheInfos, (data, operation) -> {
            operation.apply(data); // PROXY
        });
    }

    /**
     * Trying to start all input caches in parallel and skip failed caches.
     *
     * @param startCacheInfos Caches info for start.
     * @return Caches which was failed.
     * @throws IgniteCheckedException if failed.
     */
    Map<StartCacheInfo, IgniteCheckedException> prepareStartCachesIfPossible(
        Collection<StartCacheInfo> startCacheInfos
    ) throws IgniteCheckedException {
        HashMap<StartCacheInfo, IgniteCheckedException> failedCaches = new HashMap<>();

        prepareStartCaches(startCacheInfos, (data, operation) -> {
            try {
                operation.apply(data);
            }
            catch (IgniteInterruptedCheckedException e) {
                throw e;
            }
            catch (IgniteCheckedException e) {
                log.warning("Cache can not be started : cache=" + data.getStartedConfiguration().getName());

                failedCaches.put(data, e);
            }
        });

        return failedCaches;
    }

    /**
     * Start all input caches in parallel.
     *
     * @param startCacheInfos All caches information for start.
     * @param cacheStartFailHandler Fail handler for one cache start.
     */
    private void prepareStartCaches(
        Collection<StartCacheInfo> startCacheInfos,
        StartCacheFailHandler<StartCacheInfo, Void> cacheStartFailHandler
    ) throws IgniteCheckedException {
        if (!IGNITE_ALLOW_START_CACHES_IN_PARALLEL || startCacheInfos.size() <= 1) {
            for (StartCacheInfo startCacheInfo : startCacheInfos) {
                cacheStartFailHandler.handle(
                    startCacheInfo,
                    cacheInfo -> {
                        prepareCacheStart(
                            cacheInfo.getCacheDescriptor(),
                            cacheInfo.getReqNearCfg(),
                            cacheInfo.getExchangeTopVer(),
                            cacheInfo.isDisabledAfterStart(),
                            cacheInfo.isClientCache()
                        );

                        return null;
                    }
                );

                context().exchange().exchangerUpdateHeartbeat();
            }
        }
        else {
            Map<StartCacheInfo, GridCacheContext> cacheContexts = new ConcurrentHashMap<>();

            // Reserve at least 2 threads for system operations.
            int parallelismLvl = U.availableThreadCount(ctx, GridIoPolicy.SYSTEM_POOL, 2);

            doInParallel(
                parallelismLvl,
                sharedCtx.kernalContext().pools().getSystemExecutorService(),
                startCacheInfos,
                startCacheInfo -> {
                    cacheStartFailHandler.handle(
                        startCacheInfo,
                        cacheInfo -> {
                            GridCacheContext cacheCtx = prepareCacheContext(
                                cacheInfo.getCacheDescriptor(),
                                cacheInfo.getReqNearCfg(),
                                cacheInfo.getExchangeTopVer(),
                                cacheInfo.isDisabledAfterStart()
                            );
                            cacheContexts.put(cacheInfo, cacheCtx);

                            context().exchange().exchangerUpdateHeartbeat();

                            return null;
                        }
                    );

                    return null;
                }
            );

            /*
             * This hack required because we can't start sql schema in parallel by folowing reasons:
             * * checking index to duplicate(and other checking) require one order on every nodes.
             * * onCacheStart and createSchema contains a lot of mutex.
             *
             * TODO IGNITE-9729
             */
            Set<StartCacheInfo> successfullyPreparedCaches = cacheContexts.keySet();

            List<StartCacheInfo> cacheInfosInOriginalOrder = startCacheInfos.stream()
                .filter(successfullyPreparedCaches::contains)
                .collect(toList());

            for (StartCacheInfo startCacheInfo : cacheInfosInOriginalOrder) {
                cacheStartFailHandler.handle(
                    startCacheInfo,
                    cacheInfo -> {
                        GridCacheContext cctx = cacheContexts.get(cacheInfo);

                        if (!cctx.isRecoveryMode()) {
                            ctx.query().onCacheStart(
                                new GridCacheContextInfo(cctx, cacheInfo.isClientCache()),
                                cacheInfo.getCacheDescriptor().schema() != null
                                    ? cacheInfo.getCacheDescriptor().schema()
                                    : new QuerySchema(),
                                cacheInfo.getCacheDescriptor().sql()
                            );
                        }

                        context().exchange().exchangerUpdateHeartbeat();

                        return null;
                    }
                );
            }

            doInParallel(
                parallelismLvl,
                sharedCtx.kernalContext().pools().getSystemExecutorService(),
                cacheContexts.entrySet(),
                cacheCtxEntry -> {
                    cacheStartFailHandler.handle(
                        cacheCtxEntry.getKey(),
                        cacheInfo -> {
                            GridCacheContext<?, ?> cacheContext = cacheCtxEntry.getValue();

                            if (cacheContext.isRecoveryMode())
                                finishRecovery(cacheInfo.getExchangeTopVer(), cacheContext);
                            else
                                onCacheStarted(cacheCtxEntry.getValue());

                            context().exchange().exchangerUpdateHeartbeat();

                            return null;
                        }
                    );

                    return null;
                }
            );
        }
    }

    /**
     * @param desc Cache descriptor.
     * @param reqNearCfg Near configuration if specified for client cache start request.
     * @param exchTopVer Current exchange version.
     * @param disabledAfterStart If true, then we will discard restarting state from proxies. If false then we will
     * change state of proxies to restarting
     * @throws IgniteCheckedException If failed.
     */
    public void prepareCacheStart(
        DynamicCacheDescriptor desc,
        @Nullable NearCacheConfiguration reqNearCfg,
        AffinityTopologyVersion exchTopVer,
        boolean disabledAfterStart,
        boolean clientCache
    ) throws IgniteCheckedException {
        GridCacheContext cacheCtx = prepareCacheContext(desc, reqNearCfg, exchTopVer, disabledAfterStart);

        if (cacheCtx.isRecoveryMode())
            finishRecovery(exchTopVer, cacheCtx);
        else {
            ctx.query().onCacheStart(
                    new GridCacheContextInfo(cacheCtx, clientCache),
                    desc.schema() != null ? desc.schema() : new QuerySchema(),
                    desc.sql()
            );

            onCacheStarted(cacheCtx);
        }
    }

    /**
     * Preparing cache context to start.
     *
     * @param desc Cache descriptor.
     * @param reqNearCfg Near configuration if specified for client cache start request.
     * @param exchTopVer Current exchange version.
     * @param disabledAfterStart If true, then we will discard restarting state from proxies. If false then we will change
     *  state of proxies to restarting
     * @return Created {@link GridCacheContext}.
     * @throws IgniteCheckedException if failed.
     */
    private GridCacheContext prepareCacheContext(
        DynamicCacheDescriptor desc,
        @Nullable NearCacheConfiguration reqNearCfg,
        AffinityTopologyVersion exchTopVer,
        boolean disabledAfterStart
    ) throws IgniteCheckedException {
        desc = enricher().enrich(desc, isLocalAffinity(desc.cacheConfiguration()));

        CacheConfiguration startCfg = desc.cacheConfiguration();

        if (caches.containsKey(startCfg.getName())) {
            GridCacheAdapter<?, ?> existingCache = caches.get(startCfg.getName());

            GridCacheContext<?, ?> cctx = existingCache.context();

            assert cctx.isRecoveryMode();

            QuerySchema localSchema = recovery.querySchemas.get(desc.cacheId());

            QuerySchemaPatch localSchemaPatch = localSchema.makePatch(desc.schema().entities());

            // Cache schema is changed after restart, workaround is stop existing cache and start new.
            if (!localSchemaPatch.isEmpty() || localSchemaPatch.hasConflicts())
                stopCacheSafely(cctx);
            else
                return existingCache.context();
        }

        assert !caches.containsKey(startCfg.getName()) : startCfg.getName();

        CacheConfiguration ccfg = new CacheConfiguration(startCfg);

        CacheObjectContext cacheObjCtx = ctx.cacheObjects().contextForCache(ccfg);

        boolean affNode = checkForAffinityNode(desc, reqNearCfg, ccfg);

        ctx.cache().context().database().checkpointReadLock();

        try {
            CacheGroupContext grp = getOrCreateCacheGroupContext(
                desc, exchTopVer, cacheObjCtx, affNode, startCfg.getGroupName(), false);

            GridCacheContext cacheCtx = createCacheContext(ccfg,
                grp,
                null,
                desc,
                exchTopVer,
                cacheObjCtx,
                affNode,
                true,
                disabledAfterStart,
                false
            );

            initCacheContext(cacheCtx, ccfg);

            return cacheCtx;
        }
        finally {
            ctx.cache().context().database().checkpointReadUnlock();
        }
    }

    /**
     * Stops cache under checkpoint lock.
     *
     * @param cctx Cache context.
     */
    private void stopCacheSafely(GridCacheContext<?, ?> cctx) {
        stopCacheSafely(cctx, true);
    }

    /**
     * Stops cache under checkpoint lock.
     *
     * @param cctx Cache context.
     * @param clearDbObjects If {@code false} DB objects don't removed (used for cache.close() on client node).
     * The parameters was added due to make differ between cache.close() on client node and distributed destroy cache
     * (e.g. call cache.destroy()).
     *
     * Before add the parameter {@code clearDbObjects}:
     * when on client node is joined we initialize H2 objects for all caches in cluster,
     * but on client cache.close() we destroy part of created objects,
     * making impossible running SQL queries on that cache;
     * clientCache.close() should restore status quo of state right after client join instead).
     */
    private void stopCacheSafely(GridCacheContext<?, ?> cctx, boolean clearDbObjects) {
        sharedCtx.database().checkpointReadLock();

        try {
            prepareCacheStop(cctx.name(), false, false, clearDbObjects);

            if (!cctx.group().hasCaches())
                stopCacheGroup(cctx.group().groupId(), false);
        }
        finally {
            sharedCtx.database().checkpointReadUnlock();
        }

    }

    /**
     * Finishes recovery for given cache context.
     *
     * @param cacheStartVer Cache join to topology version.
     * @param cacheContext Cache context.
     * @throws IgniteCheckedException If failed.
     */
    private void finishRecovery(
        AffinityTopologyVersion cacheStartVer,
        GridCacheContext<?, ?> cacheContext
    ) throws IgniteCheckedException {
        CacheGroupContext groupContext = cacheContext.group();

        // Take cluster-wide cache descriptor and try to update local cache and cache group parameters.
        DynamicCacheDescriptor updatedDescriptor = cacheDescriptor(cacheContext.cacheId());

        groupContext.finishRecovery(
            cacheStartVer,
            updatedDescriptor.receivedFrom(),
            isLocalAffinity(updatedDescriptor.cacheConfiguration())
        );

        cacheContext.finishRecovery(cacheStartVer, updatedDescriptor);

        if (cacheContext.config().getAtomicityMode() == TRANSACTIONAL_SNAPSHOT && groupContext.affinityNode())
            sharedCtx.coordinators().ensureStarted();

        onKernalStart(cacheContext.cache());

        if (log.isInfoEnabled())
            log.info("Finished recovery for cache [cache=" + cacheContext.name()
                + ", grp=" + groupContext.cacheOrGroupName() + ", startVer=" + cacheStartVer + "]");
    }

    /**
     * Stops all caches and groups, that was recovered, but not activated on node join. Such caches can remain only if
     * it was filtered by node filter on current node. It's impossible to check whether current node is affinity node
     * for given cache before join to topology.
     */
    public void shutdownNotFinishedRecoveryCaches() {
        for (GridCacheAdapter cacheAdapter : caches.values()) {
            GridCacheContext cacheContext = cacheAdapter.context();

            if (cacheContext.isRecoveryMode()) {
                assert !isLocalAffinity(cacheContext.config())
                    : "Cache " + cacheAdapter.context() + " is still in recovery mode after start, but not activated.";

                stopCacheSafely(cacheContext);
            }
        }
    }

    /**
     * Check for affinity node and customize near configuration if needed.
     *
     * @param desc Cache descriptor.
     * @param reqNearCfg Near configuration if specified for client cache start request.
     * @param ccfg Cache configuration to use.
     * @return {@code true} if it is affinity node for cache.
     */
    private boolean checkForAffinityNode(
        DynamicCacheDescriptor desc,
        @Nullable NearCacheConfiguration reqNearCfg,
        CacheConfiguration ccfg
    ) {
        if (isLocalAffinity(desc.cacheConfiguration()))
            return true;

        ccfg.setNearConfiguration(reqNearCfg);

        return false;
    }

    /**
     * Prepare page store for start cache.
     *
     * @param desc Cache descriptor.
     * @param affNode {@code true} if it is affinity node for cache.
     * @throws IgniteCheckedException if failed.
     */
    public void preparePageStore(DynamicCacheDescriptor desc, boolean affNode) throws IgniteCheckedException {
        if (sharedCtx.pageStore() != null && affNode)
            initializationProtector.protect(
                desc.groupDescriptor().groupId(),
                () -> sharedCtx.pageStore().initializeForCache(desc.groupDescriptor(), desc.toStoredData(splitter).config())
            );
    }

    /**
     * Prepare cache group to start cache.
     *
     * @param desc Cache descriptor.
     * @param exchTopVer Current exchange version.
     * @param cacheObjCtx Cache object context.
     * @param affNode {@code true} if it is affinity node for cache.
     * @param grpName Group name.
     * @return Prepared cache group context.
     * @throws IgniteCheckedException if failed.
     */
    private CacheGroupContext getOrCreateCacheGroupContext(
        DynamicCacheDescriptor desc,
        AffinityTopologyVersion exchTopVer,
        CacheObjectContext cacheObjCtx,
        boolean affNode,
        String grpName,
        boolean recoveryMode
    ) throws IgniteCheckedException {
        if (grpName != null) {
            return initializationProtector.protect(
                desc.groupId(),
                () -> findCacheGroup(grpName),
                () -> startCacheGroup(
                    desc.groupDescriptor(),
                    desc.cacheType(),
                    affNode,
                    cacheObjCtx,
                    exchTopVer,
                    recoveryMode
                )
            );
        }

        return startCacheGroup(desc.groupDescriptor(),
            desc.cacheType(),
            affNode,
            cacheObjCtx,
            exchTopVer,
            recoveryMode
        );
    }

    /**
     * Initialize created cache context.
     *
     * @param cacheCtx Cache context to initializtion.
     * @param cfg Cache configuration.
     * @throws IgniteCheckedException if failed.
     */
    private void initCacheContext(
        GridCacheContext<?, ?> cacheCtx,
        CacheConfiguration cfg
    ) throws IgniteCheckedException {
        GridCacheAdapter cache = cacheCtx.cache();

        sharedCtx.addCacheContext(cacheCtx);

        caches.put(cacheCtx.name(), cache);

        // Intentionally compare Boolean references using '!=' below to check if the flag has been explicitly set.
        if (cfg.isStoreKeepBinary() && cfg.isStoreKeepBinary() != CacheConfiguration.DFLT_STORE_KEEP_BINARY
            && !(ctx.config().getMarshaller() instanceof BinaryMarshaller))
            U.warn(log, "CacheConfiguration.isStoreKeepBinary() configuration property will be ignored because " +
                "BinaryMarshaller is not used");

        // Start managers.
        for (GridCacheManager mgr : F.view(cacheCtx.managers(), F.notContains(dhtExcludes(cacheCtx))))
            mgr.start(cacheCtx);

        cacheCtx.initConflictResolver();

        if (GridCacheUtils.isNearEnabled(cfg)) {
            GridCacheContext<?, ?> dhtCtx = cacheCtx.near().dht().context();

            // Start DHT managers.
            for (GridCacheManager mgr : dhtManagers(dhtCtx))
                mgr.start(dhtCtx);

            dhtCtx.initConflictResolver();

            // Start DHT cache.
            dhtCtx.cache().start();

            if (log.isDebugEnabled())
                log.debug("Started DHT cache: " + dhtCtx.cache().name());
        }

        ctx.continuous().onCacheStart(cacheCtx);

        cacheCtx.cache().start();
    }

    /**
     * Handle of cache context which was fully prepared.
     *
     * @param cacheCtx Fully prepared context.
     * @throws IgniteCheckedException if failed.
     */
    private void onCacheStarted(GridCacheContext cacheCtx) throws IgniteCheckedException {
        GridCacheAdapter cache = cacheCtx.cache();
        CacheConfiguration cfg = cacheCtx.config();
        CacheGroupContext grp = cacheGrps.get(cacheCtx.groupId());

        cacheCtx.onStarted();

        String dataRegion = cfg.getDataRegionName();

        if (dataRegion == null && ctx.config().getDataStorageConfiguration() != null)
            dataRegion = ctx.config().getDataStorageConfiguration().getDefaultDataRegionConfiguration().getName();

        if (log.isInfoEnabled()) {
            String expPlcInfo = buildExpirePolicyInfo(cacheCtx);

            log.info("Started cache [name=" + cfg.getName() +
                ", id=" + cacheCtx.cacheId() +
                (cfg.getGroupName() != null ? ", group=" + cfg.getGroupName() : "") +
                ", dataRegionName=" + dataRegion +
                ", mode=" + cfg.getCacheMode() +
                ", atomicity=" + cfg.getAtomicityMode() +
                ", backups=" + cfg.getBackups() +
                ", mvcc=" + cacheCtx.mvccEnabled() +
                (expPlcInfo != null ? ", " + expPlcInfo : "") + ']');
        }

        grp.onCacheStarted(cacheCtx);

        onKernalStart(cache);

        if (ctx.performanceStatistics().enabled() && U.isLocalNodeCoordinator(ctx.discovery()))
            ctx.performanceStatistics().cacheStart(cacheCtx.cacheId(), cfg.getName());
    }

    /**
     * @param desc Cache descriptor.
     * @throws IgniteCheckedException If failed.
     */
    private GridCacheContext<?, ?> startCacheInRecoveryMode(
        DynamicCacheDescriptor desc
    ) throws IgniteCheckedException {
        // Only affinity nodes are able to start cache in recovery mode.
        desc = enricher().enrich(desc, true);

        CacheConfiguration cfg = desc.cacheConfiguration();

        CacheObjectContext cacheObjCtx = ctx.cacheObjects().contextForCache(cfg);

        preparePageStore(desc, true);

        CacheGroupContext grpCtx;
        GridCacheContext cacheCtx;

        ctx.cache().context().database().checkpointReadLock();

        try {
            grpCtx = getOrCreateCacheGroupContext(
                desc,
                AffinityTopologyVersion.NONE,
                cacheObjCtx,
                true,
                cfg.getGroupName(),
                true
            );

            cacheCtx = createCacheContext(cfg,
                grpCtx,
                null,
                desc,
                AffinityTopologyVersion.NONE,
                cacheObjCtx,
                true,
                true,
                false,
                true
            );

            initCacheContext(cacheCtx, cfg);
        }
        finally {
            ctx.cache().context().database().checkpointReadUnlock();
        }

        cacheCtx.onStarted();

        String dataRegion = cfg.getDataRegionName();

        if (dataRegion == null && ctx.config().getDataStorageConfiguration() != null)
            dataRegion = ctx.config().getDataStorageConfiguration().getDefaultDataRegionConfiguration().getName();

        grpCtx.onCacheStarted(cacheCtx);

        ctx.query().onCacheStart(new GridCacheContextInfo(cacheCtx, false),
            desc.schema() != null ? desc.schema() : new QuerySchema(), desc.sql());

        if (log.isInfoEnabled()) {
            String expPlcInfo = buildExpirePolicyInfo(cacheCtx);

            log.info("Started cache in recovery mode [name=" + cfg.getName() +
                ", id=" + cacheCtx.cacheId() +
                (cfg.getGroupName() != null ? ", group=" + cfg.getGroupName() : "") +
                ", dataRegionName=" + dataRegion +
                ", mode=" + cfg.getCacheMode() +
                ", atomicity=" + cfg.getAtomicityMode() +
                ", backups=" + cfg.getBackups() +
                ", mvcc=" + cacheCtx.mvccEnabled() +
                (expPlcInfo != null ? ", " + expPlcInfo : "") + ']');
        }

        return cacheCtx;
    }

    /**
     * Build formatted string with expire policy info.
     *
     * @param cacheCtx - cache context.
     * @return formatted expire policy info.
     */
    private String buildExpirePolicyInfo(@NotNull GridCacheContext cacheCtx) {
        ExpiryPolicy expPlc = cacheCtx.expiry();
        if (expPlc == null || expPlc instanceof EternalExpiryPolicy)
            return null;

        return String.format(EXPRITY_POLICY_MSG, expPlc.getClass().getName(), cacheCtx.ttl().eagerTtlEnabled());
    }

    /**
     * @param grpName Group name.
     * @return Found group or null.
     */
    private CacheGroupContext findCacheGroup(String grpName) {
        return cacheGrps.values().stream()
            .filter(grp -> grp.sharedGroup() && grpName.equals(grp.name()))
            .findAny()
            .orElse(null);
    }

    /**
     * Restarts proxies of caches if they was marked as restarting. Requires external synchronization - shouldn't be
     * called concurrently with another caches restart.
     */
    public void restartProxies() {
        for (IgniteCacheProxyImpl<?, ?> proxy : jCacheProxies.values()) {
            if (proxy == null)
                continue;

            GridCacheContext<?, ?> cacheCtx = sharedCtx.cacheContext(CU.cacheId(proxy.getName()));

            if (cacheCtx == null)
                continue;

            if (proxy.isRestarting()) {
                caches.get(proxy.getName()).active(true);

                proxy.onRestarted(cacheCtx, cacheCtx.cache());

                if (cacheCtx.dataStructuresCache())
                    ctx.dataStructures().restart(proxy.getName(), proxy.internalProxy());
            }
        }
    }

    /**
     * Complete stopping of caches if they were marked as restarting but it failed.
     * @return Cache names of proxies which were restarted.
     */
    public List<String> resetRestartingProxies() {
        List<String> res = new ArrayList<>();

        for (Map.Entry<String, IgniteCacheProxyImpl<?, ?>> e : jCacheProxies.entrySet()) {
            IgniteCacheProxyImpl<?, ?> proxy = e.getValue();

            if (proxy == null)
                continue;

            if (proxy.isRestarting()) {
                String cacheName = e.getKey();

                res.add(cacheName);

                jCacheProxies.remove(cacheName);

                proxy.onRestarted(null, null);

                if (DataStructuresProcessor.isDataStructureCache(cacheName))
                    ctx.dataStructures().restart(cacheName, null);
            }
        }

        cachesInfo.removeRestartingCaches();

        return res;
    }

    /**
     * @param desc Group descriptor.
     * @param cacheType Cache type.
     * @param affNode Affinity node flag.
     * @param cacheObjCtx Cache object context.
     * @param exchTopVer Current topology version.
     * @return Started cache group.
     * @throws IgniteCheckedException If failed.
     */
    private CacheGroupContext startCacheGroup(
        CacheGroupDescriptor desc,
        CacheType cacheType,
        boolean affNode,
        CacheObjectContext cacheObjCtx,
        AffinityTopologyVersion exchTopVer,
        boolean recoveryMode
    ) throws IgniteCheckedException {
        desc = enricher().enrich(desc, affNode);

        CacheConfiguration cfg = new CacheConfiguration(desc.config());

        String memPlcName = cfg.getDataRegionName();

        DataRegion dataRegion = affNode ? sharedCtx.database().dataRegion(memPlcName) : null;

        boolean needToStart = (dataRegion != null)
            && (cacheType != CacheType.USER
                || (sharedCtx.isLazyMemoryAllocation(dataRegion)
                    && !cacheObjCtx.kernalContext().clientNode()));

        if (needToStart)
            dataRegion.pageMemory().start();

        FreeList freeList = sharedCtx.database().freeList(memPlcName);
        ReuseList reuseList = sharedCtx.database().reuseList(memPlcName);

        boolean persistenceEnabled = recoveryMode || sharedCtx.localNode().isClient() ? desc.persistenceEnabled() :
            dataRegion != null && dataRegion.config().isPersistenceEnabled();

        CacheGroupContext grp = new CacheGroupContext(sharedCtx,
            desc.groupId(),
            desc.receivedFrom(),
            cacheType,
            cfg,
            affNode,
            dataRegion,
            cacheObjCtx,
            freeList,
            reuseList,
            exchTopVer,
            persistenceEnabled,
            desc.walEnabled(),
            recoveryMode
        );

        for (Object obj : grp.configuredUserObjects())
            prepare(cfg, obj, false);

        U.startLifecycleAware(grp.configuredUserObjects());

        grp.start();

        CacheGroupContext old = cacheGrps.put(desc.groupId(), grp);

        if (!grp.systemCache() && !U.IGNITE_MBEANS_DISABLED) {
            try {
                U.registerMBean(ctx.config().getMBeanServer(), ctx.igniteInstanceName(), CACHE_GRP_METRICS_MBEAN_GRP,
                    grp.cacheOrGroupName(), new CacheGroupMetricsMXBeanImpl(grp), CacheGroupMetricsMXBean.class);
            }
            catch (Throwable e) {
                U.error(log, "Failed to register MBean for cache group: " + grp.name(), e);
            }
        }

        assert old == null : old.name();

        return grp;
    }

    /**
     * @param cacheName Cache name.
     * @param stop {@code True} for stop cache, {@code false} for close cache.
     * @param restart Restart flag.
     */
    public void blockGateway(String cacheName, boolean stop, boolean restart) {
        // Break the proxy before exchange future is done.
        IgniteCacheProxyImpl<?, ?> proxy = jcacheProxy(cacheName, false);

        if (restart) {
            GridCacheAdapter<?, ?> cache = caches.get(cacheName);

            if (cache != null)
                cache.active(false);
        }

        if (stop) {
            if (restart) {
                GridCacheAdapter<?, ?> cache;

                if (proxy == null && (cache = caches.get(cacheName)) != null) {
                    proxy = new IgniteCacheProxyImpl(cache.context(), cache, false);

                    IgniteCacheProxyImpl<?, ?> oldProxy = jCacheProxies.putIfAbsent(cacheName, proxy);

                    if (oldProxy != null)
                        proxy = oldProxy;
                }

                if (proxy != null)
                    proxy.suspend();
            }

            if (proxy != null)
                proxy.context0().gate().stopped();
        }
        else if (proxy != null)
            proxy.closeProxy();
    }

    /**
     * @param req Request.
     */
    private void stopGateway(DynamicCacheChangeRequest req) {
        assert req.stop() : req;

        IgniteCacheProxyImpl<?, ?> proxy;

        // Break the proxy before exchange future is done.
        if (req.restart()) {
            if (DataStructuresProcessor.isDataStructureCache(req.cacheName()))
                ctx.dataStructures().suspend(req.cacheName());

            GridCacheAdapter<?, ?> cache = caches.get(req.cacheName());

            if (cache != null)
                cache.active(false);

            proxy = jCacheProxies.get(req.cacheName());

            if (proxy != null)
                proxy.suspend();
        }
        else {
            completeProxyInitialize(req.cacheName());

            proxy = jCacheProxies.remove(req.cacheName());
        }

        if (proxy != null)
            proxy.context0().gate().onStopped();
    }

    /**
     * @param cacheName Cache name.
     * @param callDestroy Cache destroy flag. This flag is passed to the {@link GridCacheManager#stop} method.
     * @param clearCache Cache data clear flag. Setting to {@code true} will remove all cache data.
     */
    public void prepareCacheStop(String cacheName, boolean callDestroy, boolean clearCache) {
        prepareCacheStop(cacheName, callDestroy, clearCache, true);
    }

    /**
     * @param cacheName Cache name.
     * @param callDestroy Cache destroy flag. This flag is passed to the {@link GridCacheManager#stop} method.
     * @param clearCache Cache data clear flag. Setting to {@code true} will remove all cache data.
     * @param clearDbObjects If {@code false} DB objects don't removed (used for cache.close() on client node).
     */
    public void prepareCacheStop(String cacheName, boolean callDestroy, boolean clearCache, boolean clearDbObjects) {
        assert sharedCtx.database().checkpointLockIsHeldByThread();

        GridCacheAdapter<?, ?> cache = caches.remove(cacheName);

        if (cache != null) {
            GridCacheContext<?, ?> ctx = cache.context();

            sharedCtx.removeCacheContext(ctx);

            onKernalStop(cache, true);

            stopCache(cache, true, callDestroy, clearCache, clearDbObjects);
        }
        else
            // Try to unregister query structures for not started caches.
            ctx.query().onCacheStop(cacheName);
    }

    /**
     * @param startTopVer Cache start version.
     * @param err Cache start error if any.
     */
    void initCacheProxies(AffinityTopologyVersion startTopVer, @Nullable Throwable err) {
        for (GridCacheAdapter<?, ?> cache : caches.values()) {
            GridCacheContext<?, ?> cacheCtx = cache.context();

            if (cacheCtx.startTopologyVersion().equals(startTopVer)) {
                if (!jCacheProxies.containsKey(cacheCtx.name())) {
                    IgniteCacheProxyImpl<?, ?> newProxy = new IgniteCacheProxyImpl(cache.context(), cache, false);

                    if (!cache.active())
                        newProxy.suspend();

                    addjCacheProxy(cacheCtx.name(), newProxy);
                }

                if (cacheCtx.preloader() != null)
                    cacheCtx.preloader().onInitialExchangeComplete(err);
            }
        }
    }

    /**
     * @param cachesToClose Caches to close.
     * @param retClientCaches {@code True} if return IDs of closed client caches.
     * @return Closed client caches' IDs.
     */
    Set<Integer> closeCaches(Set<String> cachesToClose, boolean retClientCaches) {
        Set<Integer> ids = null;

        for (String cacheName : cachesToClose) {
            completeProxyInitialize(cacheName);

            blockGateway(cacheName, false, false);

            GridCacheContext ctx = sharedCtx.cacheContext(CU.cacheId(cacheName));

            if (ctx == null)
                continue;

            if (retClientCaches && !ctx.affinityNode()) {
                if (ids == null)
                    ids = U.newHashSet(cachesToClose.size());

                ids.add(ctx.cacheId());
            }

            closeCache(ctx);
        }

        return ids;
    }

    /**
     * @param cctx Cache context.
     */
    private void closeCache(GridCacheContext cctx) {
        if (cctx.affinityNode()) {
            GridCacheAdapter<?, ?> cache = caches.get(cctx.name());

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

            jCacheProxies.put(cctx.name(), new IgniteCacheProxyImpl(cache.context(), cache, false));

            completeProxyInitialize(cctx.name());
        }
        else {
            cctx.gate().onStopped();

            // Do not close client cache while requests processing is in progress.
            sharedCtx.io().writeLock();

            try {
                if (!cctx.affinityNode() && cctx.transactional())
                    sharedCtx.tm().rollbackTransactionsForCache(cctx.cacheId());

                completeProxyInitialize(cctx.name());

                jCacheProxies.remove(cctx.name());

                closeCacheOnNotAffinityNode(cctx);
            }
            finally {
                sharedCtx.io().writeUnlock();
            }
        }
    }

    /**
     * @param cctx Cache context to close.
     */
    private void closeCacheOnNotAffinityNode(GridCacheContext cctx) {
        if (ctx.query().moduleEnabled())
            stopCacheSafely(cctx, false);
        else
            stopCacheSafely(cctx);
    }

    /**
     * Called during the rollback of the exchange partitions procedure in order to stop the given cache even if it's not
     * fully initialized (e.g. failed on cache init stage).
     *
     * @param topVer Topology version related to the given {@code exchActions}.
     * @param exchActions Stop requests.
     */
    void forceCloseCaches(AffinityTopologyVersion topVer, ExchangeActions exchActions) {
        assert exchActions != null && !exchActions.cacheStopRequests().isEmpty();

        processCacheStopRequestOnExchangeDone(topVer, exchActions);
    }

    /**
     * @param topVer Topology version related to the given {@code exchActions}.
     * @param exchActions Change requests.
     */
    private void processCacheStopRequestOnExchangeDone(AffinityTopologyVersion topVer, ExchangeActions exchActions) {
        // Reserve at least 2 threads for system operations.
        int parallelismLvl = U.availableThreadCount(ctx, GridIoPolicy.SYSTEM_POOL, 2);

        List<IgniteBiTuple<CacheGroupContext, Boolean>> grpsToStop = exchActions.cacheGroupsToStop().stream()
            .filter(a -> cacheGrps.containsKey(a.descriptor().groupId()))
            .map(a -> F.t(cacheGrps.get(a.descriptor().groupId()), a.destroy()))
            .collect(toList());

        // Wait until all evictions are finished.
        grpsToStop.forEach(t -> sharedCtx.evict().onCacheGroupStopped(t.get1()));

        if (!exchActions.cacheStopRequests().isEmpty())
            removeOffheapListenerAfterCheckpoint(grpsToStop);

        Map<Integer, List<ExchangeActions.CacheActionData>> cachesToStop = exchActions.cacheStopRequests().stream()
            .collect(groupingBy(action -> action.descriptor().groupId()));

        Set<Integer> grpIdToDestroy = grpsToStop.stream()
            .filter(IgniteBiTuple::get2).map(t2 -> t2.get1().groupId()).collect(toSet());

        try {
            doInParallel(
                parallelismLvl,
                sharedCtx.kernalContext().pools().getSystemExecutorService(),
                cachesToStop.entrySet(),
                cachesToStopByGrp -> {
                    Integer groupId = cachesToStopByGrp.getKey();

                    CacheGroupContext gctx = cacheGrps.get(groupId);

                    if (gctx != null)
                        gctx.preloader().pause();

                    try {
                        if (gctx != null) {
                            final String msg = "Failed to wait for topology update, cache group is stopping.";

                            // If snapshot operation in progress we must throw CacheStoppedException
                            // for correct cache proxy restart. For more details see
                            // IgniteCacheProxy.cacheException()
                            gctx.affinity().cancelFutures(new CacheStoppedException(msg));
                        }

                        for (ExchangeActions.CacheActionData action : cachesToStopByGrp.getValue()) {
                            context().tm().rollbackTransactionsForStoppingCache(action.descriptor().cacheId());

                            stopGateway(action.request());

                            String cacheName = action.request().cacheName();

                            // TTL manager has to be unregistered before the checkpointReadLock is acquired.
                            GridCacheAdapter<?, ?> cache = caches.get(cacheName);

                            if (cache != null)
                                cache.context().ttl().unregister();

                            sharedCtx.database().checkpointReadLock();

                            try {
                                boolean callDestroy = action.request().destroy();

                                // If all caches in group will be destroyed it is not necessary to clear a single cache
                                // because group will be stopped anyway.
                                // Stop will still be called with destroy {@code true}, but cache will not be cleared.
                                boolean clearCache = callDestroy && !grpIdToDestroy.contains(groupId);

                                prepareCacheStop(cacheName, callDestroy, clearCache);

                                if (callDestroy || grpIdToDestroy.contains(groupId))
                                    ctx.query().completeRebuildIndexes(cacheName);
                            }
                            finally {
                                sharedCtx.database().checkpointReadUnlock();
                            }
                        }
                    }
                    finally {
                        if (gctx != null)
                            gctx.preloader().resume();
                    }

                    return null;
                }
            );
        }
        catch (IgniteCheckedException e) {
            String msg = "Failed to stop caches";

            log.error(msg, e);

            throw new IgniteException(msg, e);
        }
        finally {
            cachesInfo.cleanupRemovedCaches(topVer);
        }

        for (IgniteBiTuple<CacheGroupContext, Boolean> grp : grpsToStop)
            stopCacheGroup(grp.get1().groupId(), grp.get2());

        if (!sharedCtx.kernalContext().clientNode())
            sharedCtx.database().onCacheGroupsStopped(grpsToStop);

        cachesInfo.cleanupRemovedCacheGroups(topVer);

        if (exchActions.deactivate())
            sharedCtx.deactivate();
    }

    /**
     * @param rmtNode Remote node to check.
     * @return Data storage configuration
     */
    private DataStorageConfiguration extractDataStorage(ClusterNode rmtNode) {
        return GridCacheUtils.extractDataStorage(
            rmtNode,
            ctx.marshallerContext().jdkMarshaller(),
            U.resolveClassLoader(ctx.config())
        );
    }

    /**
     * @param dataStorageCfg User-defined data regions.
     */
    private Map<String, DataRegionConfiguration> dataRegionCfgs(DataStorageConfiguration dataStorageCfg) {
        if (dataStorageCfg != null) {
            return Optional.ofNullable(dataStorageCfg.getDataRegionConfigurations())
                .map(Stream::of)
                .orElseGet(Stream::empty)
                .collect(Collectors.toMap(DataRegionConfiguration::getName, e -> e));
        }

        return Collections.emptyMap();
    }

    /**
     * Force checkpoint and remove offheap checkpoint listener after it was finished.
     *
     * @param grpToStop Cache group to stop.
     */
    private void removeOffheapListenerAfterCheckpoint(List<IgniteBiTuple<CacheGroupContext, Boolean>> grpToStop) {
        try {
            sharedCtx.database().waitForCheckpoint(
                "caches stop", (fut) -> removeOffheapCheckpointListener(grpToStop)
            );
        }
        catch (IgniteCheckedException e) {
            U.error(log, "Failed to wait for checkpoint finish during cache stop.", e);
        }
    }

    /**
     * @param grpToStop Group for which listener shuold be removed.
     */
    private void removeOffheapCheckpointListener(List<IgniteBiTuple<CacheGroupContext, Boolean>> grpToStop) {
        sharedCtx.database().checkpointReadLock();
        try {
            // Do not invoke checkpoint listeners for groups are going to be destroyed to prevent metadata corruption.
            grpToStop.forEach(grp -> {
                CacheGroupContext gctx = grp.getKey();

                if (gctx != null && gctx.persistenceEnabled() && sharedCtx.database() instanceof GridCacheDatabaseSharedManager) {
                    GridCacheDatabaseSharedManager mngr = (GridCacheDatabaseSharedManager)sharedCtx.database();
                    mngr.removeCheckpointListener((CheckpointListener)gctx.offheap());
                }
            });
        }
        finally {
            sharedCtx.database().checkpointReadUnlock();
        }
    }

    /**
     * Callback invoked when first exchange future for dynamic cache is completed.
     *
     * @param cacheStartVer Started caches version to create proxy for.
     * @param exchActions Change requests.
     * @param err Error.
     */
    public void onExchangeDone(
        AffinityTopologyVersion cacheStartVer,
        @Nullable ExchangeActions exchActions,
        @Nullable Throwable err
    ) {
        initCacheProxies(cacheStartVer, err);

        if (exchActions == null)
            return;

        if (exchActions.systemCachesStarting() && exchActions.stateChangeRequest() == null)
            ctx.dataStructures().restoreStructuresState(ctx);

        if (err == null)
            processCacheStopRequestOnExchangeDone(cacheStartVer, exchActions);
    }

    /**
     * @param grpId Group ID.
     * @param destroy Group destroy flag.
     */
    private void stopCacheGroup(int grpId, boolean destroy) {
        CacheGroupContext grp = cacheGrps.remove(grpId);

        if (grp != null)
            stopCacheGroup(grp, destroy);
    }

    /**
     * @param grp Cache group.
     * @param destroy Group destroy flag.
     */
    private void stopCacheGroup(CacheGroupContext grp, boolean destroy) {
        grp.stopGroup();

        U.stopLifecycleAware(log, grp.configuredUserObjects());

        cleanup(grp, destroy);
    }

    /**
     * @param cacheName Cache name.
     * @param deploymentId Future deployment ID.
     */
    void completeTemplateAddFuture(String cacheName, IgniteUuid deploymentId) {
        GridCacheProcessor.TemplateConfigurationFuture fut =
            (GridCacheProcessor.TemplateConfigurationFuture)pendingTemplateFuts.get(cacheName);

        if (fut != null && fut.deploymentId().equals(deploymentId))
            fut.onDone();
    }

    /**
     * @param req Request to complete future for.
     * @param success Future result.
     * @param err Error if any.
     */
    public void completeCacheStartFuture(DynamicCacheChangeRequest req, boolean success, @Nullable Throwable err) {
        if (ctx.localNodeId().equals(req.initiatingNodeId())) {
            DynamicCacheStartFuture fut = (DynamicCacheStartFuture)pendingFuts.get(req.requestId());

            if (fut != null)
                fut.onDone(success, err);
        }
    }

    /**
     * @param reqId Request ID.
     * @param err Error if any.
     */
    void completeClientCacheChangeFuture(UUID reqId, @Nullable Exception err) {
        DynamicCacheStartFuture fut = (DynamicCacheStartFuture)pendingFuts.get(reqId);

        if (fut != null)
            fut.onDone(false, err);
    }

    /**
     * Creates shared context.
     *
     * @param kernalCtx Kernal context.
     * @param storeSesLsnrs Store session listeners.
     * @return Shared context.
     * @throws IgniteCheckedException If failed.
     */
    @SuppressWarnings("unchecked")
    private GridCacheSharedContext createSharedContext(
        GridKernalContext kernalCtx,
        Collection<CacheStoreSessionListener> storeSesLsnrs
    ) throws IgniteCheckedException {
        IgniteTxManager tm = new IgniteTxManager();
        GridCacheMvccManager mvccMgr = new GridCacheMvccManager();
        GridCacheVersionManager verMgr = new GridCacheVersionManager();
        GridCacheDeploymentManager depMgr = new GridCacheDeploymentManager();
        GridCachePartitionExchangeManager exchMgr = new GridCachePartitionExchangeManager();

        IgniteCacheDatabaseSharedManager dbMgr;
        IgnitePageStoreManager pageStoreMgr = null;
        IgniteWriteAheadLogManager walMgr = null;

        if (CU.isPersistenceEnabled(ctx.config()) && !ctx.clientNode()) {
            dbMgr = new GridCacheDatabaseSharedManager(ctx);

            pageStoreMgr = ctx.plugins().createComponent(IgnitePageStoreManager.class);

            if (pageStoreMgr == null)
                pageStoreMgr = new FilePageStoreManager(ctx);
        }
        else {
            if (CU.isPersistenceEnabled(ctx.config()) && ctx.clientNode()) {
                U.warn(log, "Persistent Store is not supported on client nodes (Persistent Store's" +
                    " configuration will be ignored).");
            }

            dbMgr = new IgniteCacheDatabaseSharedManager(ctx);
        }

        if ((CU.isPersistenceEnabled(ctx.config()) || CU.isCdcEnabled(ctx.config())) && !ctx.clientNode()) {
            walMgr = ctx.plugins().createComponent(IgniteWriteAheadLogManager.class);

            if (walMgr == null)
                walMgr = new FileWriteAheadLogManager(ctx);
        }

        WalStateManager walStateMgr = new WalStateManager(ctx);

        IgniteSnapshotManager snapshotMgr = new IgniteSnapshotManager(ctx);
        IgniteCacheSnapshotManager snpMgr = ctx.plugins().createComponent(IgniteCacheSnapshotManager.class);

        if (snpMgr == null)
            snpMgr = new IgniteCacheSnapshotManager();

        GridCacheIoManager ioMgr = new GridCacheIoManager();
        CacheAffinitySharedManager topMgr = new CacheAffinitySharedManager();
        GridCacheSharedTtlCleanupManager ttl = new GridCacheSharedTtlCleanupManager();
        PartitionsEvictManager evict = new PartitionsEvictManager();

        CacheJtaManagerAdapter jta = JTA.createOptional();

        MvccCachingManager mvccCachingMgr = new MvccCachingManager();

        DeadlockDetectionManager deadlockDetectionMgr = new DeadlockDetectionManager();

        CacheDiagnosticManager diagnosticMgr = new CacheDiagnosticManager();

        return new GridCacheSharedContext(
            kernalCtx,
            tm,
            verMgr,
            mvccMgr,
            pageStoreMgr,
            walMgr,
            walStateMgr,
            dbMgr,
            snapshotMgr,
            snpMgr,
            depMgr,
            exchMgr,
            topMgr,
            ioMgr,
            ttl,
            evict,
            jta,
            storeSesLsnrs,
            mvccCachingMgr,
            deadlockDetectionMgr,
            diagnosticMgr
        );
    }

    /** {@inheritDoc} */
    @Nullable @Override public DiscoveryDataExchangeType discoveryDataType() {
        return CACHE_PROC;
    }

    /** {@inheritDoc} */
    @Override public void collectJoiningNodeData(DiscoveryDataBag dataBag) {
        cachesInfo.collectJoiningNodeData(dataBag);
    }

    /** {@inheritDoc} */
    @Override public void collectGridNodeData(DiscoveryDataBag dataBag) {
        cachesInfo.collectGridNodeData(dataBag, backwardCompatibleSplitter());
    }

    /** {@inheritDoc} */
    @Override public void onJoiningNodeDataReceived(JoiningNodeDiscoveryData data) {
        cachesInfo.onJoiningNodeDataReceived(data);
    }

    /** {@inheritDoc} */
    @Override public void onGridDataReceived(GridDiscoveryData data) {
        cachesInfo.onGridDataReceived(data);

        sharedCtx.walState().onCachesInfoCollected();
    }

    /** {@inheritDoc} */
    @Override public @Nullable IgniteNodeValidationResult validateNode(
        ClusterNode node, JoiningNodeDiscoveryData discoData
    ) {
        if (!cachesInfo.isMergeConfigSupports(node))
            return null;

        String validationRes = cachesInfo.validateJoiningNodeData(discoData, node.isClient());

        if (validationRes != null)
            return new IgniteNodeValidationResult(node.id(), validationRes, validationRes);

        return ValidationOnNodeJoinUtils.validateNode(node, discoData, marsh, ctx, this::cacheDescriptor);
    }

    /**
     * @param msg Message.
     */
    public void onStateChangeFinish(ChangeGlobalStateFinishMessage msg) {
        cachesInfo.onStateChangeFinish(msg);
    }

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

    /**
     * Cache statistics flag change message received.
     *
     * @param msg Message.
     */
    public void onCacheStatisticsModeChange(CacheStatisticsModeChangeMessage msg) {
        assert msg != null;

        if (msg.initial()) {
            EnableStatisticsFuture fut = manageStatisticsFuts.get(msg.requestId());

            if (fut != null && !cacheNames().containsAll(msg.caches())) {
                fut.onDone(new IgniteCheckedException("One or more cache descriptors not found [caches="
                    + caches + ']'));

                return;
            }

            for (String cacheName : msg.caches()) {
                DynamicCacheDescriptor desc = cachesInfo.registeredCaches().get(cacheName);

                if (desc != null) {
                    if (desc.cacheConfiguration().isStatisticsEnabled() != msg.enabled()) {
                        desc.cacheConfiguration().setStatisticsEnabled(msg.enabled());

                        try {
                            ctx.cache().saveCacheConfiguration(desc);
                        }
                        catch (IgniteCheckedException e) {
                            log.error("Error while saving cache configuration to disk, cfg = "
                                + desc.cacheConfiguration(), e);
                        }
                    }
                }
                else
                    log.warning("Failed to change cache descriptor configuration, cache not found [cacheName="
                        + cacheName + ']');
            }
        }
        else {
            EnableStatisticsFuture fut = manageStatisticsFuts.get(msg.requestId());

            if (fut != null)
                fut.onDone();
        }
    }

    /**
     * Cache statistics clear message received.
     *
     * @param msg Message.
     */
    private void onCacheStatisticsClear(CacheStatisticsClearMessage msg) {
        assert msg != null;

        if (msg.initial()) {
            EnableStatisticsFuture fut = manageStatisticsFuts.get(msg.requestId());

            if (fut != null && !cacheNames().containsAll(msg.caches())) {
                fut.onDone(new IgniteCheckedException("One or more cache descriptors not found [caches="
                    + caches + ']'));

                return;
            }

            for (String cacheName : msg.caches()) {
                IgniteInternalCache<?, ?> cache = ctx.cache().cache(cacheName);

                if (cache != null)
                    cache.localMxBean().clear();
                else
                    log.warning("Failed to clear cache statistics, cache not found [cacheName="
                        + cacheName + ']');
            }
        }
        else {
            EnableStatisticsFuture fut = manageStatisticsFuts.get(msg.requestId());

            if (fut != null)
                fut.onDone();
        }
    }

    /**
     * Cache statistics flag change task processed by exchange worker.
     *
     * @param msg Message.
     */
    public void processStatisticsModeChange(CacheStatisticsModeChangeMessage msg) {
        assert msg != null;

        for (String cacheName : msg.caches()) {
            IgniteInternalCache<Object, Object> cache = cache(cacheName);

            if (cache != null)
                cache.context().statisticsEnabled(msg.enabled());
            else
                log.warning("Failed to change cache configuration, cache not found [cacheName=" + cacheName + ']');
        }
    }

    /**
     * @param stoppedCaches Stopped caches.
     */
    private void stopCachesOnClientReconnect(Collection<GridCacheAdapter> stoppedCaches) {
        assert ctx.discovery().localNode().isClient();

        for (GridCacheAdapter cache : stoppedCaches) {
            CacheGroupContext grp = cache.context().group();

            onKernalStop(cache, true);
            stopCache(cache, true, false, false);

            sharedCtx.affinity().stopCacheOnReconnect(cache.context());

            if (!grp.hasCaches()) {
                stopCacheGroup(grp, false);

                sharedCtx.affinity().stopCacheGroupOnReconnect(grp);
            }
        }
    }

    /**
     * Dynamically starts cache using template configuration.
     *
     * @param cacheName Cache name.
     * @return Future that will be completed when cache is deployed.
     */
    public IgniteInternalFuture<?> createFromTemplate(String cacheName) {
        try {
            CacheConfiguration cfg = getOrCreateConfigFromTemplate(cacheName);

            return dynamicStartCache(cfg, cacheName, null, true, true, true);
        }
        catch (IgniteCheckedException e) {
            throw U.convertException(e);
        }
    }

    /**
     * Dynamically starts cache using template configuration.
     *
     * @param cacheName Cache name.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @return Future that will be completed when cache is deployed.
     */
    public IgniteInternalFuture<?> getOrCreateFromTemplate(String cacheName, boolean checkThreadTx) {
        return getOrCreateFromTemplate(cacheName, cacheName, null, checkThreadTx);
    }

    /**
     * Dynamically starts cache using template configuration.
     *
     * @param cacheName Cache name.
     * @param templateName Cache template name.
     * @param cfgOverride Cache config properties to override.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @return Future that will be completed when cache is deployed.
     */
    public IgniteInternalFuture<?> getOrCreateFromTemplate(String cacheName, String templateName,
        CacheConfigurationOverride cfgOverride, boolean checkThreadTx) {
        assert cacheName != null;

        try {
            if (publicJCache(cacheName, false, checkThreadTx) != null) // Cache with given name already started.
                return new GridFinishedFuture<>();

            CacheConfiguration ccfg = F.isEmpty(templateName)
                ? getOrCreateConfigFromTemplate(cacheName)
                : getOrCreateConfigFromTemplate(templateName);

            ccfg.setName(cacheName);

            if (cfgOverride != null)
                cfgOverride.apply(ccfg);

            return dynamicStartCache(ccfg, cacheName, null, false, true, checkThreadTx);
        }
        catch (IgniteCheckedException e) {
            return new GridFinishedFuture<>(e);
        }
    }

    /**
     * @param cacheName Cache name.
     * @return Cache configuration, or {@code null} if no template with matching name found.
     * @throws IgniteCheckedException If failed.
     */
    public CacheConfiguration getConfigFromTemplate(String cacheName) throws IgniteCheckedException {
        DynamicCacheDescriptor cfgTemplate = null;

        DynamicCacheDescriptor dfltCacheCfg = null;

        List<DynamicCacheDescriptor> wildcardNameCfgs = null;

        for (DynamicCacheDescriptor desc : cachesInfo.registeredTemplates().values()) {
            assert desc.template();

            CacheConfiguration cfg = desc.cacheConfiguration();

            assert cfg != null;

            if (F.eq(cacheName, cfg.getName())) {
                cfgTemplate = desc;

                break;
            }

            if (cfg.getName() != null) {
                if (GridCacheUtils.isCacheTemplateName(cfg.getName())) {
                    if (cfg.getName().length() > 1) {
                        if (wildcardNameCfgs == null)
                            wildcardNameCfgs = new ArrayList<>();

                        wildcardNameCfgs.add(desc);
                    }
                    else
                        dfltCacheCfg = desc; // Template with name '*'.
                }
            }
            else if (dfltCacheCfg == null)
                dfltCacheCfg = desc;
        }

        if (cfgTemplate == null && cacheName != null && wildcardNameCfgs != null) {
            wildcardNameCfgs.sort((a, b) ->
                Integer.compare(b.cacheConfiguration().getName().length(), a.cacheConfiguration().getName().length()));

            for (DynamicCacheDescriptor desc : wildcardNameCfgs) {
                String wildcardCacheName = desc.cacheConfiguration().getName();

                if (cacheName.startsWith(wildcardCacheName.substring(0, wildcardCacheName.length() - 1))) {
                    cfgTemplate = desc;

                    break;
                }
            }
        }

        if (cfgTemplate == null)
            cfgTemplate = dfltCacheCfg;

        if (cfgTemplate == null)
            return null;

        // It's safe to enrich cache configuration here because we requested this cache from current node.
        CacheConfiguration enrichedTemplate = enricher().enrichFully(
            cfgTemplate.cacheConfiguration(), cfgTemplate.cacheConfigurationEnrichment());

        enrichedTemplate = cloneCheckSerializable(enrichedTemplate);

        CacheConfiguration cfg = new CacheConfiguration(enrichedTemplate);

        cfg.setName(cacheName);

        return cfg;
    }

    /**
     * @param cacheName Cache name.
     * @return Cache configuration.
     * @throws IgniteCheckedException If failed.
     */
    private CacheConfiguration getOrCreateConfigFromTemplate(String cacheName) throws IgniteCheckedException {
        CacheConfiguration cfg = getConfigFromTemplate(cacheName);

        return cfg != null ? cfg : new CacheConfiguration(cacheName);
    }

    /**
     * Dynamically starts cache.
     *
     * @param ccfg Cache configuration.
     * @param cacheName Cache name.
     * @param nearCfg Near cache configuration.
     * @param failIfExists Fail if exists flag.
     * @param failIfNotStarted If {@code true} fails if cache is not started.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @return Future that will be completed when cache is deployed.
     */
    public IgniteInternalFuture<Boolean> dynamicStartCache(
        @Nullable CacheConfiguration ccfg,
        String cacheName,
        @Nullable NearCacheConfiguration nearCfg,
        boolean failIfExists,
        boolean failIfNotStarted,
        boolean checkThreadTx
    ) {
        return dynamicStartCache(ccfg,
            cacheName,
            nearCfg,
            CacheType.USER,
            false,
            failIfExists,
            failIfNotStarted,
            checkThreadTx);
    }

    /**
     * Dynamically starts cache as a result of SQL {@code CREATE TABLE} command.
     *
     * @param ccfg Cache configuration.
     */
    public IgniteInternalFuture<Boolean> dynamicStartSqlCache(
        CacheConfiguration ccfg
    ) {
        A.notNull(ccfg, "ccfg");

        return dynamicStartCache(ccfg,
            ccfg.getName(),
            ccfg.getNearConfiguration(),
            CacheType.USER,
            true,
            false,
            true,
            true);
    }

    /**
     * Dynamically starts cache.
     *
     * @param ccfg Cache configuration.
     * @param cacheName Cache name.
     * @param nearCfg Near cache configuration.
     * @param cacheType Cache type.
     * @param sql If the cache needs to be created as the result of SQL {@code CREATE TABLE} command.
     * @param failIfExists Fail if exists flag.
     * @param failIfNotStarted If {@code true} fails if cache is not started.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @return Future that will be completed when cache is deployed.
     */
    public IgniteInternalFuture<Boolean> dynamicStartCache(
        @Nullable CacheConfiguration ccfg,
        String cacheName,
        @Nullable NearCacheConfiguration nearCfg,
        CacheType cacheType,
        boolean sql,
        boolean failIfExists,
        boolean failIfNotStarted,
        boolean checkThreadTx
    ) {
        assert cacheName != null;

        if (checkThreadTx) {
            sharedCtx.tm().checkEmptyTransactions(
                () -> format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheName, "dynamicStartCache"));
        }

        GridPlainClosure2<Collection<byte[]>, byte[], IgniteInternalFuture<Boolean>> startCacheClsr =
            (grpKeys, masterKeyDigest) -> {
                assert ccfg == null || !ccfg.isEncryptionEnabled() || !grpKeys.isEmpty();

                DynamicCacheChangeRequest req = prepareCacheChangeRequest(
                    ccfg,
                    cacheName,
                    nearCfg,
                    cacheType,
                    sql,
                    failIfExists,
                    failIfNotStarted,
                    null,
                    false,
                    null,
                    ccfg != null && ccfg.isEncryptionEnabled() ? grpKeys.iterator().next() : null,
                    null,
                    ccfg != null && ccfg.isEncryptionEnabled() ? masterKeyDigest : null);

                if (req != null) {
                    if (req.clientStartOnly())
                        return startClientCacheChange(F.asMap(req.cacheName(), req), null);

                    return F.first(initiateCacheChanges(F.asList(req)));
                }
                else
                    return new GridFinishedFuture<>();
            };

        try {
            if (ccfg != null && ccfg.isEncryptionEnabled()) {
                ctx.encryption().checkEncryptedCacheSupported();

                return generateEncryptionKeysAndStartCacheAfter(1, startCacheClsr);
            }

            return startCacheClsr.apply(Collections.EMPTY_SET, null);
        }
        catch (Exception e) {
            return new GridFinishedFuture<>(e);
        }
    }

    /**
     * Checks that cluster in a {@link ClusterState#ACTIVE_READ_ONLY} state.
     *
     * @param opName Operation name.
     * @param cfgs Stored cache configurations.
     * @throws CacheException If cluster in a {@link ClusterState#ACTIVE_READ_ONLY} state.
     */
    private void checkReadOnlyState(String opName, Collection<StoredCacheData> cfgs) {
        IgniteOutClosure<String> cacheNameClo = null;
        IgniteOutClosure<String> cacheGrpNameClo = null;

        if (!F.isEmpty(cfgs)) {
            if (cfgs.size() == 1) {
                CacheConfiguration cfg = cfgs.iterator().next().config();

                cacheNameClo = cfg::getName;
                cacheGrpNameClo = cfg::getGroupName;
            }
            else {
                cacheNameClo = () -> cfgs.stream()
                    .map(StoredCacheData::config)
                    .map(CacheConfiguration::getName)
                    .collect(toList()).toString();

                cacheGrpNameClo = () -> cfgs.stream()
                    .map(StoredCacheData::config)
                    .map(CacheConfiguration::getGroupName)
                    .collect(toList()).toString();
            }
        }

        checkReadOnlyState(opName, cacheGrpNameClo, cacheNameClo);
    }

    /**
     * Checks that cluster in a {@link ClusterState#ACTIVE_READ_ONLY} state.
     *
     * @param opName Operation name.
     * @param cfgs Cache configurations.
     * @throws CacheException If cluster in a {@link ClusterState#ACTIVE_READ_ONLY} state.
     */
    public void checkReadOnlyState(String opName, CacheConfiguration... cfgs) {
        IgniteOutClosure<String> cacheNameClo = null;
        IgniteOutClosure<String> cacheGrpNameClo = null;

        if (!F.isEmpty(cfgs)) {
            if (cfgs.length == 1) {
                cacheNameClo = () -> cfgs[0] == null ? null : cfgs[0].getName();
                cacheGrpNameClo = () -> cfgs[0] == null ? null : cfgs[0].getGroupName();
            }
            else {
                cacheNameClo = () -> Stream.of(cfgs)
                    .map(CacheConfiguration::getName)
                    .collect(toList()).toString();

                cacheGrpNameClo = () -> Stream.of(cfgs)
                    .map(CacheConfiguration::getGroupName)
                    .collect(toList()).toString();
            }
        }

        checkReadOnlyState(opName, cacheGrpNameClo, cacheNameClo);
    }

    /**
     * Checks that cluster in a {@link ClusterState#ACTIVE_READ_ONLY} state.
     *
     * @param opName Operation name.
     * @param cacheGrpNameClo Closure for getting cache group name, if it needed (optional).
     * @param cacheNameClo Closure for getting cache name, if it needed (optional).
     * @throws CacheException If cluster in a {@link ClusterState#ACTIVE_READ_ONLY} state.
     */
    private void checkReadOnlyState(
        String opName,
        @Nullable IgniteOutClosure<String> cacheGrpNameClo,
        @Nullable IgniteOutClosure<String> cacheNameClo
    ) {
        if (sharedCtx.readOnlyMode()) {
            String cacheName = cacheNameClo == null ? null : cacheNameClo.apply();
            String cacheGrpName = cacheGrpNameClo == null ? null : cacheGrpNameClo.apply();

            String errorMsg = format(CLUSTER_READ_ONLY_MODE_ERROR_MSG_FORMAT, opName, cacheGrpName, cacheName);

            throw new CacheException(new IgniteClusterReadOnlyException(errorMsg));
        }
    }

    /**
     * Send {@code GenerateEncryptionKeyRequest} and execute {@code after} closure if succeed.
     *
     * @param keyCnt Count of keys to generate.
     * @param after Closure to execute after encryption keys would be generated.
     */
    private IgniteInternalFuture<Boolean> generateEncryptionKeysAndStartCacheAfter(int keyCnt,
        GridPlainClosure2<Collection<byte[]>, byte[], IgniteInternalFuture<Boolean>> after) {
        IgniteInternalFuture<T2<Collection<byte[]>, byte[]>> genEncKeyFut = ctx.encryption().generateKeys(keyCnt);

        GridFutureAdapter<Boolean> res = new GridFutureAdapter<>();

        genEncKeyFut.listen(new IgniteInClosure<IgniteInternalFuture<T2<Collection<byte[]>, byte[]>>>() {
            @Override public void apply(IgniteInternalFuture<T2<Collection<byte[]>, byte[]>> fut) {
                try {
                    Collection<byte[]> grpKeys = fut.result().get1();
                    byte[] masterKeyDigest = fut.result().get2();

                    if (F.size(grpKeys, F.alwaysTrue()) != keyCnt)
                        res.onDone(false, fut.error());

                    IgniteInternalFuture<Boolean> dynStartCacheFut = after.apply(grpKeys, masterKeyDigest);

                    dynStartCacheFut.listen(new IgniteInClosure<IgniteInternalFuture<Boolean>>() {
                        @Override public void apply(IgniteInternalFuture<Boolean> fut) {
                            try {
                                res.onDone(fut.get(), fut.error());
                            }
                            catch (IgniteCheckedException e) {
                                res.onDone(false, e);
                            }
                        }
                    });
                }
                catch (Exception e) {
                    res.onDone(false, e);
                }
            }
        });

        return res;
    }

    /**
     * @param startReqs Start requests.
     * @param cachesToClose Cache tp close.
     * @return Future for cache change operation.
     */
    private IgniteInternalFuture<Boolean> startClientCacheChange(
        @Nullable Map<String, DynamicCacheChangeRequest> startReqs, @Nullable Set<String> cachesToClose) {
        assert startReqs != null ^ cachesToClose != null;

        DynamicCacheStartFuture fut = new DynamicCacheStartFuture(UUID.randomUUID());

        IgniteInternalFuture old = pendingFuts.put(fut.id, fut);

        assert old == null : old;

        ctx.discovery().clientCacheStartEvent(fut.id, startReqs, cachesToClose);

        IgniteCheckedException err = checkNodeState();

        if (err != null)
            fut.onDone(err);

        return fut;
    }

    /**
     * Dynamically starts multiple caches.
     *
     * @param ccfgList Collection of cache configuration.
     * @param failIfExists Fail if exists flag.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @param disabledAfterStart If true, cache proxies will be only activated after {@link #restartProxies()}.
     * @return Future that will be completed when all caches are deployed.
     */
    public IgniteInternalFuture<Boolean> dynamicStartCaches(
        Collection<CacheConfiguration> ccfgList,
        boolean failIfExists,
        boolean checkThreadTx,
        boolean disabledAfterStart
    ) {
        return dynamicStartCachesByStoredConf(
            ccfgList.stream().map(StoredCacheData::new).collect(toList()),
            failIfExists,
            checkThreadTx,
            disabledAfterStart,
            null);
    }

    /**
     * Dynamically starts multiple caches.
     *
     * @param storedCacheDataList Collection of stored cache data.
     * @param failIfExists Fail if exists flag.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @param disabledAfterStart If true, cache proxies will be only activated after {@link #restartProxies()}.
     * @param restartId Restart requester id (it'll allow to start this cache only him).
     * @return Future that will be completed when all caches are deployed.
     */
    public IgniteInternalFuture<Boolean> dynamicStartCachesByStoredConf(
        Collection<StoredCacheData> storedCacheDataList,
        boolean failIfExists,
        boolean checkThreadTx,
        boolean disabledAfterStart,
        IgniteUuid restartId
    ) {
        if (checkThreadTx) {
            sharedCtx.tm().checkEmptyTransactions(() -> {
                List<String> cacheNames = storedCacheDataList.stream()
                    .map(StoredCacheData::config)
                    .map(CacheConfiguration::getName)
                    .collect(toList());

                return format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheNames, "dynamicStartCachesByStoredConf");
            });
        }

        GridPlainClosure2<Collection<byte[]>, byte[], IgniteInternalFuture<Boolean>> startCacheClsr = (grpKeys, masterKeyDigest) -> {
            List<DynamicCacheChangeRequest> srvReqs = null;
            Map<String, DynamicCacheChangeRequest> clientReqs = null;

            Iterator<byte[]> grpKeysIter = grpKeys.iterator();

            for (StoredCacheData ccfg : storedCacheDataList) {
                assert ccfg.groupKeyEncrypted() == null || ccfg.config().isEncryptionEnabled();

                // Reuse encription key if passed for this group. Take next generated otherwise.
                GroupKeyEncrypted encrKey = ccfg.config().isEncryptionEnabled() ? (ccfg.groupKeyEncrypted() != null ?
                    ccfg.groupKeyEncrypted() : new GroupKeyEncrypted(0, grpKeysIter.next())) : null;

                DynamicCacheChangeRequest req = prepareCacheChangeRequest(
                    ccfg.config(),
                    ccfg.config().getName(),
                    null,
                    resolveCacheType(ccfg.config()),
                    ccfg.sql(),
                    failIfExists,
                    true,
                    restartId,
                    disabledAfterStart,
                    ccfg.queryEntities(),
                    encrKey != null ? encrKey.key() : null,
                    encrKey != null ? encrKey.id() : null,
                    encrKey != null ? masterKeyDigest : null
                );

                if (req != null) {
                    if (req.clientStartOnly()) {
                        if (clientReqs == null)
                            clientReqs = U.newLinkedHashMap(storedCacheDataList.size());

                        clientReqs.put(req.cacheName(), req);
                    }
                    else {
                        if (srvReqs == null)
                            srvReqs = new ArrayList<>(storedCacheDataList.size());

                        srvReqs.add(req);
                    }
                }
            }

            if (srvReqs == null && clientReqs == null)
                return new GridFinishedFuture<>();

            if (clientReqs != null && srvReqs == null)
                return startClientCacheChange(clientReqs, null);

            GridCompoundFuture<?, Boolean> compoundFut = new GridCompoundFuture<>();

            for (DynamicCacheStartFuture fut : initiateCacheChanges(srvReqs))
                compoundFut.add((IgniteInternalFuture)fut);

            if (clientReqs != null) {
                IgniteInternalFuture<Boolean> clientStartFut = startClientCacheChange(clientReqs, null);

                compoundFut.add((IgniteInternalFuture)clientStartFut);
            }

            compoundFut.markInitialized();

            return compoundFut;
        };

        int encGrpCnt = 0;

        for (StoredCacheData ccfg : storedCacheDataList) {
            if (ccfg.config().isEncryptionEnabled())
                encGrpCnt++;
        }

        return generateEncryptionKeysAndStartCacheAfter(encGrpCnt, startCacheClsr);
    }

    /** Resolve cache type for input cacheType */
    private @NotNull CacheType resolveCacheType(CacheConfiguration ccfg) {
        if (CU.isUtilityCache(ccfg.getName()))
            return CacheType.UTILITY;
        else if (internalCaches.contains(ccfg.getName()))
            return CacheType.INTERNAL;
        else if (DataStructuresProcessor.isDataStructureCache(ccfg.getName()))
            return CacheType.DATA_STRUCTURES;
        else
            return CacheType.USER;
    }

    /**
     * @param cacheName Cache name to destroy.
     * @param sql If the cache needs to be destroyed only if it was created as the result of SQL {@code CREATE TABLE}
     * command.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @param restart Restart flag.
     * @param restartId Restart requester id (it'll allow to start this cache only him).
     * @return Future that will be completed when cache is destroyed.
     */
    public IgniteInternalFuture<Boolean> dynamicDestroyCache(
        String cacheName,
        boolean sql,
        boolean checkThreadTx,
        boolean restart,
        IgniteUuid restartId
    ) {
        assert cacheName != null;

        if (checkThreadTx) {
            sharedCtx.tm().checkEmptyTransactions(
                () -> format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheName, "dynamicDestroyCache"));
        }

        checkReadOnlyState("dynamic destroy cache", null, cacheName::toString);

        DynamicCacheChangeRequest req = DynamicCacheChangeRequest.stopRequest(ctx, cacheName, sql, true);

        req.stop(true);
        req.destroy(true);
        req.restart(restart);
        req.restartId(restartId);

        return F.first(initiateCacheChanges(F.asList(req)));
    }

    /**
     * @param cacheNames Collection of cache names to destroy.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @return Future that will be completed when cache is destroyed.
     */
    public IgniteInternalFuture<?> dynamicDestroyCaches(Collection<String> cacheNames, boolean checkThreadTx) {
        return dynamicDestroyCaches(cacheNames, checkThreadTx, true);
    }

    /**
     * @param cacheNames Collection of cache names to destroy.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @param destroy Cache data destroy flag. Setting to <code>true</code> will cause removing all cache data
     * @return Future that will be completed when cache is destroyed.
     */
    public IgniteInternalFuture<?> dynamicDestroyCaches(
        Collection<String> cacheNames,
        boolean checkThreadTx,
        boolean destroy
    ) {
        if (checkThreadTx) {
            sharedCtx.tm().checkEmptyTransactions(
                () -> format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheNames, "dynamicDestroyCaches"));
        }

        if (!F.isEmpty(cacheNames))
            checkReadOnlyState("dynamic destroy caches", null, cacheNames::toString);

        return dynamicChangeCaches(
            cacheNames.stream().map(cacheName -> createStopRequest(cacheName, false, null, destroy))
                .collect(toList())
        );
    }

    /**
     * Prepares cache stop request.
     *
     * @param cacheName Cache names to destroy.
     * @param restart Restart flag.
     * @param restartId Restart requester id (it'll allow to start this cache only him).
     * @param destroy Cache data destroy flag. Setting to {@code true} will cause removing all cache data from store.
     * @return Future that will be completed when cache is destroyed.
     */
    public @NotNull DynamicCacheChangeRequest createStopRequest(String cacheName, boolean restart, IgniteUuid restartId, boolean destroy) {
        DynamicCacheChangeRequest req = DynamicCacheChangeRequest.stopRequest(ctx, cacheName, false, true);

        req.stop(true);
        req.destroy(destroy);
        req.restart(restart);
        req.restartId(restartId);

        return req;
    }

    /**
     * Starts cache stop request as cache change batch.
     *
     * @param reqs cache stop requests.
     * @return compound future.
     */
    @NotNull public IgniteInternalFuture<?> dynamicChangeCaches(List<DynamicCacheChangeRequest> reqs) {
        return initiateCacheChanges(reqs).stream().collect(IgniteCollectors.toCompoundFuture());
    }

    /**
     * @param cacheName Cache name to close.
     * @return Future that will be completed when cache is closed.
     */
    IgniteInternalFuture<?> dynamicCloseCache(String cacheName) {
        assert cacheName != null;

        IgniteCacheProxy<?, ?> proxy = jcacheProxy(cacheName, false);

        if (proxy == null || proxy.isProxyClosed())
            return new GridFinishedFuture<>(); // No-op.

        sharedCtx.tm().checkEmptyTransactions(
            () -> format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheName, "dynamicCloseCache"));

        return startClientCacheChange(null, Collections.singleton(cacheName));
    }

    /**
     * Resets cache state after the cache has been moved to recovery state.
     *
     * @param cacheNames Cache names.
     * @return Future that will be completed when state is changed for all caches.
     */
    public IgniteInternalFuture<?> resetCacheState(Collection<String> cacheNames) throws ClusterTopologyCheckedException {
        sharedCtx.tm().checkEmptyTransactions(
            () -> format(CHECK_EMPTY_TRANSACTIONS_ERROR_MSG_FORMAT, cacheNames, "resetCacheState"));

        Collection<DynamicCacheChangeRequest> reqs = new ArrayList<>(cacheNames.size());

        for (String cacheName : cacheNames) {
            final IgniteInternalCache<Object, Object> cache0 = internalCache(cacheName);

            if (cache0 == null)
                continue;

            // Check if all lost partitions has at least one affinity owner.
            final Collection<Integer> lostParts = cache0.lostPartitions();

            if (lostParts.isEmpty())
                continue;

            for (Integer part : lostParts) {
                final Collection<ClusterNode> owners = cache0.affinity().mapPartitionToPrimaryAndBackups(part);

                if (owners.isEmpty())
                    throw new ClusterTopologyCheckedException("Cannot reset lost partitions because no baseline nodes " +
                        "are online [cache=" + cacheName + ", partition=" + part + ']');
            }

            DynamicCacheChangeRequest req = DynamicCacheChangeRequest.resetLostPartitions(ctx, cacheName);

            reqs.add(req);
        }

        return initiateCacheChanges(reqs).stream().collect(IgniteCollectors.toCompoundFuture());
    }

    /**
     * @param cacheName Cache name.
     * @return Cache type.
     */
    public CacheType cacheType(String cacheName) {
        if (CU.isUtilityCache(cacheName))
            return CacheType.UTILITY;
        else if (internalCaches.contains(cacheName))
            return CacheType.INTERNAL;
        else if (DataStructuresProcessor.isDataStructureCache(cacheName))
            return CacheType.DATA_STRUCTURES;
        else
            return CacheType.USER;
    }

    /**
     * @return {@code True} if cache group {@code cacheGrpId} is encrypted. {@code False} otherwise.
     */
    public boolean isEncrypted(int cacheGrpId) {
        return cacheGrpId != MetaStorage.METASTORAGE_CACHE_ID && cacheGroup(cacheGrpId).config().isEncryptionEnabled();
    }

    /**
     * Save cache configuration to persistent store if necessary.
     *
     * @param desc Cache descriptor.
     */
    public void saveCacheConfiguration(DynamicCacheDescriptor desc) throws IgniteCheckedException {
        assert desc != null;

        locCfgMgr.saveCacheConfiguration(desc.toStoredData(splitter), true);
    }

    /**
     * Remove all persistent files for all registered caches.
     */
    public void cleanupCachesDirectories() throws IgniteCheckedException {
        if (sharedCtx.pageStore() == null || sharedCtx.kernalContext().clientNode())
            return;

        for (DynamicCacheDescriptor desc : cacheDescriptors().values()) {
            if (isPersistentCache(desc.cacheConfiguration(), sharedCtx.gridConfig().getDataStorageConfiguration()))
                sharedCtx.pageStore().cleanupPersistentSpace(desc.cacheConfiguration());
        }
    }

    /**
     * @param reqs Requests.
     * @return Collection of futures.
     */
    private Collection<DynamicCacheStartFuture> initiateCacheChanges(
        Collection<DynamicCacheChangeRequest> reqs
    ) {
        Collection<DynamicCacheStartFuture> res = new ArrayList<>(reqs.size());

        Collection<DynamicCacheChangeRequest> sndReqs = new ArrayList<>(reqs.size());

        for (DynamicCacheChangeRequest req : reqs) {
            authorizeCacheChange(ctx.security(), req);

            DynamicCacheStartFuture fut = new DynamicCacheStartFuture(req.requestId());

            try {
                if (req.stop()) {
                    DynamicCacheDescriptor desc = cacheDescriptor(req.cacheName());

                    if (desc == null)
                        // No-op.
                        fut.onDone(false);
                }

                if (req.start() && req.startCacheConfiguration() != null) {
                    CacheConfiguration ccfg = req.startCacheConfiguration();

                    try {
                        cachesInfo.validateStartCacheConfiguration(ccfg);
                    }
                    catch (IgniteCheckedException e) {
                        fut.onDone(e);
                    }
                }

                if (fut.isDone())
                    continue;

                DynamicCacheStartFuture old = (DynamicCacheStartFuture)pendingFuts.putIfAbsent(
                    req.requestId(), fut);

                assert old == null;

                if (fut.isDone())
                    continue;

                sndReqs.add(req);
            }
            catch (Exception e) {
                fut.onDone(e);
            }
            finally {
                res.add(fut);
            }
        }

        Exception err = null;

        if (!sndReqs.isEmpty()) {
            try {
                ctx.discovery().sendCustomEvent(new DynamicCacheChangeBatch(sndReqs));

                err = checkNodeState();
            }
            catch (IgniteCheckedException e) {
                err = e;
            }
        }

        if (err != null) {
            for (DynamicCacheStartFuture fut : res)
                fut.onDone(err);
        }

        return res;
    }

    /**
     * Authorizes cache change request.
     *
     * @param security Security.
     * @param req Cache change request.
     */
    static void authorizeCacheChange(IgniteSecurity security, DynamicCacheChangeRequest req) {
        if (req.cacheType() == null || req.cacheType() == CacheType.USER) {
            if (req.start())
                authorizeCacheCreate(security, req.startCacheConfiguration());
            else if (req.stop())
                authorizeCacheDestroy(security, req.cacheName());
        }
    }

    /**
     * Authorizes cache destroy.
     *
     * @param security Security.
     * @param cacheName Cache name.
     */
    static void authorizeCacheDestroy(IgniteSecurity security, String cacheName) {
        security.authorize(cacheName, SecurityPermission.CACHE_DESTROY);
    }

    /**
     * Authorizes cache create.
     *
     * @param security Security.
     * @param cacheCfg Cache configuration.
     */
    static void authorizeCacheCreate(IgniteSecurity security, @Nullable CacheConfiguration cacheCfg) {
        if (cacheCfg == null)
            return;

        security.authorize(cacheCfg.getName(), SecurityPermission.CACHE_CREATE);

        if (cacheCfg.isOnheapCacheEnabled() &&
                IgniteSystemProperties.getBoolean(IgniteSystemProperties.IGNITE_DISABLE_ONHEAP_CACHE))
            throw new SecurityException("Authorization failed for enabling on-heap cache.");
    }

    /**
     * @return Non null exception if node is stopping or disconnected.
     */
    private @Nullable IgniteCheckedException checkNodeState() {
        if (ctx.isStopping()) {
            return new IgniteCheckedException("Failed to execute dynamic cache change request, " +
                "node is stopping.");
        }
        else if (ctx.clientDisconnected()) {
            return new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
                "Failed to execute dynamic cache change request, client node disconnected.");
        }

        return null;
    }

    /**
     * @param type Event type.
     * @param customMsg Custom message instance.
     * @param node Event node.
     * @param topVer Topology version.
     * @param state Cluster state.
     */
    public void onDiscoveryEvent(
        int type,
        @Nullable DiscoveryCustomMessage customMsg,
        ClusterNode node,
        AffinityTopologyVersion topVer,
        DiscoveryDataClusterState state
    ) {
        cachesInfo.onDiscoveryEvent(type, node, topVer);

        sharedCtx.affinity().onDiscoveryEvent(type, customMsg, node, topVer, state);
    }

    /**
     * Callback invoked from discovery thread when discovery custom message is received.
     *
     * @param msg Customer message.
     * @param topVer Current topology version.
     * @param node Node sent message.
     * @return {@code True} if minor topology version should be increased.
     */
    public boolean onCustomEvent(DiscoveryCustomMessage msg, AffinityTopologyVersion topVer, ClusterNode node) {
        if (msg instanceof SchemaAbstractDiscoveryMessage) {
            ctx.query().onDiscovery((SchemaAbstractDiscoveryMessage)msg);

            return false;
        }

        if (msg instanceof CacheAffinityChangeMessage)
            return sharedCtx.affinity().onCustomEvent(((CacheAffinityChangeMessage)msg));

        if (msg instanceof SnapshotDiscoveryMessage &&
            ((SnapshotDiscoveryMessage)msg).needExchange())
            return true;

        if (msg instanceof WalStateAbstractMessage) {
            WalStateAbstractMessage msg0 = (WalStateAbstractMessage)msg;

            if (msg0 instanceof WalStateProposeMessage)
                sharedCtx.walState().onProposeDiscovery((WalStateProposeMessage)msg);
            else if (msg0 instanceof WalStateFinishMessage)
                sharedCtx.walState().onFinishDiscovery((WalStateFinishMessage)msg);

            return msg0.needExchange();
        }

        if (msg instanceof DynamicCacheChangeBatch) {
            boolean changeRequested = cachesInfo.onCacheChangeRequested((DynamicCacheChangeBatch)msg, topVer);

            ctx.query().onCacheChangeRequested((DynamicCacheChangeBatch)msg);

            return changeRequested;
        }

        if (msg instanceof DynamicCacheChangeFailureMessage)
            cachesInfo.onCacheChangeRequested((DynamicCacheChangeFailureMessage)msg, topVer);

        if (msg instanceof ClientCacheChangeDiscoveryMessage)
            cachesInfo.onClientCacheChange((ClientCacheChangeDiscoveryMessage)msg, node);

        if (msg instanceof CacheStatisticsModeChangeMessage)
            onCacheStatisticsModeChange((CacheStatisticsModeChangeMessage)msg);

        if (msg instanceof CacheStatisticsClearMessage)
            onCacheStatisticsClear((CacheStatisticsClearMessage)msg);

        if (msg instanceof TxTimeoutOnPartitionMapExchangeChangeMessage)
            sharedCtx.tm().onTxTimeoutOnPartitionMapExchangeChange((TxTimeoutOnPartitionMapExchangeChangeMessage)msg);

        return false;
    }

    /** {@inheritDoc} */
    @Override public @Nullable IgniteNodeValidationResult validateNode(ClusterNode node) {
        IgniteNodeValidationResult res = validateHashIdResolvers(node, ctx, cacheDescriptors());

        if (res == null)
            res = validateRestartingCaches(node);

        if (res == null)
            res = validateRestoringCaches(node);

        return res;
    }

    /**
     * @param cacheName Cache to check.
     * @return Cache is under restarting.
     */
    public boolean isCacheRestarting(String cacheName) {
        return cachesInfo.isRestarting(cacheName);
    }

    /**
     * @param node Joining node to validate.
     * @return Node validation result if there was an issue with the joining node, {@code null} otherwise.
     */
    private IgniteNodeValidationResult validateRestartingCaches(ClusterNode node) {
        if (cachesInfo.hasRestartingCaches()) {
            String msg = "Joining node during caches restart is not allowed [joiningNodeId=" + node.id() +
                ", restartingCaches=" + new HashSet<>(cachesInfo.restartingCaches()) + ']';

            return new IgniteNodeValidationResult(node.id(), msg);
        }

        return null;
    }

    /**
     * @param node Joining node to validate.
     * @return Node validation result if there was an issue with the joining node, {@code null} otherwise.
     */
    private IgniteNodeValidationResult validateRestoringCaches(ClusterNode node) {
        if (ctx.cache().context().snapshotMgr().isRestoring()) {
            String msg = "Joining node during caches restore is not allowed [joiningNodeId=" + node.id() + ']';

            return new IgniteNodeValidationResult(node.id(), msg);
        }

        return null;
    }

    /**
     * @return Keep static cache configuration flag. If {@code true}, static cache configuration will override
     * configuration persisted on disk.
     */
    public boolean keepStaticCacheConfiguration() {
        return keepStaticCacheConfiguration;
    }

    /**
     * @param name Cache name.
     * @param <K> type of keys.
     * @param <V> type of values.
     * @return Cache instance for given name.
     */
    public <K, V> @Nullable IgniteInternalCache<K, V> cache(String name) {
        assert name != null;

        if (log.isDebugEnabled())
            log.debug("Getting cache for name: " + name);

        IgniteCacheProxy<K, V> jcache = (IgniteCacheProxy<K, V>)jcacheProxy(name, true);

        return jcache == null ? null : jcache.internalProxy();
    }

    /**
     * Await proxy initialization.
     *
     * @param jcache Cache proxy.
     */
    private void awaitInitializeProxy(IgniteCacheProxyImpl<?, ?> jcache) {
        if (jcache != null) {
            CountDownLatch initLatch = jcache.getInitLatch();

            try {
                while (initLatch.getCount() > 0) {
                    initLatch.await(2000, TimeUnit.MILLISECONDS);

                    if (log.isInfoEnabled())
                        log.info("Failed to wait proxy initialization, cache=" + jcache.getName() +
                            ", localNodeId=" + ctx.localNodeId());
                }
            }
            catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                // Ignore intteruption.
            }
        }
    }

    /**
     * @param name Cache name.
     */
    public void completeProxyInitialize(String name) {
        IgniteCacheProxyImpl<?, ?> jcache = jCacheProxies.get(name);

        if (jcache != null) {
            CountDownLatch proxyInitLatch = jcache.getInitLatch();

            if (proxyInitLatch.getCount() > 0) {
                if (log.isInfoEnabled())
                    log.info("Finish proxy initialization, cacheName=" + name +
                        ", localNodeId=" + ctx.localNodeId());

                proxyInitLatch.countDown();
            }
        }
        else {
            if (log.isInfoEnabled())
                log.info("Can not finish proxy initialization because proxy does not exist, cacheName=" + name +
                    ", localNodeId=" + ctx.localNodeId());
        }
    }

    /**
     * @param name Cache name.
     * @return Cache instance for given name.
     * @throws IgniteCheckedException If failed.
     */
    public <K, V> IgniteInternalCache<K, V> getOrStartCache(String name) throws IgniteCheckedException {
        return getOrStartCache(name, null);
    }

    /**
     * @param name Cache name.
     * @return Cache instance for given name.
     * @throws IgniteCheckedException If failed.
     */
    public <K, V> IgniteInternalCache<K, V> getOrStartCache(
        String name,
        CacheConfiguration ccfg
    ) throws IgniteCheckedException {
        assert name != null;

        if (log.isDebugEnabled())
            log.debug("Getting cache for name: " + name);

        IgniteCacheProxy<?, ?> cache = jcacheProxy(name, true);

        if (cache == null) {
            dynamicStartCache(ccfg, name, null, false, ccfg == null, true).get();

            cache = jcacheProxy(name, true);
        }

        return cache == null ? null : (IgniteInternalCache<K, V>)cache.internalProxy();
    }

    /**
     * @return All configured cache instances.
     */
    public Collection<IgniteInternalCache<?, ?>> caches() {
        return F.viewReadOnly(jCacheProxies.values(),
            (IgniteClosure<IgniteCacheProxy<?, ?>, IgniteInternalCache<?, ?>>)IgniteCacheProxy::internalProxy);
    }

    /**
     * @return All configured cache instances.
     */
    public Collection<IgniteCacheProxy<?, ?>> jcaches() {
        return F.viewReadOnly(jCacheProxies.values(),
            (IgniteClosure<IgniteCacheProxyImpl<?, ?>, IgniteCacheProxy<?, ?>>)IgniteCacheProxyImpl::gatewayWrapper);
    }

    /**
     * Gets utility cache.
     *
     * @return Utility cache.
     */
    public <K, V> IgniteInternalCache<K, V> utilityCache() {
        return internalCacheEx(CU.UTILITY_CACHE_NAME);
    }

    /**
     * @param name Cache name.
     * @return Cache.
     */
    private <K, V> IgniteInternalCache<K, V> internalCacheEx(String name) {
        if (ctx.discovery().localNode().isClient()) {
            IgniteCacheProxy<K, V> proxy = (IgniteCacheProxy<K, V>)jcacheProxy(name, true);

            if (proxy == null) {
                GridCacheAdapter<?, ?> cacheAdapter = caches.get(name);

                if (cacheAdapter != null) {
                    proxy = new IgniteCacheProxyImpl(cacheAdapter.context(), cacheAdapter, false);

                    IgniteCacheProxyImpl<?, ?> prev = addjCacheProxy(name, (IgniteCacheProxyImpl<?, ?>)proxy);

                    if (prev != null)
                        proxy = (IgniteCacheProxy<K, V>)prev;

                    completeProxyInitialize(proxy.getName());
                }
            }

            assert proxy != null : name;

            return proxy.internalProxy();
        }

        return internalCache(name);
    }

    /**
     * @param name Cache name.
     * @param <K> type of keys.
     * @param <V> type of values.
     * @return Cache instance for given name.
     * @throws IllegalArgumentException If cache not exists.
     */
    public <K, V> IgniteInternalCache<K, V> publicCache(String name) {
        assert name != null;

        if (log.isDebugEnabled())
            log.debug("Getting public cache for name: " + name);

        DynamicCacheDescriptor desc = cacheDescriptor(name);

        if (desc == null)
            throw new IllegalArgumentException("Cache is not started: " + name);

        if (!desc.cacheType().userCache())
            throw new IllegalStateException("Failed to get cache because it is a system cache: " + name);

        IgniteCacheProxy<K, V> jcache = (IgniteCacheProxy<K, V>)jcacheProxy(name, true);

        if (jcache == null)
            throw new IllegalArgumentException("Cache is not started: " + name);

        return jcache.internalProxy();
    }

    /**
     * @param cacheName Cache name.
     * @param <K> type of keys.
     * @param <V> type of values.
     * @return Cache instance for given name.
     * @throws IgniteCheckedException If failed.
     */
    public <K, V> IgniteCacheProxy<K, V> publicJCache(String cacheName) throws IgniteCheckedException {
        return publicJCache(cacheName, true, true);
    }

    /**
     * @param cacheName Cache name.
     * @param failIfNotStarted If {@code true} throws {@link IllegalArgumentException} if cache is not started,
     * otherwise returns {@code null} in this case.
     * @param checkThreadTx If {@code true} checks that current thread does not have active transactions.
     * @return Cache instance for given name.
     * @throws IgniteCheckedException If failed.
     */
    @SuppressWarnings({"ConstantConditions"})
    public @Nullable <K, V> IgniteCacheProxy<K, V> publicJCache(String cacheName,
        boolean failIfNotStarted,
        boolean checkThreadTx) throws IgniteCheckedException {
        assert cacheName != null;

        if (log.isDebugEnabled())
            log.debug("Getting public cache for name: " + cacheName);

        DynamicCacheDescriptor desc = cacheDescriptor(cacheName);

        if (desc != null && !desc.cacheType().userCache())
            throw new IllegalStateException("Failed to get cache because it is a system cache: " + cacheName);

        IgniteCacheProxyImpl<?, ?> proxy = jcacheProxy(cacheName, true);

        // Try to start cache, there is no guarantee that cache will be instantiated.
        if (proxy == null) {
            dynamicStartCache(null, cacheName, null, false, failIfNotStarted, checkThreadTx).get();

            proxy = jcacheProxy(cacheName, true);
        }

        return proxy != null ? (IgniteCacheProxy<K, V>)proxy.gatewayWrapper() : null;
    }

    /**
     * Get configuration for the given cache.
     *
     * @param name Cache name.
     * @return Cache configuration.
     */
    public CacheConfiguration cacheConfiguration(String name) {
        assert name != null;

        DynamicCacheDescriptor desc = cacheDescriptor(name);

        if (desc == null) {
            if (cachesInfo.isRestarting(name)) {
                IgniteCacheProxyImpl<?, ?> proxy = jCacheProxies.get(name);

                assert proxy != null : name;

                proxy.internalProxy(); //should throw exception

                // we have procceed, try again
                return cacheConfiguration(name);
            }

            throw new IllegalStateException("Cache doesn't exist: " + name);
        }
        else
            return desc.cacheConfiguration();
    }

    /**
     * Get registered cache descriptor.
     *
     * @param name Name.
     * @return Descriptor.
     */
    public DynamicCacheDescriptor cacheDescriptor(String name) {
        return cachesInfo.registeredCaches().get(name);
    }

    /**
     * @return Cache descriptors.
     */
    public Map<String, DynamicCacheDescriptor> cacheDescriptors() {
        return cachesInfo.registeredCaches();
    }

    /**
     * @return Collection of persistent cache descriptors.
     */
    public Collection<DynamicCacheDescriptor> persistentCaches() {
        return cachesInfo.registeredCaches().values()
            .stream()
            .filter(desc -> isPersistentCache(desc.cacheConfiguration(), ctx.config().getDataStorageConfiguration()))
            .collect(toList());
    }

    /**
     * @return Collection of persistent cache group descriptors.
     */
    public Collection<CacheGroupDescriptor> persistentGroups() {
        return cachesInfo.registeredCacheGroups().values()
            .stream()
            .filter(CacheGroupDescriptor::persistenceEnabled)
            .collect(toList());
    }

    /**
     * @return Cache group descriptors.
     */
    public Map<Integer, CacheGroupDescriptor> cacheGroupDescriptors() {
        return cachesInfo.registeredCacheGroups();
    }

    /**
     * Tries to find cache group descriptor either in registered cache groups
     * or in marked for deletion collection if cache group is considered to be stopped.
     *
     * @param grpId Group id.
     */
    public CacheGroupDescriptor cacheGroupDescriptor(int grpId) {
        CacheGroupDescriptor desc = cacheGroupDescriptors().get(grpId);

        // Try to find descriptor if it was marked for deletion.
        if (desc == null)
            return cachesInfo.markedForDeletionCacheGroupDesc(grpId);

        return desc;
    }

    /**
     * @param cacheId Cache ID.
     * @return Cache descriptor.
     */
    public @Nullable DynamicCacheDescriptor cacheDescriptor(int cacheId) {
        return cachesInfo.registeredCachesById().get(cacheId);
    }

    /**
     * @param cacheCfg Cache configuration template.
     * @throws IgniteCheckedException If failed.
     */
    public void addCacheConfiguration(CacheConfiguration cacheCfg) throws IgniteCheckedException {
        assert cacheCfg.getName() != null;

        String name = cacheCfg.getName();

        DynamicCacheDescriptor desc = cachesInfo.registeredTemplates().get(name);

        if (desc != null)
            return;

        DynamicCacheChangeRequest req = DynamicCacheChangeRequest.addTemplateRequest(ctx, cacheCfg,
            backwardCompatibleSplitter().split(cacheCfg));

        TemplateConfigurationFuture fut = new TemplateConfigurationFuture(req.cacheName(), req.deploymentId());

        TemplateConfigurationFuture old =
            (TemplateConfigurationFuture)pendingTemplateFuts.putIfAbsent(cacheCfg.getName(), fut);

        if (old != null)
            fut = old;

        Exception err = null;

        try {
            ctx.discovery().sendCustomEvent(new DynamicCacheChangeBatch(Collections.singleton(req)));

            if (ctx.isStopping()) {
                err = new IgniteCheckedException("Failed to execute dynamic cache change request, " +
                    "node is stopping.");
            }
            else if (ctx.clientDisconnected()) {
                err = new IgniteClientDisconnectedCheckedException(ctx.cluster().clientReconnectFuture(),
                    "Failed to execute dynamic cache change request, client node disconnected.");
            }
        }
        catch (IgniteCheckedException e) {
            err = e;
        }

        if (err != null)
            fut.onDone(err);

        fut.get();
    }

    /**
     * @param name Cache name.
     * @return Cache instance for given name.
     */
    @SuppressWarnings("unchecked")
    public <K, V> IgniteCacheProxy<K, V> jcache(String name) {
        assert name != null;

        IgniteCacheProxy<K, V> cache = (IgniteCacheProxy<K, V>)jcacheProxy(name, true);

        if (cache == null) {
            GridCacheAdapter<?, ?> cacheAdapter = caches.get(name);

            if (cacheAdapter != null) {
                cache = new IgniteCacheProxyImpl(cacheAdapter.context(), cacheAdapter, false);

                IgniteCacheProxyImpl<?, ?> prev = addjCacheProxy(name, (IgniteCacheProxyImpl<?, ?>)cache);

                if (prev != null)
                    cache = (IgniteCacheProxy<K, V>)prev;

                completeProxyInitialize(cache.getName());
            }
        }

        if (cache == null)
            throw new IllegalArgumentException("Cache is not configured: " + name);

        return cache;
    }

    /**
     * @param name Cache name.
     * @param awaitInit Await proxy initialization.
     * @return Cache proxy.
     */
    public @Nullable IgniteCacheProxyImpl<?, ?> jcacheProxy(String name, boolean awaitInit) {
        IgniteCacheProxyImpl<?, ?> cache = jCacheProxies.get(name);

        if (awaitInit)
            awaitInitializeProxy(cache);

        return cache;
    }

    /**
     * @param name Cache name.
     * @param proxy Cache proxy.
     * @return Previous cache proxy.
     */
    public @Nullable IgniteCacheProxyImpl<?, ?> addjCacheProxy(String name, IgniteCacheProxyImpl<?, ?> proxy) {
        return jCacheProxies.putIfAbsent(name, proxy);
    }

    /**
     * @return All configured public cache instances.
     */
    public Collection<IgniteCacheProxy<?, ?>> publicCaches() {
        Collection<IgniteCacheProxy<?, ?>> res = new ArrayList<>(jCacheProxies.size());

        for (IgniteCacheProxyImpl<?, ?> proxy : jCacheProxies.values()) {
            if (proxy.context().userCache())
                res.add(proxy.gatewayWrapper());
        }

        return res;
    }

    /**
     * @param name Cache name.
     * @param <K> type of keys.
     * @param <V> type of values.
     * @return Cache instance for given name.
     */
    public <K, V> GridCacheAdapter<K, V> internalCache(String name) {
        assert name != null;

        if (log.isDebugEnabled())
            log.debug("Getting internal cache adapter: " + name);

        return (GridCacheAdapter<K, V>)caches.get(name);
    }

    /**
     * Cancel all user operations.
     */
    private void cancelFutures() {
        sharedCtx.mvcc().onStop();

        Exception err = new IgniteCheckedException("Operation has been cancelled (node is stopping).");

        for (IgniteInternalFuture fut : pendingFuts.values())
            ((GridFutureAdapter)fut).onDone(err);

        for (IgniteInternalFuture fut : pendingTemplateFuts.values())
            ((GridFutureAdapter)fut).onDone(err);

        for (EnableStatisticsFuture fut : manageStatisticsFuts.values())
            fut.onDone(err);
    }

    /**
     * @return All internal cache instances.
     */
    public Collection<GridCacheAdapter<?, ?>> internalCaches() {
        return caches.values();
    }

    /**
     * @param name Cache name.
     * @return {@code True} if specified cache is system, {@code false} otherwise.
     */
    public boolean systemCache(String name) {
        assert name != null;

        DynamicCacheDescriptor desc = cacheDescriptor(name);

        return desc != null && !desc.cacheType().userCache();
    }

    /** {@inheritDoc} */
    @Override public void printMemoryStats() {
        X.println(">>> ");

        for (GridCacheAdapter c : caches.values()) {
            X.println(">>> Cache memory stats [igniteInstanceName=" + ctx.igniteInstanceName() +
                ", cache=" + c.name() + ']');

            c.context().printMemoryStats();
        }
    }

    /**
     * Callback invoked by deployment manager for whenever a class loader gets undeployed.
     *
     * @param ldr Class loader.
     */
    public void onUndeployed(ClassLoader ldr) {
        if (!ctx.isStopping()) {
            for (GridCacheAdapter<?, ?> cache : caches.values()) {
                // Do not notify system caches and caches for which deployment is disabled.
                if (cache.context().userCache() && cache.context().deploymentEnabled())
                    cache.onUndeploy(ldr);
            }
        }
    }

    /**
     * @return Shared context.
     */
    public <K, V> GridCacheSharedContext<K, V> context() {
        return (GridCacheSharedContext<K, V>)sharedCtx;
    }

    /** @return Local config manager. */
    public GridLocalConfigManager configManager() {
        return locCfgMgr;
    }

    /**
     * @return Transactions interface implementation.
     */
    public IgniteTransactionsEx transactions() {
        return transactions;
    }

    /**
     * Registers MBean for cache components.
     *
     * @param obj Cache component.
     * @param cacheName Cache name.
     * @param near Near flag.
     * @throws IgniteCheckedException If registration failed.
     */
    private void registerMbean(Object obj, @Nullable String cacheName, boolean near)
        throws IgniteCheckedException {
        if (U.IGNITE_MBEANS_DISABLED)
            return;

        assert obj != null;

        MBeanServer srvr = ctx.config().getMBeanServer();

        assert srvr != null;

        cacheName = U.maskName(cacheName);

        cacheName = near ? cacheName + "-near" : cacheName;

        final Object mbeanImpl = (obj instanceof IgniteMBeanAware) ? ((IgniteMBeanAware)obj).getMBean() : obj;

        for (Class<?> itf : mbeanImpl.getClass().getInterfaces()) {
            if (itf.getName().endsWith("MBean") || itf.getName().endsWith("MXBean")) {
                try {
                    U.registerMBean(srvr, ctx.igniteInstanceName(), cacheName, obj.getClass().getName(), mbeanImpl,
                        (Class<Object>)itf);
                }
                catch (Throwable e) {
                    throw new IgniteCheckedException("Failed to register MBean for component: " + obj, e);
                }

                break;
            }
        }
    }

    /**
     * Unregisters MBean for cache components.
     *
     * @param o Cache component.
     * @param cacheName Cache name.
     * @param near Near flag.
     */
    private void unregisterMbean(Object o, @Nullable String cacheName, boolean near) {
        if (U.IGNITE_MBEANS_DISABLED)
            return;

        assert o != null;

        MBeanServer srvr = ctx.config().getMBeanServer();

        assert srvr != null;

        cacheName = U.maskName(cacheName);

        cacheName = near ? cacheName + "-near" : cacheName;

        boolean needToUnregister = o instanceof IgniteMBeanAware;

        if (!needToUnregister) {
            for (Class<?> itf : o.getClass().getInterfaces()) {
                if (itf.getName().endsWith("MBean") || itf.getName().endsWith("MXBean")) {
                    needToUnregister = true;

                    break;
                }
            }
        }

        if (needToUnregister) {
            try {
                srvr.unregisterMBean(U.makeMBeanName(ctx.igniteInstanceName(), cacheName, o.getClass().getName()));
            }
            catch (Throwable e) {
                U.error(log, "Failed to unregister MBean for component: " + o, e);
            }
        }
    }

    /**
     * @param grp Cache group.
     * @param ccfg Cache configuration.
     * @param objs Extra components.
     * @return Components provided in cache configuration which can implement {@link LifecycleAware} interface.
     */
    private Iterable<Object> lifecycleAwares(CacheGroupContext grp, CacheConfiguration ccfg, Object... objs) {
        Collection<Object> ret = new ArrayList<>(7 + objs.length);

        if (grp.affinityFunction() != ccfg.getAffinity())
            ret.add(ccfg.getAffinity());

        ret.add(ccfg.getAffinityMapper());
        ret.add(ccfg.getEvictionFilter());
        ret.add(ccfg.getEvictionPolicyFactory());
        ret.add(ccfg.getEvictionPolicy());
        ret.add(ccfg.getInterceptor());

        NearCacheConfiguration nearCfg = ccfg.getNearConfiguration();

        if (nearCfg != null) {
            ret.add(nearCfg.getNearEvictionPolicyFactory());
            ret.add(nearCfg.getNearEvictionPolicy());
        }

        Collections.addAll(ret, objs);

        return ret;
    }

    /**
     * @param val Object to check.
     * @return Configuration copy.
     * @throws IgniteCheckedException If validation failed.
     */
    CacheConfiguration cloneCheckSerializable(final CacheConfiguration val) throws IgniteCheckedException {
        if (val == null)
            return null;

        return withBinaryContext(new IgniteOutClosureX<CacheConfiguration>() {
            @Override public CacheConfiguration applyx() throws IgniteCheckedException {
                if (val.getCacheStoreFactory() != null) {
                    try {
                        ClassLoader ldr = ctx.config().getClassLoader();

                        if (ldr == null)
                            ldr = val.getCacheStoreFactory().getClass().getClassLoader();

                        U.unmarshal(marsh, U.marshal(marsh, val.getCacheStoreFactory()),
                            U.resolveClassLoader(ldr, ctx.config()));
                    }
                    catch (IgniteCheckedException e) {
                        throw new IgniteCheckedException("Failed to validate cache configuration. " +
                            "Cache store factory is not serializable. Cache name: " + U.maskName(val.getName()), e);
                    }
                }

                try {
                    return U.unmarshal(marsh, U.marshal(marsh, val), U.resolveClassLoader(ctx.config()));
                }
                catch (IgniteCheckedException e) {
                    throw new IgniteCheckedException("Failed to validate cache configuration " +
                        "(make sure all objects in cache configuration are serializable): " + U.maskName(val.getName()), e);
                }
            }
        });
    }

    /**
     * @param c Closure.
     * @return Closure result.
     * @throws IgniteCheckedException If failed.
     */
    private <T> T withBinaryContext(IgniteOutClosureX<T> c) throws IgniteCheckedException {
        IgniteCacheObjectProcessor objProc = ctx.cacheObjects();
        BinaryContext oldCtx = null;

        if (objProc instanceof CacheObjectBinaryProcessorImpl) {
            GridBinaryMarshaller binMarsh = ((CacheObjectBinaryProcessorImpl)objProc).marshaller();

            oldCtx = binMarsh == null ? null : binMarsh.pushContext();
        }

        try {
            return c.applyx();
        }
        finally {
            if (objProc instanceof CacheObjectBinaryProcessorImpl)
                GridBinaryMarshaller.popContext(oldCtx);
        }
    }

    /**
     * Prepares DynamicCacheChangeRequest for cache creation.
     *
     * @param ccfg Cache configuration
     * @param cacheName Cache name
     * @param nearCfg Near cache configuration
     * @param cacheType Cache type
     * @param sql Whether the cache needs to be created as the result of SQL {@code CREATE TABLE} command.
     * @param failIfExists Fail if exists flag.
     * @param failIfNotStarted If {@code true} fails if cache is not started.
     * @param restartId Restart requester id (it'll allow to start this cache only him).
     * @param disabledAfterStart If true, cache proxies will be only activated after {@link #restartProxies()}.
     * @param qryEntities Query entities.
     * @param encKey Encryption key.
     * @param encKeyId Id of the encryption key.
     * @param masterKeyDigest Master key digest.
     * @return Request or {@code null} if cache already exists.
     * @throws IgniteCheckedException if some of pre-checks failed
     * @throws CacheExistsException if cache exists and failIfExists flag is {@code true}
     */
    private DynamicCacheChangeRequest prepareCacheChangeRequest(
        @Nullable CacheConfiguration ccfg,
        String cacheName,
        @Nullable NearCacheConfiguration nearCfg,
        CacheType cacheType,
        boolean sql,
        boolean failIfExists,
        boolean failIfNotStarted,
        IgniteUuid restartId,
        boolean disabledAfterStart,
        @Nullable Collection<QueryEntity> qryEntities,
        @Nullable byte[] encKey,
        @Nullable Integer encKeyId,
        @Nullable byte[] masterKeyDigest
    ) throws IgniteCheckedException {
        DynamicCacheDescriptor desc = cacheDescriptor(cacheName);

        DynamicCacheChangeRequest req = new DynamicCacheChangeRequest(UUID.randomUUID(), cacheName, ctx.localNodeId());

        req.sql(sql);

        req.failIfExists(failIfExists);

        req.disabledAfterStart(disabledAfterStart);

        req.masterKeyDigest(masterKeyDigest);

        req.encryptionKey(encKey);

        req.encryptionKeyId(encKeyId);

        req.restartId(restartId);

        if (ccfg != null) {
            cloneCheckSerializable(ccfg);

            if (desc != null) {
                if (failIfExists) {
                    throw new CacheExistsException("Failed to start cache " +
                        "(a cache with the same name is already started): " + cacheName);
                }
                else {
                    CacheConfiguration descCfg = desc.cacheConfiguration();

                    // Check if we were asked to start a near cache.
                    if (nearCfg != null) {
                        if (isLocalAffinity(descCfg)) {
                            // If we are on a data node and near cache was enabled, return success, else - fail.
                            if (descCfg.getNearConfiguration() != null)
                                return null;
                            else
                                throw new IgniteCheckedException("Failed to start near " +
                                    "cache (local node is an affinity node for cache): " + cacheName);
                        }
                        else
                            // If local node has near cache, return success.
                            req.clientStartOnly(true);
                    }
                    else if (!isLocalAffinity(descCfg))
                        req.clientStartOnly(true);

                    req.deploymentId(desc.deploymentId());

                    T2<CacheConfiguration, CacheConfigurationEnrichment> splitCfg = backwardCompatibleSplitter().split(desc);

                    req.startCacheConfiguration(splitCfg.get1());
                    req.cacheConfigurationEnrichment(splitCfg.get2());

                    req.schema(desc.schema());
                }
            }
            else {
                CacheConfiguration cfg = new CacheConfiguration(ccfg);

                CacheObjectContext cacheObjCtx = ctx.cacheObjects().contextForCache(cfg);

                // Cache configuration must be initialized before splitting.
                initialize(cfg, cacheObjCtx);

                req.deploymentId(IgniteUuid.randomUuid());

                T2<CacheConfiguration, CacheConfigurationEnrichment> splitCfg = backwardCompatibleSplitter().split(cfg);

                req.startCacheConfiguration(splitCfg.get1());
                req.cacheConfigurationEnrichment(splitCfg.get2());

                cfg = splitCfg.get1();

                if (restartId != null)
                    req.schema(new QuerySchema(qryEntities == null ? cfg.getQueryEntities() : qryEntities));
                else
                    req.schema(new QuerySchema(qryEntities != null ? QueryUtils.normalizeQueryEntities(ctx, qryEntities, cfg)
                            : cfg.getQueryEntities()));
            }
        }
        else {
            req.clientStartOnly(true);

            if (desc != null)
                ccfg = desc.cacheConfiguration();

            if (ccfg == null) {
                if (failIfNotStarted) {
                    throw new CacheExistsException("Failed to start client cache " +
                        "(a cache with the given name is not started): " + cacheName);
                }
                else
                    return null;
            }

            req.deploymentId(desc.deploymentId());

            T2<CacheConfiguration, CacheConfigurationEnrichment> splitCfg = backwardCompatibleSplitter().split(ccfg);

            req.startCacheConfiguration(splitCfg.get1());
            req.cacheConfigurationEnrichment(splitCfg.get2());

            req.schema(desc.schema());
        }

        if (nearCfg != null)
            req.nearCacheConfiguration(nearCfg);

        req.cacheType(cacheType);

        return req;
    }

    /**
     * Enable/disable statistics globally for the caches
     *
     * @param cacheNames Collection of cache names.
     * @param enabled Statistics enabled flag.
     */
    public void enableStatistics(Collection<String> cacheNames, boolean enabled) throws IgniteCheckedException {
        Collection<IgniteInternalCache> caches = manageStatisticsCaches(cacheNames);

        Collection<String> globalCaches = new HashSet<>(U.capacity(caches.size()));

        for (IgniteInternalCache cache : caches) {
            cache.context().statisticsEnabled(enabled);

            globalCaches.add(cache.name());
        }

        if (globalCaches.isEmpty())
            return;

        CacheStatisticsModeChangeMessage msg = new CacheStatisticsModeChangeMessage(UUID.randomUUID(), globalCaches, enabled);

        EnableStatisticsFuture fut = new EnableStatisticsFuture(msg.requestId());

        manageStatisticsFuts.put(msg.requestId(), fut);

        ctx.grid().context().discovery().sendCustomEvent(msg);

        fut.get();
    }

    /**
     * Clear statistics globally for the caches
     *
     * @param cacheNames Collection of cache names.
     */
    public void clearStatistics(Collection<String> cacheNames) throws IgniteCheckedException {
        Collection<IgniteInternalCache> caches = manageStatisticsCaches(cacheNames);

        Collection<String> globalCaches = new HashSet<>(U.capacity(caches.size()));

        for (IgniteInternalCache cache : caches)
            globalCaches.add(cache.name());

        if (globalCaches.isEmpty())
            return;

        CacheStatisticsClearMessage msg = new CacheStatisticsClearMessage(UUID.randomUUID(), globalCaches);

        EnableStatisticsFuture fut = new EnableStatisticsFuture(msg.requestId());

        manageStatisticsFuts.put(msg.requestId(), fut);

        ctx.grid().context().discovery().sendCustomEvent(msg);

        fut.get();
    }

    /**
     *
     */
    private Collection<IgniteInternalCache> manageStatisticsCaches(Collection<String> caches)
        throws IgniteCheckedException {
        assert caches != null;

        Collection<IgniteInternalCache> res = new ArrayList<>(caches.size());

        if (!cacheNames().containsAll(caches))
            throw new IgniteCheckedException("One or more cache descriptors not found [caches=" + caches + ']');

        for (String cacheName : caches) {
            IgniteInternalCache cache = cache(cacheName);

            if (cache == null)
                throw new IgniteCheckedException("Cache not found [cacheName=" + cacheName + ']');

            res.add(cache);
        }

        return res;
    }

    /**
     * @param obj Object to clone.
     * @return Object copy.
     * @throws IgniteCheckedException If failed.
     */
    public <T> T clone(final T obj) throws IgniteCheckedException {
        return withBinaryContext(new IgniteOutClosureX<T>() {
            @Override public T applyx() throws IgniteCheckedException {
                return U.unmarshal(marsh, U.marshal(marsh, obj), U.resolveClassLoader(ctx.config()));
            }
        });
    }

    /**
     * @param oldFormat Old format.
     */
    private CacheConfigurationSplitter splitter(boolean oldFormat) {
        // Requesting splitter with old format support is rare operation.
        // It's acceptable to allocate it every time by request.
        return oldFormat ? new CacheConfigurationSplitterOldFormat(enricher) : splitter;
    }

    /**
     * @return By default it returns splitter without old format configuration support.
     */
    public CacheConfigurationSplitter splitter() {
        return splitter(false);
    }

    /**
     * If not all nodes in cluster support splitted cache configurations it returns old format splitter.
     * In other case it returns default splitter.
     *
     * @return Cache configuration splitter with or without old format support depending on cluster state.
     */
    private CacheConfigurationSplitter backwardCompatibleSplitter() {
        IgniteDiscoverySpi spi = (IgniteDiscoverySpi)ctx.discovery().getInjectedDiscoverySpi();

        boolean oldFormat = !spi.allNodesSupport(IgniteFeatures.SPLITTED_CACHE_CONFIGURATIONS);

        return splitter(oldFormat);
    }

    /**
     * @return Cache configuration enricher.
     */
    public CacheConfigurationEnricher enricher() {
        return enricher;
    }

    /**
     * Pages list view supplier.
     *
     * @param filter Filter.
     */
    private Iterable<CachePagesListView> pagesListViewSupplier(Map<String, Object> filter) {
        Integer cacheGrpId = (Integer)filter.get(CachePagesListViewWalker.CACHE_GROUP_ID_FILTER);
        Integer partId = (Integer)filter.get(CachePagesListViewWalker.PARTITION_ID_FILTER);
        Integer bucketNum = (Integer)filter.get(CachePagesListViewWalker.BUCKET_NUMBER_FILTER);

        Iterable<IgniteCacheOffheapManager.CacheDataStore> dataStores = F.flat(F.iterator(
            filteredMap(cacheGrps, cacheGrpId).values(), grp -> grp.offheap().cacheDataStores(), true));

        return F.flat(F.iterator(dataStores, dataStore -> {
            RowStore rowStore = dataStore.rowStore();

            if (rowStore == null || !(dataStore instanceof GridCacheOffheapManager.GridCacheDataStore))
                return Collections.emptySet();

            PagesList pagesList = (PagesList)rowStore.freeList();

            if (bucketNum != null) {
                return bucketNum >= 0 && bucketNum < pagesList.bucketsCount() ?
                    Collections.singleton(new CachePagesListView(pagesList, bucketNum, dataStore.partId())) :
                    Collections.emptyList();
            }

            return IntStream.range(0, pagesList.bucketsCount())
                .mapToObj(bucket -> new CachePagesListView(pagesList, bucket, dataStore.partId()))
                .collect(toList());
        }, true, cacheDataStore -> partId == null || cacheDataStore.partId() == partId));
    }

    /**
     * Partition states view supplier.
     *
     * @param filter Filter.
     */
    private Iterable<PartitionStateView> partStatesViewSupplier(Map<String, Object> filter) {
        Integer cacheGrpId = (Integer)filter.get(PartitionStateViewWalker.CACHE_GROUP_ID_FILTER);
        UUID nodeId = (UUID)filter.get(PartitionStateViewWalker.NODE_ID_FILTER);
        Integer partId = (Integer)filter.get(PartitionStateViewWalker.PARTITION_ID_FILTER);

        return () -> F.concat(F.concat(F.iterator(filteredMap(cacheGrps, cacheGrpId).values(),
            grp -> F.iterator(filteredMap(grp.topology().partitionMap(false), nodeId).entrySet(),
                nodeToParts -> F.iterator(filteredMap(nodeToParts.getValue().map(),
                    partId == null || partId < 0 ? null : partId).entrySet(),
                    partToStates -> new PartitionStateView(
                        grp.groupId(),
                        nodeToParts.getKey(),
                        partToStates.getKey(),
                        partToStates.getValue(),
                        isPrimary(grp, nodeToParts.getKey(), partToStates.getKey())),
                    true),
                true),
            true)));
    }

    /**
     * Filter map by key.
     *
     * @param map Map.
     * @param key Filtering key.
     */
    private static <K, V> Map<K, V> filteredMap(Map<K, V> map, K key) {
        if (key == null)
            return map;

        V val = map.get(key);

        return val != null ? F.asMap(key, val) : Collections.emptyMap();
    }

    /**
     * @param grp Cache group.
     * @param nodeId Node id.
     * @param part Partition.
     */
    private static boolean isPrimary(CacheGroupContext grp, UUID nodeId, int part) {
        List<ClusterNode> nodes = grp.affinity().lastReadyAffinity().get(part);

        if (F.isEmpty(nodes))
            return false;

        ClusterNode primaryNode = nodes.get(0);

        return primaryNode != null && nodeId.equals(primaryNode.id());
    }

    /**
     * Recovery lifecycle for caches.
     */
    private class CacheRecoveryLifecycle implements MetastorageLifecycleListener, DatabaseLifecycleListener {
        /**
         * Set of QuerySchema's saved on recovery. It's needed if cache query schema has changed after node joined to
         * topology.
         */
        private final Map<Integer, QuerySchema> querySchemas = new ConcurrentHashMap<>();

        /** Flag for stopping warm-up. */
        private final AtomicBoolean stopWarmUp = new AtomicBoolean();

        /** Currently running warm-up strategy. */
        private volatile WarmUpStrategy curWarmUpStrat;

        /** {@inheritDoc} */
        @Override public void onBaselineChange() {
            for (GridCacheAdapter<?, ?> cache : caches.values())
                if (cache != null)
                    cache.context().ttl().unregister();

            onKernalStopCaches(true);

            stopCaches(true);

            sharedCtx.coordinators().stopTxLog();

            sharedCtx.database().cleanupRestoredCaches();
        }

        /** {@inheritDoc} */
        @Override public void onReadyForRead(ReadOnlyMetastorage metastorage) throws IgniteCheckedException {
            CacheJoinNodeDiscoveryData data = locCfgMgr.restoreCacheConfigurations();

            cachesInfo.onStart(data);
        }

        /** {@inheritDoc} */
        @Override public void beforeBinaryMemoryRestore(
            IgniteCacheDatabaseSharedManager mgr) throws IgniteCheckedException {
            for (DynamicCacheDescriptor cacheDescriptor : persistentCaches())
                preparePageStore(cacheDescriptor, true);
        }

        /** {@inheritDoc} */
        @Override public void afterBinaryMemoryRestore(
            IgniteCacheDatabaseSharedManager mgr,
            GridCacheDatabaseSharedManager.RestoreBinaryState restoreState) throws IgniteCheckedException {

            Object consistentId = ctx.pdsFolderResolver().resolveFolders().consistentId();
            DetachedClusterNode clusterNode = new DetachedClusterNode(consistentId, ctx.nodeAttributes());

            for (DynamicCacheDescriptor cacheDescriptor : persistentCaches()) {
                boolean affinityNode = CU.affinityNode(clusterNode, cacheDescriptor.cacheConfiguration().getNodeFilter());

                if (!affinityNode)
                    continue;

                startCacheInRecoveryMode(cacheDescriptor);

                querySchemas.put(cacheDescriptor.cacheId(), cacheDescriptor.schema().copy());
            }
        }

        /** {@inheritDoc} */
        @Override public void afterLogicalUpdatesApplied(
            IgniteCacheDatabaseSharedManager mgr,
            GridCacheDatabaseSharedManager.RestoreLogicalState restoreState
        ) throws IgniteCheckedException {
            Collection<CacheGroupContext> cacheGrps = cacheGroups();

            restorePartitionStates(cacheGrps, restoreState.partitionRecoveryStates());

            // Start warm-up only after restoring memory storage, but before starting GridDiscoveryManager.
            if (!cacheGrps.isEmpty())
                startWarmUp();
        }

        /**
         * Restoring the state of partitions for cache groups.
         *
         * @param forGroups Cache groups.
         * @param partStates Partition states.
         * @throws IgniteCheckedException If failed.
         */
        private void restorePartitionStates(
            Collection<CacheGroupContext> forGroups,
            Map<GroupPartitionId, Integer> partStates
        ) throws IgniteCheckedException {
            long startRestorePart = U.currentTimeMillis();

            if (log.isInfoEnabled())
                log.info("Restoring partition state for local groups.");

            AtomicReference<IgniteCheckedException> restoreStateError = new AtomicReference<>();

            ExecutorService sysPool = ctx.pools().getSystemExecutorService();

            final int totalPart = forGroups.stream().mapToInt(grpCtx -> grpCtx.affinity().partitions()).sum();

            CountDownLatch completionLatch = new CountDownLatch(totalPart);

            Map<Thread, RestorePartitionStateThreadContext> threadCtxs = new ConcurrentHashMap<>();

            final int topPartRefLimit = 5;

            for (CacheGroupContext grpCtx : forGroups) {
                for (int i = 0; i < grpCtx.affinity().partitions(); i++) {
                    final int partId = i;

                    sysPool.execute(() -> {
                        GroupPartitionId grpPartId = new GroupPartitionId(grpCtx.groupId(), partId);

                        try {
                            long time = grpCtx.offheap().restoreStateOfPartition(partId, partStates.get(grpPartId));

                            if (log.isInfoEnabled()) {
                                T3<Long, Long, GroupPartitionId> curPart = new T3<>(time, U.currentTimeMillis(), grpPartId);

                                RestorePartitionStateThreadContext threadCtx = threadCtxs.computeIfAbsent(
                                    Thread.currentThread(),
                                    t -> new RestorePartitionStateThreadContext()
                                );

                                Comparator<T3<Long, Long, GroupPartitionId>> cmp = processedPartitionComparator();

                                threadCtx.topPartRef.updateAndGet(prev -> {
                                    if (prev == null ||
                                        cmp.compare(prev.last(), curPart) < 0) {
                                        SortedSet<T3<Long, Long, GroupPartitionId>> top = new TreeSet<>(cmp);

                                        top.add(curPart);

                                        if (prev != null)
                                            top.addAll(prev);

                                        trimToSize(top, topPartRefLimit);

                                        return top;
                                    }
                                    else
                                        return prev;
                                });

                                threadCtx.incrementProcessedCnt();
                            }
                        }
                        catch (IgniteCheckedException | RuntimeException | Error e) {
                            U.error(log, "Failed to restore partition state for " +
                                "groupName=" + grpCtx.name() + " groupId=" + grpCtx.groupId(), e);

                            IgniteCheckedException ex = e instanceof IgniteCheckedException
                                ? ((IgniteCheckedException)e)
                                : new IgniteCheckedException(e);

                            if (!restoreStateError.compareAndSet(null, ex))
                                restoreStateError.get().addSuppressed(ex);
                        }
                        finally {
                            completionLatch.countDown();
                        }
                    });
                }
            }

            boolean printTop = false;

            try {
                // Await completion restore state tasks in all stripes.
                if (!log.isInfoEnabled())
                    completionLatch.await();
                else {
                    long timeout = TIMEOUT_OUTPUT_RESTORE_PARTITION_STATE_PROGRESS;

                    while (!completionLatch.await(timeout, TimeUnit.MILLISECONDS)) {
                        if (log.isInfoEnabled()) {
                            SortedSet<T3<Long, Long, GroupPartitionId>> top =
                                collectTopProcessedParts(threadCtxs.values(), topPartRefLimit);

                            long totalProcessed = threadCtxs.values().stream().mapToLong(c -> c.processedCnt).sum();

                            log.info("Restore partitions state progress [grpCnt=" +
                                (forGroups.size() - completionLatch.getCount()) + '/' + forGroups.size() +
                                ", partitionCnt=" + totalProcessed + '/' + totalPart + (top.isEmpty() ? "" :
                                ", topProcessedPartitions=" + toStringTopProcessingPartitions(top, forGroups)) + ']');
                        }

                        timeout = TIMEOUT_OUTPUT_RESTORE_PARTITION_STATE_PROGRESS / 5;

                        printTop = true;
                    }
                }
            }
            catch (InterruptedException e) {
                throw new IgniteInterruptedException(e);
            }

            for (CacheGroupContext grpCtx : forGroups)
                grpCtx.offheap().confirmPartitionStatesRestored();

            // Checking error after all task applied.
            if (restoreStateError.get() != null)
                throw restoreStateError.get();

            if (log.isInfoEnabled()) {
                SortedSet<T3<Long, Long, GroupPartitionId>> t =
                    printTop ? collectTopProcessedParts(threadCtxs.values(), topPartRefLimit) : null;

                long totalProcessed = threadCtxs.values().stream().mapToLong(c -> c.processedCnt).sum();

                log.info("Finished restoring partition state for local groups [" +
                    "groupsProcessed=" + forGroups.size() +
                    ", partitionsProcessed=" + totalProcessed +
                    ", time=" + U.humanReadableDuration(U.currentTimeMillis() - startRestorePart) +
                    (t == null ? "" : ", topProcessedPartitions=" + toStringTopProcessingPartitions(t, forGroups)) +
                    "]");
            }
        }

        /**
         * Collects top processed partitions from thread local contexts of restore partition state process.
         *
         * @param threadCtxs Thread local contexts.
         * @param topPartRefLimit Limit of top partitions collection size.
         */
        private SortedSet<T3<Long, Long, GroupPartitionId>> collectTopProcessedParts(
            Collection<RestorePartitionStateThreadContext> threadCtxs,
            int topPartRefLimit
        ) {
            SortedSet<T3<Long, Long, GroupPartitionId>> top = new TreeSet<>(processedPartitionComparator());

            for (RestorePartitionStateThreadContext threadCtx : threadCtxs) {
                SortedSet<T3<Long, Long, GroupPartitionId>> threadTop = threadCtx.topPartRef.get();

                if (threadTop != null)
                    top.addAll(threadTop);
            }

            trimToSize(top, topPartRefLimit);

            return top;
        }

        /**
         * Start warming up sequentially for each persist data region.
         *
         * @throws IgniteCheckedException If failed.
         */
        private void startWarmUp() throws IgniteCheckedException {
            boolean start = false;

            try {
                // Collecting custom and default data regions.
                DataStorageConfiguration dsCfg = ctx.config().getDataStorageConfiguration();

                List<DataRegionConfiguration> regCfgs =
                    new ArrayList<>(asList(dsCfg.getDefaultDataRegionConfiguration()));

                if (nonNull(dsCfg.getDataRegionConfigurations()))
                    regCfgs.addAll(asList(dsCfg.getDataRegionConfigurations()));

                // Warm-up start.
                Map<Class<? extends WarmUpConfiguration>, WarmUpStrategy> warmUpStrats = CU.warmUpStrategies(ctx);

                WarmUpConfiguration dfltWarmUpCfg = dsCfg.getDefaultWarmUpConfiguration();

                for (DataRegionConfiguration regCfg : regCfgs) {
                    if (stopWarmUp.get())
                        return;

                    if (!regCfg.isPersistenceEnabled())
                        continue;

                    WarmUpConfiguration warmUpCfg = nonNull(regCfg.getWarmUpConfiguration()) ?
                        regCfg.getWarmUpConfiguration() : dfltWarmUpCfg;

                    if (isNull(warmUpCfg))
                        continue;

                    WarmUpStrategy warmUpStrat = (curWarmUpStrat = warmUpStrats.get(warmUpCfg.getClass()));

                    DataRegion region = sharedCtx.database().dataRegion(regCfg.getName());

                    if (!stopWarmUp.get()) {
                        if (!start && (start = true) && log.isInfoEnabled())
                            log.info("Warm-up start.");

                        if (log.isInfoEnabled()) {
                            log.info("Start warm-up for data region [name=" + regCfg.getName()
                                + ", warmUpStrategy=" + warmUpStrat + ", warmUpConfig=" + warmUpCfg + ", isDefault="
                                + (warmUpCfg == dfltWarmUpCfg) + ']');
                        }

                        warmUpStrat.warmUp(warmUpCfg, region);

                        if (log.isInfoEnabled())
                            log.info("Finish of warm-up data region: " + region.config().getName());
                    }
                }
            }
            finally {
                if (stopWarmUp.get() && log.isInfoEnabled())
                    log.info("Warm-up stop.");
                else if (start && log.isInfoEnabled())
                    log.info("Warm-up finish.");

                stopWarmUp.set(true);
                curWarmUpStrat = null;
            }
        }
    }

    /**
     * Stop warming up and current running strategy.
     *
     * @return {@code true} if stopped by this call.
     * @throws IgniteCheckedException If there is an error when stopping warm-up.
     */
    public boolean stopWarmUp() throws IgniteCheckedException {
        if (recovery.stopWarmUp.compareAndSet(false, true)) {
            WarmUpStrategy strat = recovery.curWarmUpStrat;

            if (log.isInfoEnabled())
                log.info("Stopping warm-up strategy: " + strat);

            if (nonNull(strat))
                strat.stop();

            return true;
        }

        return false;
    }

    /**
     * Handle of fail during cache start.
     *
     * @param <T> Type of started data.
     */
    private interface StartCacheFailHandler<T, R> {
        /**
         * Handle of fail.
         *
         * @param data Start data.
         * @param startCacheOperation Operation for start cache.
         * @throws IgniteCheckedException if failed.
         */
        void handle(T data, IgniteThrowableFunction<T, R> startCacheOperation) throws IgniteCheckedException;
    }

    /**
     *
     */
    private class DynamicCacheStartFuture extends GridFutureAdapter<Boolean> {
        /** */
        private UUID id;

        /**
         * @param id Future ID.
         */
        private DynamicCacheStartFuture(UUID id) {
            this.id = id;
        }

        /** {@inheritDoc} */
        @Override public boolean onDone(@Nullable Boolean res, @Nullable Throwable err) {
            // Make sure to remove future before completion.
            pendingFuts.remove(id, this);

            context().exchange().exchangerUpdateHeartbeat();

            return super.onDone(res, err);
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(DynamicCacheStartFuture.class, this);
        }
    }

    /**
     *
     */
    private class TemplateConfigurationFuture extends GridFutureAdapter<Object> {
        /** Start ID. */
        @GridToStringInclude
        private IgniteUuid deploymentId;

        /** Cache name. */
        private String cacheName;

        /**
         * @param cacheName Cache name.
         * @param deploymentId Deployment ID.
         */
        private TemplateConfigurationFuture(String cacheName, IgniteUuid deploymentId) {
            this.deploymentId = deploymentId;
            this.cacheName = cacheName;
        }

        /**
         * @return Start ID.
         */
        public IgniteUuid deploymentId() {
            return deploymentId;
        }

        /** {@inheritDoc} */
        @Override public boolean onDone(@Nullable Object res, @Nullable Throwable err) {
            // Make sure to remove future before completion.
            pendingTemplateFuts.remove(cacheName, this);

            return super.onDone(res, err);
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(TemplateConfigurationFuture.class, this);
        }
    }

    /**
     * Enable statistics future.
     */
    private class EnableStatisticsFuture extends GridFutureAdapter<Void> {
        /** */
        private UUID id;

        /**
         * @param id Future ID.
         */
        private EnableStatisticsFuture(UUID id) {
            this.id = id;
        }

        /** {@inheritDoc} */
        @Override public boolean onDone(@Nullable Void res, @Nullable Throwable err) {
            // Make sure to remove future before completion.
            manageStatisticsFuts.remove(id, this);

            return super.onDone(res, err);
        }

        /** {@inheritDoc} */
        @Override public String toString() {
            return S.toString(EnableStatisticsFuture.class, this);
        }
    }

    /**
     * Creation of a string representation of the top (descending) partitions, the processing of which took the most time.
     *
     * @param top Top (ascending) processed partitions.
     * @param groups Cache group contexts.
     * @return String representation.
     */
    static String toStringTopProcessingPartitions(
        SortedSet<T3<Long, Long, GroupPartitionId>> top,
        Collection<CacheGroupContext> groups
    ) {
        if (top.isEmpty())
            return "[]";

        StringJoiner sj0 = new StringJoiner(", ", "[", "]");

        TreeMap<Long, List<GroupPartitionId>> top0 = top.stream().collect(groupingBy(
            T3::get1,
            TreeMap::new,
            mapping(T3::get3, toList())
        ));

        for (Map.Entry<Long, List<GroupPartitionId>> e0 : top0.descendingMap().entrySet()) {
            Map<Integer, List<GroupPartitionId>> byCacheGrpId =
                e0.getValue().stream().collect(groupingBy(GroupPartitionId::getGroupId));

            StringJoiner sj1 = new StringJoiner(", ", "[", "]");

            for (Map.Entry<Integer, List<GroupPartitionId>> e1 : byCacheGrpId.entrySet()) {
                @Nullable CacheGroupContext grp =
                    groups.stream().filter(g -> g.groupId() == e1.getKey()).findAny().orElse(null);

                String parts = e1.getValue().stream().map(GroupPartitionId::getPartitionId).sorted()
                    .map(p -> grp == null ? p.toString() : p + ":" + grp.topology().localPartition(p).fullSize())
                    .collect(Collectors.joining(", ", "[", "]"));

                sj1.add("[grp=" + (grp == null ? e1.getKey() : grp.cacheOrGroupName()) + ", part=" + parts + ']');
            }

            sj0.add("[time=" + U.humanReadableDuration(e0.getKey()) + ' ' + sj1.toString() + ']');
        }

        return sj0.toString();
    }

    /**
     * Trimming the set to the required size.
     * Removing items will be in ascending order.
     *
     * @param set Set.
     * @param size Size.
     */
    static <E> void trimToSize(SortedSet<E> set, int size) {
        while (set.size() > size)
            set.remove(set.first());
    }

    /**
     * Comparator of processed partitions.
     * T3 -> 1 - duration, 2 - timestamp, 3 - partition of group.
     * Sort order: duration -> timestamp (reversed order) -> partition of group.
     *
     * @return Comparator.
     */
    static Comparator<T3<Long, Long, GroupPartitionId>> processedPartitionComparator() {
        Comparator<T3<Long, Long, GroupPartitionId>> comp = Comparator.comparing(T3::get1);

        return comp.thenComparing(T3::get2, Comparator.reverseOrder()).thenComparing(T3::get3);
    }

    /**
     * Thread local context of restore partition state progress.
     */
    private static class RestorePartitionStateThreadContext {
        /** Field updater. */
        static final AtomicLongFieldUpdater<RestorePartitionStateThreadContext> PROCESSED_CNT_UPD =
            AtomicLongFieldUpdater.newUpdater(RestorePartitionStateThreadContext.class, "processedCnt");

        /** Top partitions by processing time. */
        final AtomicReference<SortedSet<T3<Long, Long, GroupPartitionId>>> topPartRef =
            new AtomicReference<>();

        /** Processed partitions count. It is always updated from the same thread. */
        volatile long processedCnt = 0;

        /**
         * Increment {@code processedCnt} field.
         */
        void incrementProcessedCnt() {
            PROCESSED_CNT_UPD.incrementAndGet(this);
        }
    }
}