/*
 * 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.persistence.snapshot;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.nio.file.OpenOption;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Predicate;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.Ignition;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
import org.apache.ignite.cache.query.ScanQuery;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemandMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionSupplyMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsAbstractMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.PartitionsExchangeAware;
import org.apache.ignite.internal.processors.cache.persistence.file.FileIO;
import org.apache.ignite.internal.processors.cache.persistence.file.FileIOFactory;
import org.apache.ignite.internal.processors.cache.persistence.file.RandomAccessFileIOFactory;
import org.apache.ignite.internal.processors.cache.persistence.partstate.GroupPartitionId;
import org.apache.ignite.internal.processors.metric.MetricRegistry;
import org.apache.ignite.internal.processors.metric.impl.ObjectGauge;
import org.apache.ignite.internal.util.distributed.DistributedProcess;
import org.apache.ignite.internal.util.distributed.FullMessage;
import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.T2;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.spi.metric.LongMetric;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.transactions.Transaction;
import org.junit.Before;
import org.junit.Test;

import static org.apache.ignite.cluster.ClusterState.ACTIVE;
import static org.apache.ignite.events.EventType.EVTS_CLUSTER_SNAPSHOT;
import static org.apache.ignite.events.EventType.EVT_CLUSTER_SNAPSHOT_FAILED;
import static org.apache.ignite.events.EventType.EVT_CLUSTER_SNAPSHOT_FINISHED;
import static org.apache.ignite.events.EventType.EVT_CLUSTER_SNAPSHOT_STARTED;
import static org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT;
import static org.apache.ignite.internal.processors.cache.distributed.rebalancing.GridCacheRebalancingSyncSelfTest.checkPartitionMapExchangeFinished;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNAPSHOT_METRICS;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNP_IN_PROGRESS_ERR_MSG;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.SNP_NODE_STOPPING_ERR_MSG;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.isSnapshotOperation;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.resolveSnapshotWorkDirectory;
import static org.apache.ignite.testframework.GridTestUtils.assertThrowsAnyCause;
import static org.apache.ignite.testframework.GridTestUtils.assertThrowsWithCause;

/**
 * Cluster-wide snapshot test.
 */
public class IgniteClusterSnapshotSelfTest extends AbstractSnapshotSelfTest {
    /** Time to wait while rebalance may happen. */
    private static final long REBALANCE_AWAIT_TIME = GridTestUtils.SF.applyLB(10_000, 3_000);

    /** Cache configuration for test. */
    private static final CacheConfiguration<Integer, Integer> atomicCcfg = new CacheConfiguration<Integer, Integer>("atomicCacheName")
        .setAtomicityMode(CacheAtomicityMode.ATOMIC)
        .setBackups(2)
        .setAffinity(new RendezvousAffinityFunction(false, CACHE_PARTS_COUNT));

    /** {@code true} if node should be started in separate jvm. */
    protected volatile boolean jvm;

    /** @throws Exception If fails. */
    @Before
    @Override public void beforeTestSnapshot() throws Exception {
        super.beforeTestSnapshot();

        jvm = false;
    }

    /**
     * Take snapshot from the whole cluster and check snapshot consistency when the
     * cluster tx load starts on a new topology version.
     * Note: Client nodes and server nodes not in baseline topology must not be affected.
     *
     * @throws Exception If fails.
     */
    @Test
    public void testConsistentClusterSnapshotLoadNewTopology() throws Exception {
        int grids = 3;
        String snpName = "backup23012020";
        AtomicInteger atKey = new AtomicInteger(CACHE_KEYS_RANGE);
        AtomicInteger txKey = new AtomicInteger(CACHE_KEYS_RANGE);

        IgniteEx ignite = startGrids(grids);
        startClientGrid();

        ignite.cluster().baselineAutoAdjustEnabled(false);
        ignite.cluster().state(ACTIVE);

        ignite.events().localListen(e -> locEvts.add(e.type()), EVTS_CLUSTER_SNAPSHOT);

        // Start node not in baseline.
        IgniteEx notBltIgnite = startGrid(grids);
        File locSnpDir = snp(notBltIgnite).snapshotLocalDir(SNAPSHOT_NAME);
        String notBltDirName = folderName(notBltIgnite);

        IgniteCache<Integer, Integer> atCache = ignite.createCache(atomicCcfg);

        for (int idx = 0; idx < CACHE_KEYS_RANGE; idx++) {
            atCache.put(atKey.incrementAndGet(), -1);
            ignite.cache(DEFAULT_CACHE_NAME).put(txKey.incrementAndGet(), -1);
        }

        forceCheckpoint();

        CountDownLatch loadLatch = new CountDownLatch(1);

        ignite.context().cache().context().exchange().registerExchangeAwareComponent(new PartitionsExchangeAware() {
            /** {@inheritDoc} */
            @Override public void onInitBeforeTopologyLock(GridDhtPartitionsExchangeFuture fut) {
                if (fut.firstEvent().type() != EVT_DISCOVERY_CUSTOM_EVT)
                    return;

                // First discovery custom event will be a snapshot operation.
                assertTrue(isSnapshotOperation(fut.firstEvent()));
                assertTrue("Snapshot must use pme-free exchange", fut.context().exchangeFreeSwitch());
            }

            /** {@inheritDoc} */
            @Override public void onInitAfterTopologyLock(GridDhtPartitionsExchangeFuture fut) {
                if (fut.firstEvent().type() != EVT_DISCOVERY_CUSTOM_EVT)
                    return;

                DiscoveryCustomMessage msg = ((DiscoveryCustomEvent)fut.firstEvent()).customMessage();

                assertNotNull(msg);

                if (msg instanceof SnapshotDiscoveryMessage)
                    loadLatch.countDown();
            }
        });

        // Start cache load.
        IgniteInternalFuture<Long> loadFut = GridTestUtils.runMultiThreadedAsync(() -> {
            try {
                U.await(loadLatch);

                while (!Thread.currentThread().isInterrupted()) {
                    ThreadLocalRandom rnd = ThreadLocalRandom.current();

                    int atIdx = rnd.nextInt(grids);

                    // Zero out the sign bit.
                    grid(atIdx).cache(atomicCcfg.getName()).put(txKey.incrementAndGet(), rnd.nextInt() & Integer.MAX_VALUE);

                    int txIdx = rnd.nextInt(grids);

                    grid(txIdx).cache(DEFAULT_CACHE_NAME).put(atKey.incrementAndGet(), rnd.nextInt() & Integer.MAX_VALUE);
                }
            }
            catch (IgniteInterruptedCheckedException e) {
                throw new RuntimeException(e);
            }
        }, 3, "cache-put-");

        try {
            IgniteFuture<Void> fut = ignite.snapshot().createSnapshot(snpName);

            U.await(loadLatch, 10, TimeUnit.SECONDS);

            fut.get();

            waitForEvents(Arrays.asList(EVT_CLUSTER_SNAPSHOT_STARTED, EVT_CLUSTER_SNAPSHOT_FINISHED));
        }
        finally {
            loadFut.cancel();
        }

        // Cluster can be deactivated but we must test snapshot restore when binary recovery also occurred.
        stopAllGrids();

        assertTrue("Snapshot directory must be empty for node not in baseline topology: " + notBltDirName,
            !searchDirectoryRecursively(locSnpDir.toPath(), notBltDirName).isPresent());

        IgniteEx snpIg0 = startGridsFromSnapshot(grids, snpName);

        assertEquals("The number of all (primary + backup) cache keys mismatch for cache: " + DEFAULT_CACHE_NAME,
            CACHE_KEYS_RANGE, snpIg0.cache(DEFAULT_CACHE_NAME).size());

        assertEquals("The number of all (primary + backup) cache keys mismatch for cache: " + atomicCcfg.getName(),
            CACHE_KEYS_RANGE, snpIg0.cache(atomicCcfg.getName()).size());

        snpIg0.cache(DEFAULT_CACHE_NAME).query(new ScanQuery<>(null))
            .forEach(e -> assertTrue("Snapshot must contains only negative values " +
                "[cache=" + DEFAULT_CACHE_NAME + ", entry=" + e + ']', (Integer)e.getValue() < 0));

        snpIg0.cache(atomicCcfg.getName()).query(new ScanQuery<>(null))
            .forEach(e -> assertTrue("Snapshot must contains only negative values " +
                "[cache=" + atomicCcfg.getName() + ", entry=" + e + ']', (Integer)e.getValue() < 0));
    }

    /** @throws Exception If fails. */
    @Test
    public void testSnapshotPrimaryBackupsTheSame() throws Exception {
        int grids = 3;
        AtomicInteger cacheKey = new AtomicInteger();

        IgniteEx ignite = startGridsWithCache(grids, dfltCacheCfg, CACHE_KEYS_RANGE);

        IgniteInternalFuture<Long> atLoadFut = GridTestUtils.runMultiThreadedAsync(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                ThreadLocalRandom rnd = ThreadLocalRandom.current();

                int gId = rnd.nextInt(grids);

                IgniteCache<Integer, Integer> txCache = grid(gId).getOrCreateCache(dfltCacheCfg.getName());

                try (Transaction tx = grid(gId).transactions().txStart()) {
                    txCache.put(cacheKey.incrementAndGet(), 0);

                    txCache.put(cacheKey.incrementAndGet(), 1);

                    tx.commit();
                }
            }
        }, 5, "tx-cache-put-");

        IgniteInternalFuture<Long> txLoadFut = GridTestUtils.runMultiThreadedAsync(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                ThreadLocalRandom rnd = ThreadLocalRandom.current();

                IgniteCache<Integer, Integer> atomicCache = grid(rnd.nextInt(grids))
                    .getOrCreateCache(atomicCcfg);

                atomicCache.put(cacheKey.incrementAndGet(), 0);
            }
        }, 5, "atomic-cache-put-");

        try {
            IgniteFuture<Void> fut = ignite.snapshot().createSnapshot(SNAPSHOT_NAME);

            fut.get();
        }
        finally {
            txLoadFut.cancel();
            atLoadFut.cancel();
        }

        stopAllGrids();

        IgniteEx snpIg0 = startGridsFromSnapshot(grids, cfg -> resolveSnapshotWorkDirectory(cfg).getAbsolutePath(), SNAPSHOT_NAME, false);

        // Block whole rebalancing.
        for (Ignite g : G.allGrids())
            TestRecordingCommunicationSpi.spi(g).blockMessages((node, msg) -> msg instanceof GridDhtPartitionDemandMessage);

        snpIg0.cluster().state(ACTIVE);

        assertFalse("Primary and backup in snapshot must have the same counters. Rebalance must not happen.",
            GridTestUtils.waitForCondition(() -> {
                boolean hasMsgs = false;

                for (Ignite g : G.allGrids())
                    hasMsgs |= TestRecordingCommunicationSpi.spi(g).hasBlockedMessages();

                return hasMsgs;
            }, REBALANCE_AWAIT_TIME));

        TestRecordingCommunicationSpi.stopBlockAll();

        assertPartitionsSame(idleVerify(snpIg0, dfltCacheCfg.getName(), atomicCcfg.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotConsistencyUnderLoad() throws Exception {
        int clientsCnt = 50;
        int balance = 10_000;
        int transferLimit = 1000;
        int total = clientsCnt * balance * 2;
        int grids = 3;
        int transferThreadCnt = 4;
        AtomicBoolean stop = new AtomicBoolean(false);
        CountDownLatch txStarted = new CountDownLatch(1);

        CacheConfiguration<Integer, Account> eastCcfg = txCacheConfig(new CacheConfiguration<>("east"));
        CacheConfiguration<Integer, Account> westCcfg = txCacheConfig(new CacheConfiguration<>("west"));

        startGridsWithCache(grids, clientsCnt, key -> new Account(key, balance), eastCcfg, westCcfg);

        Ignite client = startClientGrid(grids);

        assertEquals("The initial summary value in all caches is not correct.",
            total, sumAllCacheValues(client, clientsCnt, eastCcfg.getName(), westCcfg.getName()));

        forceCheckpoint();

        IgniteInternalFuture<?> txLoadFut = GridTestUtils.runMultiThreadedAsync(
            () -> {
                ThreadLocalRandom rnd = ThreadLocalRandom.current();

                int amount;

                try {
                    while (!stop.get()) {
                        IgniteEx ignite = grid(rnd.nextInt(grids));
                        IgniteCache<Integer, Account> east = ignite.cache("east");
                        IgniteCache<Integer, Account> west = ignite.cache("west");

                        amount = rnd.nextInt(transferLimit);

                        txStarted.countDown();

                        try (Transaction tx = ignite.transactions().txStart()) {
                            Integer id = rnd.nextInt(clientsCnt);

                            Account acc0 = east.get(id);
                            Account acc1 = west.get(id);

                            acc0.balance -= amount;
                            acc1.balance += amount;

                            east.put(id, acc0);
                            west.put(id, acc1);

                            tx.commit();
                        }
                    }
                }
                catch (Throwable e) {
                    U.error(log, e);

                    fail("Tx must not be failed.");
                }
            }, transferThreadCnt, "transfer-account-thread-");

        try {
            U.await(txStarted);

            grid(0).snapshot().createSnapshot(SNAPSHOT_NAME).get();
        }
        finally {
            stop.set(true);
        }

        txLoadFut.get();

        assertEquals("The summary value should not changed during tx transfers.",
            total, sumAllCacheValues(client, clientsCnt, eastCcfg.getName(), westCcfg.getName()));

        stopAllGrids();

        IgniteEx snpIg0 = startGridsFromSnapshot(grids, SNAPSHOT_NAME);

        assertEquals("The total amount of all cache values must not changed in snapshot.",
            total, sumAllCacheValues(snpIg0, clientsCnt, eastCcfg.getName(), westCcfg.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotWithCacheNodeFilter() throws Exception {
        int grids = 4;

        CacheConfiguration<Integer, Integer> ccfg = txCacheConfig(new CacheConfiguration<Integer, Integer>(DEFAULT_CACHE_NAME))
            .setNodeFilter(node -> node.consistentId().toString().endsWith("1"));

        IgniteEx ig0 = startGridsWithoutCache(grids);

        for (int i = 0; i < CACHE_KEYS_RANGE; i++)
            ig0.getOrCreateCache(ccfg).put(i, i);

        ig0.snapshot().createSnapshot(SNAPSHOT_NAME).get();

        stopAllGrids();

        IgniteEx snp = startGridsFromSnapshot(grids,
            cfg -> resolveSnapshotWorkDirectory(cfg.setCacheConfiguration()).getAbsolutePath(),
            SNAPSHOT_NAME,
            true);

        awaitPartitionMapExchange();
        checkCacheDiscoveryDataConsistent();

        CacheGroupDescriptor descr = snp.context().cache().cacheGroupDescriptors()
            .get(CU.cacheId(ccfg.getName()));

        assertNotNull(descr);
        assertNotNull(descr.config().getNodeFilter());
        assertEquals(ccfg.getNodeFilter().apply(grid(1).localNode()),
            descr.config().getNodeFilter().apply(grid(1).localNode()));
        assertSnapshotCacheKeys(snp.cache(ccfg.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testRejectCacheStopDuringClusterSnapshot() throws Exception {
        // Block the full message, so cluster-wide snapshot operation would not be fully completed.
        IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);

        BlockingCustomMessageDiscoverySpi spi = discoSpi(ignite);
        spi.block((msg) -> {
            if (msg instanceof FullMessage) {
                FullMessage<?> msg0 = (FullMessage<?>)msg;

                assertEquals("Snapshot distributed process must be used",
                    DistributedProcess.DistributedProcessType.START_SNAPSHOT.ordinal(), msg0.type());

                assertTrue("Snapshot has to be finished successfully on all nodes", msg0.error().isEmpty());

                return true;
            }

            return false;
        });

        IgniteFuture<Void> fut = ignite.snapshot().createSnapshot(SNAPSHOT_NAME);

        spi.waitBlocked(10_000L);

        // Creating of new caches should not be blocked.
        ignite.getOrCreateCache(dfltCacheCfg.setName("default2"))
            .put(1, 1);

        forceCheckpoint();

        assertThrowsAnyCause(log,
            () -> {
                ignite.destroyCache(DEFAULT_CACHE_NAME);

                return 0;
            },
            IgniteCheckedException.class,
            SNP_IN_PROGRESS_ERR_MSG);

        spi.unblock();

        fut.get();
    }

    /** @throws Exception If fails. */
    @Test
    public void testBltChangeDuringClusterSnapshot() throws Exception {
        IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);

        startGrid(3);

        long topVer = ignite.cluster().topologyVersion();

        BlockingCustomMessageDiscoverySpi spi = discoSpi(ignite);
        spi.block((msg) -> msg instanceof FullMessage);

        IgniteFuture<Void> fut = ignite.snapshot().createSnapshot(SNAPSHOT_NAME);

        spi.waitBlocked(10_000L);

        // Not baseline node joins successfully.
        String grid4Dir = folderName(startGrid(4));

        // Not blt node left the cluster and snapshot not affected.
        stopGrid(4);

        // Client node must connect successfully.
        startClientGrid(4);

        // Changing baseline complete successfully.
        ignite.cluster().setBaselineTopology(topVer);

        spi.unblock();

        fut.get();

        assertTrue("Snapshot directory must be empty for node 0 due to snapshot future fail: " + grid4Dir,
            !searchDirectoryRecursively(snp(ignite).snapshotLocalDir(SNAPSHOT_NAME).toPath(), grid4Dir).isPresent());
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotExOnInitiatorLeft() throws Exception {
        IgniteEx ignite = startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        BlockingCustomMessageDiscoverySpi spi = discoSpi(ignite);
        spi.block((msg) -> msg instanceof FullMessage);

        IgniteFuture<Void> fut = ignite.snapshot().createSnapshot(SNAPSHOT_NAME);

        spi.waitBlocked(10_000L);

        ignite.close();

        assertThrowsAnyCause(log,
            fut::get,
            NodeStoppingException.class,
            SNP_NODE_STOPPING_ERR_MSG);
    }

    /** @throws Exception If fails. */
    @Test
    public void testSnapshotExistsException() throws Exception {
        IgniteEx ignite = startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();

        assertThrowsAnyCause(log,
            () -> ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get(),
            IgniteException.class,
            "Snapshot with given name already exists on local node.");

        stopAllGrids();

        // Check that snapshot has not been accidentally deleted.
        IgniteEx snp = startGridsFromSnapshot(2, SNAPSHOT_NAME);

        assertSnapshotCacheKeys(snp.cache(dfltCacheCfg.getName()));

        waitForEvents(Arrays.asList(EVT_CLUSTER_SNAPSHOT_STARTED, EVT_CLUSTER_SNAPSHOT_FAILED));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotCleanedOnLeft() throws Exception {
        CountDownLatch block = new CountDownLatch(1);
        CountDownLatch partProcessed = new CountDownLatch(1);

        IgniteEx ignite = startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        File locSnpDir = snp(ignite).snapshotLocalDir(SNAPSHOT_NAME);
        String dirNameIgnite0 = folderName(ignite);

        String dirNameIgnite1 = folderName(grid(1));

        snp(grid(1)).localSnapshotSenderFactory(
            blockingLocalSnapshotSender(grid(1), partProcessed, block));

        TestRecordingCommunicationSpi commSpi1 = TestRecordingCommunicationSpi.spi(grid(1));
        commSpi1.blockMessages((node, msg) -> msg instanceof SingleNodeMessage);

        IgniteFuture<?> fut = ignite.snapshot().createSnapshot(SNAPSHOT_NAME);

        U.await(partProcessed);

        stopGrid(1);

        block.countDown();

        assertThrowsAnyCause(log,
            fut::get,
            IgniteCheckedException.class,
            "Execution of local snapshot tasks fails");

        waitForEvents(Arrays.asList(EVT_CLUSTER_SNAPSHOT_STARTED, EVT_CLUSTER_SNAPSHOT_FAILED));

        assertTrue("Snapshot directory must be empty for node 0 due to snapshot future fail: " + dirNameIgnite0,
            !searchDirectoryRecursively(locSnpDir.toPath(), dirNameIgnite0).isPresent());

        startGrid(1);

        awaitPartitionMapExchange();

        // Snapshot directory must be cleaned.
        assertTrue("Snapshot directory must be empty for node 1 due to snapshot future fail: " + dirNameIgnite1,
            !searchDirectoryRecursively(locSnpDir.toPath(), dirNameIgnite1).isPresent());

        List<String> allSnapshots = snp(ignite).localSnapshotNames();

        assertTrue("Snapshot directory must be empty due to snapshot fail: " + allSnapshots,
            allSnapshots.isEmpty());
    }

    /** @throws Exception If fails. */
    @Test
    public void testRecoveryClusterSnapshotJvmHalted() throws Exception {
        IgniteEx ignite = startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        String grid0Dir = folderName(ignite);
        String grid1Dir = folderName(grid(1));
        File locSnpDir = snp(ignite).snapshotLocalDir(SNAPSHOT_NAME);

        jvm = true;

        IgniteConfiguration cfg2 = optimize(getConfiguration(getTestIgniteInstanceName(2)));

        cfg2.getDataStorageConfiguration()
            .setFileIOFactory(new HaltJvmFileIOFactory(new RandomAccessFileIOFactory(),
                (Predicate<File> & Serializable) file -> {
                    // Trying to create FileIO over partition file.
                    return file.getAbsolutePath().contains(SNAPSHOT_NAME);
                }));

        startGrid(cfg2);

        String grid2Dir = U.maskForFileName(cfg2.getConsistentId().toString());

        jvm = false;

        ignite.cluster().setBaselineTopology(ignite.cluster().topologyVersion());

        awaitPartitionMapExchange();

        assertThrowsAnyCause(log,
            () -> ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get(),
            IgniteCheckedException.class,
            "Execution of local snapshot tasks fails");

        assertTrue("Snapshot directory must be empty: " + grid0Dir,
            !searchDirectoryRecursively(locSnpDir.toPath(), grid0Dir).isPresent());

        assertTrue("Snapshot directory must be empty: " + grid1Dir,
            !searchDirectoryRecursively(locSnpDir.toPath(), grid1Dir).isPresent());

        assertTrue("Snapshot directory must exist due to grid2 has been halted and cleanup not fully performed: " + grid2Dir,
            searchDirectoryRecursively(locSnpDir.toPath(), grid2Dir).isPresent());

        IgniteEx grid2 = startGrid(2);

        assertTrue("Snapshot directory must be empty after recovery: " + grid2Dir,
            !searchDirectoryRecursively(locSnpDir.toPath(), grid2Dir).isPresent());

        awaitPartitionMapExchange();

        assertTrue("Snapshot directory must be empty", grid2.context().cache().context().snapshotMgr().localSnapshotNames().isEmpty());

        ignite.snapshot().createSnapshot(SNAPSHOT_NAME)
            .get();

        stopAllGrids();

        IgniteEx snp = startGridsFromSnapshot(2, SNAPSHOT_NAME);

        assertSnapshotCacheKeys(snp.cache(dfltCacheCfg.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotWithRebalancing() throws Exception {
        IgniteEx ignite = startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        TestRecordingCommunicationSpi commSpi = TestRecordingCommunicationSpi.spi(ignite);
        commSpi.blockMessages((node, msg) -> msg instanceof GridDhtPartitionSupplyMessage);

        startGrid(2);

        ignite.cluster().setBaselineTopology(ignite.cluster().topologyVersion());

        commSpi.waitForBlocked();

        IgniteFuture<Void> fut = ignite.snapshot().createSnapshot(SNAPSHOT_NAME);

        commSpi.stopBlock(true);

        fut.get();

        waitForEvents(Arrays.asList(EVT_CLUSTER_SNAPSHOT_STARTED, EVT_CLUSTER_SNAPSHOT_FINISHED));

        stopAllGrids();

        IgniteEx snp = startGridsFromSnapshot(3, SNAPSHOT_NAME);

        awaitPartitionMapExchange();
        checkPartitionMapExchangeFinished();

        assertSnapshotCacheKeys(snp.cache(dfltCacheCfg.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotWithExplicitPath() throws Exception {
        File exSnpDir = U.resolveWorkDirectory(U.defaultWorkDirectory(), "ex_snapshots", true);

        try {
            IgniteEx ignite = null;

            for (int i = 0; i < 2; i++) {
                IgniteConfiguration cfg = optimize(getConfiguration(getTestIgniteInstanceName(i)));

                cfg.setSnapshotPath(exSnpDir.getAbsolutePath());

                ignite = startGrid(cfg);
            }

            ignite.cluster().baselineAutoAdjustEnabled(false);
            ignite.cluster().state(ACTIVE);

            for (int i = 0; i < CACHE_KEYS_RANGE; i++)
                ignite.cache(DEFAULT_CACHE_NAME).put(i, i);

            ignite.snapshot().createSnapshot(SNAPSHOT_NAME)
                .get();

            stopAllGrids();

            IgniteEx snp = startGridsFromSnapshot(2, cfg -> exSnpDir.getAbsolutePath(), SNAPSHOT_NAME, true);

            assertSnapshotCacheKeys(snp.cache(dfltCacheCfg.getName()));
        }
        finally {
            stopAllGrids();

            U.delete(exSnpDir);
        }
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotMetrics() throws Exception {
        String newSnapshotName = SNAPSHOT_NAME + "_new";
        CountDownLatch deltaApply = new CountDownLatch(1);
        CountDownLatch deltaBlock = new CountDownLatch(1);
        IgniteEx ignite = startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        MetricRegistry mreg0 = ignite.context().metric().registry(SNAPSHOT_METRICS);

        LongMetric startTime = mreg0.findMetric("LastSnapshotStartTime");
        LongMetric endTime = mreg0.findMetric("LastSnapshotEndTime");
        ObjectGauge<String> snpName = mreg0.findMetric("LastSnapshotName");
        ObjectGauge<String> errMsg = mreg0.findMetric("LastSnapshotErrorMessage");
        ObjectGauge<List<String>> snpList = mreg0.findMetric("LocalSnapshotNames");

        // Snapshot process will be blocked when delta partition files processing starts.
        snp(ignite).localSnapshotSenderFactory(
            blockingLocalSnapshotSender(ignite, deltaApply, deltaBlock));

        assertEquals("Snapshot start time must be undefined prior to snapshot operation started.",
            0, startTime.value());
        assertEquals("Snapshot end time must be undefined to snapshot operation started.",
            0, endTime.value());
        assertTrue("Snapshot name must not exist prior to snapshot operation started.", snpName.value().isEmpty());
        assertTrue("Snapshot error message must null prior to snapshot operation started.", errMsg.value().isEmpty());
        assertTrue("Snapshots on local node must not exist", snpList.value().isEmpty());

        long cutoffStartTime = U.currentTimeMillis();

        IgniteFuture<Void> fut0 = ignite.snapshot().createSnapshot(SNAPSHOT_NAME);

        U.await(deltaApply);

        assertTrue("Snapshot start time must be set prior to snapshot operation started " +
            "[startTime=" + startTime.value() + ", cutoffTime=" + cutoffStartTime + ']',
            startTime.value() >= cutoffStartTime);
        assertEquals("Snapshot end time must be zero prior to snapshot operation started.",
            0, endTime.value());
        assertEquals("Snapshot name must be set prior to snapshot operation started.",
            SNAPSHOT_NAME, snpName.value());
        assertTrue("Snapshot error message must null prior to snapshot operation started.",
            errMsg.value().isEmpty());

        IgniteFuture<Void> fut1 = grid(1).snapshot().createSnapshot(newSnapshotName);

        assertThrowsWithCause((Callable<Object>)fut1::get, IgniteException.class);

        MetricRegistry mreg1 = grid(1).context().metric().registry(SNAPSHOT_METRICS);

        LongMetric startTime1 = mreg1.findMetric("LastSnapshotStartTime");
        LongMetric endTime1 = mreg1.findMetric("LastSnapshotEndTime");
        ObjectGauge<String> snpName1 = mreg1.findMetric("LastSnapshotName");
        ObjectGauge<String> errMsg1 = mreg1.findMetric("LastSnapshotErrorMessage");

        assertTrue("Snapshot start time must be greater than zero for finished snapshot.",
            startTime1.value() > 0);
        assertEquals("Snapshot end time must zero for failed on start snapshots.",
            0, endTime1.value());
        assertEquals("Snapshot name must be set when snapshot operation already finished.",
            newSnapshotName, snpName1.value());
        assertNotNull("Concurrent snapshot operation must failed.",
            errMsg1.value());

        deltaBlock.countDown();

        fut0.get();

        assertTrue("Snapshot start time must be greater than zero for finished snapshot.",
            startTime.value() > 0);
        assertTrue("Snapshot end time must be greater than zero for finished snapshot.",
            endTime.value() > 0);
        assertEquals("Snapshot name must be set when snapshot operation already finished.",
            SNAPSHOT_NAME, snpName.value());
        assertTrue("Concurrent snapshot operation must finished successfully.",
            errMsg.value().isEmpty());
        assertEquals("Only the first snapshot must be created and stored on disk.",
            Collections.singletonList(SNAPSHOT_NAME), snpList.value());
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotIncorrectNameFails() throws Exception {
        IgniteEx ignite = startGridsWithCache(1, dfltCacheCfg, CACHE_KEYS_RANGE);

        assertThrowsAnyCause(log,
            () -> ignite.snapshot().createSnapshot("--№=+.:(snapshot)").get(),
            IllegalArgumentException.class,
            "Snapshot name must satisfy the following name pattern: a-zA-Z0-9_");
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotWithOfflineBlt() throws Exception {
        IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);

        stopGrid(2);

        ignite.snapshot().createSnapshot(SNAPSHOT_NAME)
            .get();

        stopAllGrids();

        IgniteEx snp = startGridsFromSnapshot(2, SNAPSHOT_NAME);

        awaitPartitionMapExchange();

        assertSnapshotCacheKeys(snp.cache(dfltCacheCfg.getName()));
        assertPartitionsSame(idleVerify(snp, dfltCacheCfg.getName()));
    }


    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotWithSharedCacheGroup() throws Exception {
        CacheConfiguration<Integer, Integer> ccfg1 = txCacheConfig(new CacheConfiguration<>("tx1"));
        CacheConfiguration<Integer, Integer> ccfg2 = txCacheConfig(new CacheConfiguration<>("tx2"));

        ccfg1.setGroupName("group");
        ccfg2.setGroupName("group");

        IgniteEx ignite = startGridsWithCache(3, CACHE_KEYS_RANGE, Integer::new, ccfg1, ccfg2);

        ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();

        waitForEvents(Arrays.asList(EVT_CLUSTER_SNAPSHOT_STARTED, EVT_CLUSTER_SNAPSHOT_FINISHED));

        stopAllGrids();

        IgniteEx snp = startGridsFromSnapshot(3, SNAPSHOT_NAME);

        awaitPartitionMapExchange();

        assertSnapshotCacheKeys(snp.cache(ccfg1.getName()));
        assertSnapshotCacheKeys(snp.cache(ccfg2.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotCoordinatorStopped() throws Exception {
        CountDownLatch block = new CountDownLatch(1);
        startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);
        startClientGrid(3);

        awaitPartitionMapExchange();

        for (IgniteEx grid : Arrays.asList(grid(1), grid(2))) {
            grid.context().cache().context().exchange()
                .registerExchangeAwareComponent(new PartitionsExchangeAware() {
                    /** {@inheritDoc} */
                    @Override public void onInitBeforeTopologyLock(GridDhtPartitionsExchangeFuture fut) {
                        try {
                            block.await(TIMEOUT, TimeUnit.MILLISECONDS);
                        }
                        catch (InterruptedException e) {
                            fail("Must not catch exception here: " + e.getMessage());
                        }
                    }
                });
        }

        for (Ignite grid : G.allGrids()) {
            TestRecordingCommunicationSpi.spi(grid)
                .blockMessages((node, msg) -> {
                    if (msg instanceof GridDhtPartitionsSingleMessage)
                        return ((GridDhtPartitionsAbstractMessage)msg).exchangeId() != null;

                    return false;
                });
        }

        IgniteFuture<Void> fut = grid(1).snapshot().createSnapshot(SNAPSHOT_NAME);

        stopGrid(0);

        block.countDown();

        // There are two exchanges happen: snapshot, node left (with pme-free).
        // Both of them are not require for sending messages.
        assertFalse("Pme-free switch doesn't expect messaging exchanging between nodes",
            GridTestUtils.waitForCondition(() -> {
                boolean hasMsgs = false;

                for (Ignite g : G.allGrids())
                    hasMsgs |= TestRecordingCommunicationSpi.spi(g).hasBlockedMessages();

                return hasMsgs;
            }, 5_000));

        assertThrowsWithCause((Callable<Object>)fut::get, IgniteException.class);

        List<GridDhtPartitionsExchangeFuture> exchFuts =
            grid(1).context().cache().context().exchange().exchangeFutures();

        assertFalse("Exchanges cannot be empty due to snapshot and node left happened",
            exchFuts.isEmpty());

        for (GridDhtPartitionsExchangeFuture exch : exchFuts) {
            assertTrue("Snapshot and node left events must keep `rebalanced` state" + exch,
                exch.rebalanced());
        }
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotOnMovingPartitionsCoordinatorLeft() throws Exception {
        startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        for (Ignite grid : G.allGrids()) {
            TestRecordingCommunicationSpi.spi(grid)
                .blockMessages((node, msg) -> msg instanceof GridDhtPartitionSupplyMessage);
        }

        Ignite ignite = startGrid(2);

        ignite.cluster().setBaselineTopology(ignite.cluster().topologyVersion());

        TestRecordingCommunicationSpi.spi(grid(0))
            .waitForBlocked();

        CountDownLatch latch = new CountDownLatch(G.allGrids().size());
        IgniteInternalFuture<?> stopFut = GridTestUtils.runAsync(() -> {
            try {
                U.await(latch);

                stopGrid(0);
            }
            catch (IgniteInterruptedCheckedException e) {
                fail("Must not fail here: " + e.getMessage());
            }
        });

        Queue<T2<GridDhtPartitionExchangeId, Boolean>> exchFuts = new ConcurrentLinkedQueue<>();

        for (Ignite ig : G.allGrids()) {
            ((IgniteEx)ig).context().cache().context().exchange()
                .registerExchangeAwareComponent(new PartitionsExchangeAware() {
                    /** {@inheritDoc} */
                    @Override public void onInitBeforeTopologyLock(GridDhtPartitionsExchangeFuture fut) {
                        if (!(fut.firstEvent() instanceof DiscoveryCustomEvent))
                            return;

                        try {
                            exchFuts.add(new T2<>(fut.exchangeId(), fut.rebalanced()));
                            latch.countDown();

                            stopFut.get();
                        }
                        catch (IgniteCheckedException e) {
                            U.log(log, "Interrupted on coordinator: " + e.getMessage());
                        }
                    }
                });
        }

        IgniteFuture<Void> fut = ignite.snapshot().createSnapshot(SNAPSHOT_NAME);

        stopFut.get();

        assertThrowsAnyCause(log,
            fut::get,
            IgniteException.class,
            "Snapshot creation has been finished with an error");

        assertEquals("Snapshot futures expected: " + exchFuts, 3, exchFuts.size());

        for (T2<GridDhtPartitionExchangeId, Boolean> exch : exchFuts)
            assertFalse("Snapshot `rebalanced` must be false with moving partitions: " + exch.get1(), exch.get2());
    }

    /** @throws Exception If fails. */
    @Test
    public void testSnapshotPartitionExchangeAwareOrder() throws Exception {
        IgniteEx ignite = startGridsWithCache(3, dfltCacheCfg, CACHE_KEYS_RANGE);

        Map<UUID, PartitionsExchangeAware> comps = new HashMap<>();

        for (Ignite ig : G.allGrids()) {
            PartitionsExchangeAware comp;

            ((IgniteEx)ig).context().cache().context().exchange()
                .registerExchangeAwareComponent(comp = new PartitionsExchangeAware() {
                    private final AtomicInteger order = new AtomicInteger();

                    @Override public void onInitBeforeTopologyLock(GridDhtPartitionsExchangeFuture fut) {
                        assertEquals("Exchange order violated: " + fut.firstEvent(), 0, order.getAndIncrement());
                    }

                    @Override public void onInitAfterTopologyLock(GridDhtPartitionsExchangeFuture fut) {
                        assertEquals("Exchange order violated: " + fut.firstEvent(), 1, order.getAndIncrement());
                    }

                    @Override public void onDoneBeforeTopologyUnlock(GridDhtPartitionsExchangeFuture fut) {
                        assertEquals("Exchange order violated: " + fut.firstEvent(), 2, order.getAndIncrement());
                    }

                    @Override public void onDoneAfterTopologyUnlock(GridDhtPartitionsExchangeFuture fut) {
                        assertEquals("Exchange order violated: " + fut.firstEvent(), 3, order.getAndSet(0));
                    }
                });

            comps.put(((IgniteEx)ig).localNode().id(), comp);
        }

        ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();

        for (Ignite ig : G.allGrids()) {
            ((IgniteEx)ig).context().cache().context().exchange()
                .unregisterExchangeAwareComponent(comps.get(((IgniteEx)ig).localNode().id()));
        }

        awaitPartitionMapExchange();

        assertEquals("Some of ignite instances failed during snapshot", 3, G.allGrids().size());

        stopAllGrids();

        IgniteEx snp = startGridsFromSnapshot(3, SNAPSHOT_NAME);

        assertSnapshotCacheKeys(snp.cache(dfltCacheCfg.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotFromClient() throws Exception {
        startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);
        IgniteEx clnt = startClientGrid(2);

        clnt.snapshot().createSnapshot(SNAPSHOT_NAME).get();

        waitForEvents(Arrays.asList(EVT_CLUSTER_SNAPSHOT_STARTED, EVT_CLUSTER_SNAPSHOT_FINISHED));

        stopAllGrids();

        IgniteEx snp = startGridsFromSnapshot(2, SNAPSHOT_NAME);

        awaitPartitionMapExchange();
        assertSnapshotCacheKeys(snp.cache(dfltCacheCfg.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testConcurrentClusterSnapshotFromClient() throws Exception {
        IgniteEx grid = startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        IgniteEx clnt = startClientGrid(2);

        IgniteSnapshotManager mgr = snp(grid);
        Function<String, SnapshotSender> old = mgr.localSnapshotSenderFactory();

        BlockingExecutor block = new BlockingExecutor(mgr.snapshotExecutorService());

        mgr.localSnapshotSenderFactory((snpName) ->
            new DelegateSnapshotSender(log, block, old.apply(snpName)));

        IgniteFuture<Void> fut = grid.snapshot().createSnapshot(SNAPSHOT_NAME);

        assertThrowsAnyCause(log,
            () -> clnt.snapshot().createSnapshot(SNAPSHOT_NAME).get(),
            IgniteException.class,
            "Snapshot has not been created");

        block.unblock();
        fut.get();
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotFromClientDisconnected() throws Exception {
        startGridsWithCache(1, dfltCacheCfg, CACHE_KEYS_RANGE);
        IgniteEx clnt = startClientGrid(1);

        stopGrid(0);

        assertThrowsAnyCause(log,
            () -> clnt.snapshot().createSnapshot(SNAPSHOT_NAME).get(),
            IgniteException.class,
            "Client disconnected. Snapshot result is unknown");
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotInProgressCancelled() throws Exception {
        IgniteEx srv = startGridsWithCache(1, dfltCacheCfg, CACHE_KEYS_RANGE);
        IgniteEx startCli = startClientGrid(1);
        IgniteEx killCli = startClientGrid(2);

        doSnapshotCancellationTest(startCli, Collections.singletonList(srv), srv.cache(dfltCacheCfg.getName()),
            snpName -> killCli.snapshot().cancelSnapshot(snpName).get());

        waitForEvents(Arrays.asList(EVT_CLUSTER_SNAPSHOT_STARTED, EVT_CLUSTER_SNAPSHOT_FAILED));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotFinishedTryCancel() throws Exception {
        IgniteEx ignite = startGridsWithCache(2, dfltCacheCfg, CACHE_KEYS_RANGE);

        ignite.snapshot().createSnapshot(SNAPSHOT_NAME).get();
        ignite.snapshot().cancelSnapshot(SNAPSHOT_NAME).get();

        stopAllGrids();

        IgniteEx snpIg = startGridsFromSnapshot(2, SNAPSHOT_NAME);

        assertSnapshotCacheKeys(snpIg.cache(dfltCacheCfg.getName()));
    }

    /** @throws Exception If fails. */
    @Test
    public void testClusterSnapshotInMemoryFail() throws Exception {
        persistence = false;

        IgniteEx srv = startGrid(0);

        srv.cluster().state(ACTIVE);

        IgniteEx clnt = startClientGrid(1);

        IgniteFuture<?> fut = clnt.snapshot().createSnapshot(SNAPSHOT_NAME);

        assertThrowsAnyCause(log,
            fut::get,
            IgniteException.class,
            "Snapshots on an in-memory clusters are not allowed.");
    }

    /**
     * @param ignite Ignite instance.
     * @param started Latch will be released when delta partition processing starts.
     * @param blocked Latch to await delta partition processing.
     * @return Factory which produces local snapshot senders.
     */
    private Function<String, SnapshotSender> blockingLocalSnapshotSender(IgniteEx ignite,
        CountDownLatch started,
        CountDownLatch blocked
    ) {
        Function<String, SnapshotSender> old = snp(ignite).localSnapshotSenderFactory();

        return (snpName) -> new DelegateSnapshotSender(log, snp(ignite).snapshotExecutorService(), old.apply(snpName)) {
            @Override public void sendDelta0(File delta, String cacheDirName, GroupPartitionId pair) {
                if (log.isInfoEnabled())
                    log.info("Processing delta file has been blocked: " + delta.getName());

                started.countDown();

                try {
                    U.await(blocked);

                    if (log.isInfoEnabled())
                        log.info("Latch released. Processing delta file continued: " + delta.getName());

                    super.sendDelta0(delta, cacheDirName, pair);
                }
                catch (IgniteInterruptedCheckedException e) {
                    throw new IgniteException("Interrupted by node stop", e);
                }
            }
        };
    }

    /** {@inheritDoc} */
    @Override protected boolean isMultiJvm() {
        return jvm;
    }

    /**
     * @param ignite Ignite instance.
     * @param caches Cache names to read values.
     * @return Summary value.
     */
    private static int sumAllCacheValues(Ignite ignite, int keys, String... caches) {
        AtomicInteger total = new AtomicInteger();

        for (String name : caches) {
            IgniteCache<Integer, Account> cache = ignite.cache(name);

            for (int key = 0; key < keys; key++)
                total.addAndGet(cache.get(key).balance);
        }

        return total.get();
    }

    /**
     * I/O Factory which will halt JVM on conditions occurred.
     */
    private static class HaltJvmFileIOFactory implements FileIOFactory {
        /** Serial version UID. */
        private static final long serialVersionUID = 0L;

        /** Delegate factory. */
        private final FileIOFactory delegate;

        /** Condition to halt. */
        private final Predicate<File> pred;

        /**
         * @param delegate Delegate factory.
         */
        public HaltJvmFileIOFactory(FileIOFactory delegate, Predicate<File> pred) {
            this.delegate = delegate;
            this.pred = pred;
        }

        /** {@inheritDoc} */
        @Override public FileIO create(File file, OpenOption... modes) throws IOException {
            FileIO delegate = this.delegate.create(file, modes);

            if (pred.test(file))
                Runtime.getRuntime().halt(Ignition.KILL_EXIT_CODE);

            return delegate;
        }
    }
}