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

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiConsumer;
import java.util.function.Predicate;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteDataStreamer;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteInterruptedException;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
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.IgniteConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.events.DiscoveryCustomEvent;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch;
import org.apache.ignite.internal.processors.cache.ExchangeActions;
import org.apache.ignite.internal.processors.cache.ExchangeActions.CacheActionData;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionExchangeId;
import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage;
import org.apache.ignite.internal.util.lang.RunnableX;
import org.apache.ignite.internal.util.typedef.X;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteBiPredicate;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessage;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeLeftMessage;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.jetbrains.annotations.NotNull;
import org.junit.Test;

import static org.apache.ignite.cache.CacheAtomicityMode.ATOMIC;
import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
import static org.apache.ignite.cache.CacheMode.PARTITIONED;
import static org.apache.ignite.cache.CacheMode.REPLICATED;

/**
 *
 */
public abstract class CacheBlockOnReadAbstractTest extends GridCommonAbstractTest {
    /** Default cache entries count. */
    private static final int DFLT_CACHE_ENTRIES_CNT = 2 * 1024;

    /** Ip finder. */
    private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);

    /** List of baseline nodes started at the beginning of the test. */
    protected final List<IgniteEx> baseline = new CopyOnWriteArrayList<>();

    /** List of server nodes started at the beginning of the test. */
    protected final List<IgniteEx> srvs = new CopyOnWriteArrayList<>();

    /** List of client nodes started at the beginning of the test. */
    protected final List<IgniteEx> clients = new CopyOnWriteArrayList<>();

    /** Start node in client mode. */
    private volatile boolean startNodesInClientMode;

    /** Latch that is used to wait until all required messages are blocked. */
    private volatile CountDownLatch cntFinishedReadOperations;

    /** Custom ip finder. Replaces {@link #IP_FINDER} if present at the moment of node starting. */
    private volatile TcpDiscoveryIpFinder customIpFinder;

    /** Discovery message processor. Used in every started node. */
    private volatile BiConsumer<TcpDiscoveryAbstractMessage, String> discoveryMsgProcessor;

    /**
     * Number of baseline servers to start before test.
     *
     * @see Params#baseline()
     */
    protected int baselineServersCount() {
        return currentTestParams().baseline();
    }

    /**
     * Number of non-baseline servers to start before test.
     *
     * @see Params#servers()
     */
    protected int serversCount() {
        return currentTestParams().servers();
    }

    /**
     * Number of clients to start before test.
     *
     * @see Params#clients()
     */
    protected int clientsCount() {
        return currentTestParams().clients();
    }

    /**
     * Number of backups to configure in caches by default.
     */
    protected int backupsCount() {
        return Math.min(3, baselineServersCount() - 1);
    }

    /**
     * Number of milliseconds to warmup reading process. Used to lower fluctuations in run time. Might be 0.
     *
     * @see Params#warmup()
     */
    protected long warmup() {
        return currentTestParams().warmup();
    }

    /**
     * Number of milliseconds to wait on the potentially blocking operation.
     *
     * @see Params#timeout()
     */
    protected long timeout() {
        return currentTestParams().timeout();
    }

    /**
     * Cache atomicity mode.
     *
     * @see Params#atomicityMode()
     */
    protected CacheAtomicityMode atomicityMode() {
        return currentTestParams().atomicityMode();
    }

    /**
     * Cache mode.
     *
     * @see Params#cacheMode()
     */
    protected CacheMode cacheMode() {
        return currentTestParams().cacheMode();
    }

    /**
     * Whether allowing {@link ClusterTopologyCheckedException} as the valid reading result or not.
     *
     * @see Params#allowException()
     */
    protected boolean allowException() {
        return currentTestParams().allowException();
    }

    /**
     * @param startNodesInClientMode Start nodes on client mode.
     */
    public void startNodesInClientMode(boolean startNodesInClientMode) {
        this.startNodesInClientMode = startNodesInClientMode;
    }

    /** List of baseline nodes started at the beginning of the test. */
    public List<? extends IgniteEx> baseline() {
        return baseline;
    }

    /** List of server nodes started at the beginning of the test. */
    public List<? extends IgniteEx> servers() {
        return srvs;
    }

    /** List of client nodes started at the beginning of the test. */
    public List<? extends IgniteEx> clients() {
        return clients;
    }

    /**
     * Annotation to configure test methods in {@link CacheBlockOnReadAbstractTest}. Its values are used throughout
     * test implementation.
     */
    @Target(ElementType.METHOD)
    @Retention(RetentionPolicy.RUNTIME)
    public @interface Params {
        /**
         * Number of baseline servers to start before test.
         */
        int baseline() default 3;

        /**
         * Number of non-baseline servers to start before test.
         */
        int servers() default 1;

        /**
         * Number of clients to start before test.
         */
        int clients() default 1;

        /**
         * Number of milliseconds to warmup reading process. Used to lower fluctuations in run time. Might be 0.
         */
        long warmup() default 2000L;

        /**
         * Number of milliseconds to wait on the potentially blocking operation.
         */
        long timeout() default 3000L;

        /**
         * Cache atomicity mode.
         */
        CacheAtomicityMode atomicityMode();

        /**
         * Cache mode.
         */
        CacheMode cacheMode();

        /**
         * Whether allowing {@link ClusterTopologyCheckedException} as the valid reading result or not.
         */
        boolean allowException() default false;
    }


    /** {@inheritDoc} */
    @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
        IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);

        cfg.setConsistentId(igniteInstanceName);

        cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());

        cfg.setDiscoverySpi(new TestTcpDiscoverySpi() {
            /** {@inheritDoc} */
            @Override protected void startMessageProcess(TcpDiscoveryAbstractMessage msg) {
                if (discoveryMsgProcessor != null)
                    discoveryMsgProcessor.accept(msg, igniteInstanceName);
            }
        }.setIpFinder(customIpFinder == null ? IP_FINDER : customIpFinder));

        cfg.setDataStorageConfiguration(
            new DataStorageConfiguration()
                .setDefaultDataRegionConfiguration(
                    new DataRegionConfiguration()
                        .setPersistenceEnabled(true)
                )
        );

        cfg.setClientMode(startNodesInClientMode);

        return cfg;
    }

    /** {@inheritDoc} */
    @Override public void beforeTest() throws Exception {
        stopAllGrids();

        cleanPersistenceDir();

        // Checking prerequisites.
        assertTrue("Positive timeout is required for the test.", timeout() > 0);

        assertTrue("No baseline servers were requested.", baselineServersCount() > 0);

        int idx = 0;

        // Start baseline nodes.
        for (int i = 0; i < baselineServersCount(); i++)
            baseline.add(startGrid(idx++));

        // Activate cluster.
        baseline.get(0).cluster().state(ClusterState.ACTIVE);

        // Start server nodes in activated cluster.
        for (int i = 0; i < serversCount(); i++)
            srvs.add(startGrid(idx++));

        // Start client nodes.
        startNodesInClientMode(true);

        customIpFinder = new TcpDiscoveryVmIpFinder(false)
            .setAddresses(
                Collections.singletonList("127.0.0.1:47500")
            );

        for (int i = 0; i < clientsCount(); i++)
            clients.add(startGrid(idx++));

        customIpFinder = null;
    }

    /** {@inheritDoc} */
    @Override public void afterTest() throws Exception {
        baseline.clear();

        srvs.clear();

        clients.clear();

        grid(0).cluster().state(ClusterState.INACTIVE);

        stopAllGrids();

        cleanPersistenceDir();
    }


    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testCreateCacheAtomicPartitioned() throws Exception {
        testCreateCacheTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testCreateCacheAtomicReplicated() throws Exception {
        testCreateCacheTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testCreateCacheTransactionalPartitioned() throws Exception {
        testCreateCacheTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
    @Test
    public void testCreateCacheTransactionalReplicated() throws Exception {
        doTest(
            asMessagePredicate(CacheBlockOnReadAbstractTest::createCachePredicate),
            () -> baseline.get(0).createCache(UUID.randomUUID().toString())
        );
    }

    /**
     * @throws Exception If failed.
     */
    @Params(timeout = 5000L, atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testDestroyCacheAtomicPartitioned() throws Exception {
        testDestroyCacheTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(timeout = 5000L, atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testDestroyCacheAtomicReplicated() throws Exception {
        testDestroyCacheTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(timeout = 5000L, atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testDestroyCacheTransactionalPartitioned() throws Exception {
        testDestroyCacheTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(timeout = 5000L, atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
    @Test
    public void testDestroyCacheTransactionalReplicated() throws Exception {
        List<String> cacheNames = new ArrayList<>(Arrays.asList(
            UUID.randomUUID().toString(),
            UUID.randomUUID().toString(),
            UUID.randomUUID().toString())
        );

        for (String cacheName : cacheNames)
            baseline.get(0).createCache(cacheName);

        doTest(
            asMessagePredicate(CacheBlockOnReadAbstractTest::destroyCachePredicate),
            () -> baseline.get(0).destroyCache(cacheNames.remove(0))
        );
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testStartServerAtomicPartitioned() throws Exception {
        testStartServerTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testStartServerAtomicReplicated() throws Exception {
        testStartServerTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testStartServerTransactionalPartitioned() throws Exception {
        testStartServerTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
    @Test
    public void testStartServerTransactionalReplicated() throws Exception {
        startNodesInClientMode(false);

        doTest(
            asMessagePredicate(discoEvt -> discoEvt.type() == EventType.EVT_NODE_JOINED),
            () -> startGrid(UUID.randomUUID().toString())
        );
    }

    /**
     * @throws Exception If failed.
     */
    @Params(servers = 4, atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testStopServerAtomicPartitioned() throws Exception {
        testStopServerTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(servers = 4, atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testStopServerAtomicReplicated() throws Exception {
        testStopServerTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(servers = 4, atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testStopServerTransactionalPartitioned() throws Exception {
        testStopServerTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(servers = 4, atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
    @Test
    public void testStopServerTransactionalReplicated() throws Exception {
        doTest(
            asMessagePredicate(discoEvt -> discoEvt.type() == EventType.EVT_NODE_LEFT),
            () -> stopGrid(srvs.remove(srvs.size() - 1).name())
        );
    }

    /**
     * @throws Exception If failed.
     */
    @Params(baseline = 4, atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testRestartBaselineAtomicPartitioned() throws Exception {
        testRestartBaselineTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(baseline = 4, atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testRestartBaselineAtomicReplicated() throws Exception {
        testRestartBaselineTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(baseline = 4, atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testRestartBaselineTransactionalPartitioned() throws Exception {
        testRestartBaselineTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(baseline = 4, atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
    @Test
    public void testRestartBaselineTransactionalReplicated() throws Exception {
        doTest(
            asMessagePredicate(discoEvt -> discoEvt.type() == EventType.EVT_NODE_JOINED),
            () -> {
                IgniteEx node = baseline.get(baseline.size() - 1);

                TestRecordingCommunicationSpi.spi(node).stopBlock();

                stopGrid(node.name());

                for (int i = 0; i < baselineServersCount() - 2; i++)
                    cntFinishedReadOperations.countDown();

                startGrid(node.name());
            }
        );
    }

    /**
     * @throws Exception If failed.
     */
    @Params(timeout = 5000L, atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testUpdateBaselineTopologyAtomicPartitioned() throws Exception {
        testUpdateBaselineTopologyTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(timeout = 5000L, atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testUpdateBaselineTopologyAtomicReplicated() throws Exception {
        testUpdateBaselineTopologyTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(timeout = 5000L, atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testUpdateBaselineTopologyTransactionalPartitioned() throws Exception {
        testUpdateBaselineTopologyTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(timeout = 5000L, atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
    @Test
    public void testUpdateBaselineTopologyTransactionalReplicated() throws Exception {
        doTest(
            asMessagePredicate(discoEvt -> {
                if (discoEvt instanceof DiscoveryCustomEvent) {
                    DiscoveryCustomEvent discoCustomEvt = (DiscoveryCustomEvent)discoEvt;

                    DiscoveryCustomMessage customMsg = discoCustomEvt.customMessage();

                    return customMsg instanceof ChangeGlobalStateMessage;
                }

                return false;
            }),
            () -> {
                startNodesInClientMode(false);

                IgniteEx ignite = startGrid(UUID.randomUUID().toString());

                baseline.get(0).cluster().setBaselineTopology(baseline.get(0).context().discovery().topologyVersion());

                baseline.add(ignite);
            }
        );
    }

    /**
     * @throws Exception If failed.
     */
    @Params(baseline = 9, atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testStopBaselineAtomicPartitioned() throws Exception {
        testStopBaselineTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(baseline = 9, atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testStopBaselineAtomicReplicated() throws Exception {
        testStopBaselineTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(baseline = 9, atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testStopBaselineTransactionalPartitioned() throws Exception {
        testStopBaselineTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(baseline = 9, atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
    @Test
    public void testStopBaselineTransactionalReplicated() throws Exception {
        AtomicInteger cntDownCntr = new AtomicInteger(0);

        doTest(
            asMessagePredicate(discoEvt -> discoEvt.type() == EventType.EVT_NODE_LEFT),
            () -> {
                IgniteEx node = baseline.get(baseline.size() - cntDownCntr.get() - 1);

                TestRecordingCommunicationSpi.spi(node).stopBlock();

                cntDownCntr.incrementAndGet();

                for (int i = 0; i < cntDownCntr.get(); i++)
                    cntFinishedReadOperations.countDown(); // This node and previously stopped nodes as well.

                stopGrid(node.name());
            }
        );
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testStartClientAtomicPartitioned() throws Exception {
        testStartClientTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testStartClientAtomicReplicated() throws Exception {
        testStartClientTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testStartClientTransactionalPartitioned() throws Exception {
        testStartClientTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED)
    @Test
    public void testStartClientTransactionalReplicated() throws Exception {
        doTest(
            TcpDiscoveryNodeAddFinishedMessage.class,
            () -> {
                startNodesInClientMode(true);

                customIpFinder = new TcpDiscoveryVmIpFinder(false)
                    .setAddresses(
                        Collections.singletonList("127.0.0.1:47502")
                    );

                try {
                    startGrid(UUID.randomUUID().toString());
                }
                finally {
                    customIpFinder = null;
                }
            }
        );
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = ATOMIC, cacheMode = PARTITIONED)
    @Test
    public void testStopClientAtomicPartitioned() throws Exception {
        testStopClientTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = ATOMIC, cacheMode = REPLICATED)
    @Test
    public void testStopClientAtomicReplicated() throws Exception {
        testStopClientTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = TRANSACTIONAL, cacheMode = PARTITIONED)
    @Test
    public void testStopClientTransactionalPartitioned() throws Exception {
        testStopClientTransactionalReplicated();
    }

    /**
     * @throws Exception If failed.
     */
    @Params(atomicityMode = TRANSACTIONAL, cacheMode = REPLICATED, timeout = 5_000L)
    @Test
    public void testStopClientTransactionalReplicated() throws Exception {
        startNodesInClientMode(true);

        customIpFinder = new TcpDiscoveryVmIpFinder(false)
            .setAddresses(
                Collections.singletonList("127.0.0.1:47502")
            );

        for (int i = 0; i < 3; i++)
            clients.add(startGrid(UUID.randomUUID().toString()));

        customIpFinder = null;

        doTest(
            TcpDiscoveryNodeLeftMessage.class,
            () -> stopGrid(clients.remove(clients.size() - 1).name())
        );
    }

    /**
     * Checks that given discovery event is from "Create cache" operation.
     *
     * @param discoEvt Discovery event.
     */
    private static boolean createCachePredicate(DiscoveryEvent discoEvt) {
        if (discoEvt instanceof DiscoveryCustomEvent) {

            DiscoveryCustomEvent discoCustomEvt = (DiscoveryCustomEvent)discoEvt;

            DiscoveryCustomMessage customMsg = discoCustomEvt.customMessage();

            if (customMsg instanceof DynamicCacheChangeBatch) {
                DynamicCacheChangeBatch cacheChangeBatch = (DynamicCacheChangeBatch)customMsg;

                ExchangeActions exchangeActions = U.field(cacheChangeBatch, "exchangeActions");

                Collection<CacheActionData> startRequests = exchangeActions.cacheStartRequests();

                return !startRequests.isEmpty();
            }
        }

        return false;
    }

    /**
     * Checks that given discovery event is from "Destroy cache" operation.
     *
     * @param discoEvt Discovery event.
     */
    private static boolean destroyCachePredicate(DiscoveryEvent discoEvt) {
        if (discoEvt instanceof DiscoveryCustomEvent) {

            DiscoveryCustomEvent discoCustomEvt = (DiscoveryCustomEvent)discoEvt;

            DiscoveryCustomMessage customMsg = discoCustomEvt.customMessage();

            if (customMsg instanceof DynamicCacheChangeBatch) {
                DynamicCacheChangeBatch cacheChangeBatch = (DynamicCacheChangeBatch)customMsg;

                ExchangeActions exchangeActions = U.field(cacheChangeBatch, "exchangeActions");

                Collection<CacheActionData> stopRequests = exchangeActions.cacheStopRequests();

                return !stopRequests.isEmpty();
            }
        }

        return false;
    }

    /**
     * Read operation tat is going to be executed during blocking operation.
     */
    @NotNull protected abstract CacheReadBackgroundOperation getReadOperation();

    /**
     * Checks that {@code block} closure doesn't block read operation.
     * Does it for client, baseline and regular server node.
     *
     * @param blockMsgPred Predicate that check whether the message corresponds to the {@code block} or not.
     * @param block Blocking operation.
     * @throws Exception If failed.
     */
    public void doTest(Predicate<Message> blockMsgPred, RunnableX block) throws Exception {
        BackgroundOperation backgroundOperation = new BlockMessageOnBaselineBackgroundOperation(
            block,
            blockMsgPred
        );

        doTest(backgroundOperation);
    }

    /**
     * Checks that {@code block} closure doesn't block read operation.
     * Does it for client, baseline and regular server node.
     *
     * @param blockMsgCls Class of discovery message to block.
     * @param block Blocking operation.
     * @throws Exception If failed.
     */
    public void doTest(Class<? extends TcpDiscoveryAbstractMessage> blockMsgCls, RunnableX block) throws Exception {
        BlockDiscoveryMessageBackgroundOperation backgroundOperation = new BlockDiscoveryMessageBackgroundOperation(
            block,
            blockMsgCls
        );

        doTest(backgroundOperation);
    }

    /**
     * Checks that {@code block} closure doesn't block read operation.
     * Does it for client, baseline and regular server node.
     *
     * @param backgroundOperation Background operation.
     * @throws Exception If failed.
     */
    public void doTest(BackgroundOperation backgroundOperation) throws Exception {
        CacheReadBackgroundOperation<?, ?> readOperation = getReadOperation();

        readOperation.initCache(baseline.get(0), true);

        // Warmup.
        if (warmup() > 0) {
            try (AutoCloseable read = readOperation.start()) {
                Thread.sleep(warmup());
            }

            assertEquals(
                readOperation.readOperationsFailed() + " read operations failed during warmup.",
                0,
                readOperation.readOperationsFailed()
            );

            assertTrue(
                "No read operations were finished during warmup.",
                readOperation.readOperationsFinishedUnderBlock() > 0
            );
        }

        doTest0(clients.get(0), readOperation, backgroundOperation);

        doTest0(srvs.get(0), readOperation, backgroundOperation);

        doTest0(baseline.get(0), readOperation, backgroundOperation);

        try (AutoCloseable read = readOperation.start()) {
            Thread.sleep(500L);
        }

        assertEquals(
            readOperation.readOperationsFailed() + " read operations failed during finish stage.",
            0,
            readOperation.readOperationsFailed()
        );

        assertTrue(
            "No read operations were finished during finish stage.",
            readOperation.readOperationsFinishedUnderBlock() > 0
        );
    }

    /**
     * Internal part for {@link CacheBlockOnReadAbstractTest#doTest(Predicate, RunnableX)}.
     *
     * @param ignite Ignite instance. Client / baseline / server node.
     * @param readOperation Read operation.
     * @param backgroundOperation Background operation.
     */
    private void doTest0(
        IgniteEx ignite,
        CacheReadBackgroundOperation<?, ?> readOperation,
        BackgroundOperation backgroundOperation
    ) throws Exception {
        // Reinit internal cache state with given ignite instance.
        readOperation.initCache(ignite, false);

        cntFinishedReadOperations = new CountDownLatch(baseline.size() - 1);

        // Read while potentially blocking operation is executing.
        try (AutoCloseable block = backgroundOperation.start()) {
            cntFinishedReadOperations.await(5 * timeout(), TimeUnit.MILLISECONDS);

            // Possible if test itself is wrong.
            assertEquals("Messages weren't blocked in time", 0, cntFinishedReadOperations.getCount());

            try (AutoCloseable read = readOperation.start()) {
                Thread.sleep(timeout());
            }
        }
        finally {
            cntFinishedReadOperations = null;
        }

        log.info("Operations finished: " + readOperation.readOperationsFinishedUnderBlock());
        log.info("Longest operation took " + readOperation.maxReadDuration() + "ms");

        // None of read operations should fail.
        assertEquals(
            readOperation.readOperationsFailed() + " read operations failed.",
            0,
            readOperation.readOperationsFailed()
        );

        assertTrue(
            "No read operations were finished during timeout.",
            readOperation.readOperationsFinishedUnderBlock() > 0
        );

        // There were no operations as long as blocking timeout.
        assertNotAlmostEqual(timeout(), readOperation.maxReadDuration());

        // On average every read operation was much faster then blocking timeout.
        double avgDuration = (double)timeout() / readOperation.readOperationsFinishedUnderBlock();

        assertTrue("Avarage duration was too long.", avgDuration < timeout() * 0.25);
    }

    /**
     * Utility class that allows to start and stop some background operation many times.
     */
    protected abstract static class BackgroundOperation {
        /** */
        private IgniteInternalFuture<?> fut;

        /**
         * Invoked strictly before background thread is started.
         */
        protected void init() {
            // No-op.
        }

        /**
         * Operation itself. Will be executed in separate thread. Thread interruption has to be considered as a valid
         * way to stop operation.
         */
        protected abstract void execute();

        /**
         * @return Allowed time to wait in {@link BackgroundOperation#stop()} method before canceling background thread.
         */
        protected abstract long stopTimeout();

        /**
         * Start separate thread and execute method {@link BackgroundOperation#execute()} in it.
         *
         * @return {@link AutoCloseable} that invokes {@link BackgroundOperation#stop()} on closing.
         */
        AutoCloseable start() {
            if (fut != null)
                throw new UnsupportedOperationException("Only one simultanious operation is allowed");

            init();

            CountDownLatch threadStarted = new CountDownLatch(1);

            fut = GridTestUtils.runAsync(() -> {
                try {
                    threadStarted.countDown();

                    execute();
                }
                catch (Exception e) {
                    throw new IgniteException("Unexpected exception in background operation thread", e);
                }
            });

            try {
                threadStarted.await();
            }
            catch (InterruptedException e) {
                try {
                    fut.cancel();
                }
                catch (IgniteCheckedException e1) {
                    e.addSuppressed(e1);
                }

                throw new IgniteException(e);
            }

            return this::stop;
        }

        /**
         * Interrupt the operation started in {@link BackgroundOperation#start()} method and join interrupted thread.
         */
        void stop() throws Exception {
            if (fut == null)
                return;

            try {
                fut.get(stopTimeout());
            }
            catch (IgniteFutureTimeoutCheckedException e) {
                fut.cancel();

                fut.get();
            }
            finally {
                fut = null;
            }
        }
    }

    /**
     * @param discoEvtPred Predicate that tests discovery events.
     * @return New predicate that test any message based on {@code discoEvtPred} predicate.
     */
    public static Predicate<Message> asMessagePredicate(Predicate<DiscoveryEvent> discoEvtPred) {
        return msg -> {
            if (msg instanceof GridDhtPartitionsFullMessage) {
                GridDhtPartitionsFullMessage fullMsg = (GridDhtPartitionsFullMessage)msg;

                GridDhtPartitionExchangeId exchangeId = fullMsg.exchangeId();

                if (exchangeId != null)
                    return discoEvtPred.test(U.field(exchangeId, "discoEvt"));
            }

            return false;
        };
    }

    /**
     * Background operation that executes some node request and doesn't allow its exchange messages to be fully
     * processed until operation is stopped.
     */
    protected class BlockMessageOnBaselineBackgroundOperation extends BackgroundOperation {
        /** */
        private final RunnableX block;

        /** */
        private final Predicate<Message> blockMsg;

        /**
         * @param block Blocking operation.
         * @param blockMsgPred Predicate that checks whether to block message or not.
         *
         * @see BlockMessageOnBaselineBackgroundOperation#blockMessage(ClusterNode, Message)
         */
        protected BlockMessageOnBaselineBackgroundOperation(
            RunnableX block,
            Predicate<Message> blockMsgPred
        ) {
            this.block = block;
            blockMsg = blockMsgPred;
        }

        /** {@inheritDoc} */
        @Override protected void execute() {
            for (IgniteEx server : baseline) {
                TestRecordingCommunicationSpi spi = TestRecordingCommunicationSpi.spi(server);

                spi.blockMessages(this::blockMessage);
            }

            block.run();
        }

        /**
         * Function to pass into {@link TestRecordingCommunicationSpi#blockMessages(IgniteBiPredicate)}.
         *
         * @param node Node that receives message.
         * @param msg Message.
         * @return Whether the given message should be blocked or not.
         */
        private boolean blockMessage(ClusterNode node, Message msg) {
            boolean block = blockMsg.test(msg)
                && baseline.stream().map(IgniteEx::name).anyMatch(node.consistentId()::equals);

            if (block)
                cntFinishedReadOperations.countDown();

            return block;
        }

        /** {@inheritDoc} */
        @Override protected long stopTimeout() {
            // Should be big enough so thread will stop by it's own. Otherwise test will fail, but that's fine.
            return 30_000L;
        }

        /** {@inheritDoc} */
        @Override void stop() throws Exception {
            for (IgniteEx server : baseline) {
                TestRecordingCommunicationSpi spi = TestRecordingCommunicationSpi.spi(server);

                spi.stopBlock();
            }

            super.stop();
        }
    }

    /**
     * Background operation that executes some node request and doesn't allow its discovery messages to be fully
     * processed until operation is stopped.
     */
    protected class BlockDiscoveryMessageBackgroundOperation extends BackgroundOperation {
        /** */
        private final RunnableX block;

        /** */
        private final Class<? extends TcpDiscoveryAbstractMessage> blockMsgCls;

        /** */
        private volatile CountDownLatch blockLatch;

        /**
         * @param block Blocking operation.
         * @param blockMsgCls Class of message to block.
         *
         * @see BlockMessageOnBaselineBackgroundOperation#blockMessage(ClusterNode, Message)
         */
        protected BlockDiscoveryMessageBackgroundOperation(
            RunnableX block,
            Class<? extends TcpDiscoveryAbstractMessage> blockMsgCls
        ) {
            this.block = block;
            this.blockMsgCls = blockMsgCls;
        }

        /** {@inheritDoc} */
        @Override protected void execute() {
            try {
                blockLatch = new CountDownLatch(1);

                discoveryMsgProcessor = this::processMessage;

                for (int i = 0; i < baselineServersCount() - 2; i++)
                    cntFinishedReadOperations.countDown();

                block.run();
            }
            finally {
                discoveryMsgProcessor = null;
            }
        }

        /**
         * Process discovery spi message.
         *
         * @param msg Message.
         * @param receiverConsistentId Consistent ID of message receiver.
         */
        private void processMessage(TcpDiscoveryAbstractMessage msg, String receiverConsistentId) {
            if (!blockMsgCls.isInstance(msg))
                return;

            boolean baselineSnd = Objects.equals(
                baseline.get(1).localNode().consistentId(),
                receiverConsistentId
            );

            if (baselineSnd) {
                cntFinishedReadOperations.countDown();

                try {
                    blockLatch.await();
                }
                catch (InterruptedException ignore) {
                }
            }
        }

        /** {@inheritDoc} */
        @Override protected long stopTimeout() {
            // Should be big enough so thread will stop by it's own. Otherwise test will fail, but that's fine.
            return 30_000L;
        }

        /** {@inheritDoc} */
        @Override void stop() throws Exception {
            blockLatch.countDown();

            super.stop();
        }
    }

    /**
     * {@link BackgroundOperation} implementation for cache reading operations.
     */
    protected abstract class ReadBackgroundOperation extends BackgroundOperation {

        /** Counter for successfully finished operations. */
        private final AtomicInteger readOperationsFinishedUnderBlock = new AtomicInteger();

        /** Counter for failed operations. */
        private final AtomicInteger readOperationsFailed = new AtomicInteger();

        /** Duration of the longest read operation. */
        private final AtomicLong maxReadDuration = new AtomicLong(-1);

        /**
         * Do single iteration of reading operation. Will be executed in a loop.
         */
        protected abstract void doRead() throws Exception;


        /** {@inheritDoc} */
        @Override protected void init() {
            readOperationsFinishedUnderBlock.set(0);

            readOperationsFailed.set(0);

            maxReadDuration.set(-1);
        }

        /** {@inheritDoc} */
        @Override protected void execute() {
            Set<String> loggedMsgs = new HashSet<>();

            while (!Thread.currentThread().isInterrupted()) {
                long prevTs = System.currentTimeMillis();

                try {
                    doRead();

                    readOperationsFinishedUnderBlock.incrementAndGet();
                }
                catch (Exception e) {
                    boolean threadInterrupted = X.hasCause(e,
                        InterruptedException.class,
                        IgniteInterruptedException.class,
                        IgniteInterruptedCheckedException.class
                    );

                    if (threadInterrupted)
                        Thread.currentThread().interrupt();
                    else if (allowException() && X.hasCause(e, ClusterTopologyCheckedException.class))
                        readOperationsFinishedUnderBlock.incrementAndGet();
                    else {
                        readOperationsFailed.incrementAndGet();

                        if (loggedMsgs.add(e.getMessage()))
                            log.error("Error during read operation execution", e);

                        continue;
                    }
                }

                maxReadDuration.set(Math.max(maxReadDuration.get(), System.currentTimeMillis() - prevTs));
            }
        }

        /** {@inheritDoc} */
        @Override protected long stopTimeout() {
            return 0;
        }

        /**
         * @return Number of successfully finished operations.
         */
        public int readOperationsFinishedUnderBlock() {
            return readOperationsFinishedUnderBlock.get();
        }

        /**
         * @return Number of failed operations.
         */
        public int readOperationsFailed() {
            return readOperationsFailed.get();
        }

        /**
         * @return Duration of the longest read operation.
         */
        public long maxReadDuration() {
            return maxReadDuration.get();
        }
    }

    /**
     *
     */
    protected abstract class CacheReadBackgroundOperation<KeyType, ValueType> extends ReadBackgroundOperation {
        /**
         * {@link CacheReadBackgroundOperation#cache()} method backing field. Updated on each
         * {@link CacheReadBackgroundOperation#initCache(IgniteEx, boolean)} invocation.
         */
        private IgniteCache<KeyType, ValueType> cache;

        /**
         * Reinit internal cache using passed ignite instance and fill it with data if required.
         *
         * @param ignite Node to get or create cache from.
         * @param fillData Whether the cache should be filled with new data or not.
         */
        public void initCache(IgniteEx ignite, boolean fillData) {
            cache = ignite.getOrCreateCache(
                createCacheConfiguration()
                    .setAtomicityMode(atomicityMode())
                    .setCacheMode(cacheMode())
            );

            if (fillData) {
                try (IgniteDataStreamer<KeyType, ValueType> dataStreamer = ignite.dataStreamer(cache.getName())) {
                    dataStreamer.allowOverwrite(true);

                    for (int i = 0; i < entriesCount(); i++)
                        dataStreamer.addData(createKey(i), createValue(i));
                }
            }
        }

        /**
         * @return Cache configuration.
         */
        protected CacheConfiguration<KeyType, ValueType> createCacheConfiguration() {
            return new CacheConfiguration<KeyType, ValueType>(DEFAULT_CACHE_NAME)
                .setBackups(backupsCount())
                .setAffinity(
                    new RendezvousAffinityFunction()
                        .setPartitions(32)
                );
        }

        /**
         * @return Current cache.
         */
        protected final IgniteCache<KeyType, ValueType> cache() {
            return cache;
        }

        /**
         * @return Count of cache entries to create in {@link CacheReadBackgroundOperation#initCache(IgniteEx, boolean)}
         * method.
         */
        protected int entriesCount() {
            return DFLT_CACHE_ENTRIES_CNT;
        }

        /**
         * @param idx Unique number.
         * @return Key to be used for inserting into cache.
         * @see CacheReadBackgroundOperation#createValue(int)
         */
        protected abstract KeyType createKey(int idx);

        /**
         * @param idx Unique number.
         * @return Value to be used for inserting into cache.
         * @see CacheReadBackgroundOperation#createKey(int)
         */
        protected abstract ValueType createValue(int idx);
    }

    /**
     * {@link CacheReadBackgroundOperation} implementation for (int -> int) cache. Keys and values are equal by default.
     */
    protected abstract class IntCacheReadBackgroundOperation
        extends CacheReadBackgroundOperation<Integer, Integer> {
        /** {@inheritDoc} */
        @Override protected Integer createKey(int idx) {
            return idx;
        }

        /** {@inheritDoc} */
        @Override protected Integer createValue(int idx) {
            return idx;
        }
    }

    /**
     * @return {@link Params} annotation object from the current test method.
     */
    protected Params currentTestParams() {
        Params params = currentTestAnnotation(Params.class);

        assertNotNull("Test " + getName() + " is not annotated with @Param annotation.", params);

        return params;
    }

    /**
     * Assert that two numbers are close to each other.
     */
    private static void assertAlmostEqual(long exp, long actual) {
        assertTrue(String.format("Numbers differ too much [exp=%d, actual=%d]", exp, actual), almostEqual(exp, actual));
    }

    /**
     * Assert that two numbers are not close to each other.
     */
    private static void assertNotAlmostEqual(long exp, long actual) {
        assertFalse(String.format("Numbers are almost equal [exp=%d, actual=%d]", exp, actual), almostEqual(exp, actual));
    }

    /**
     * Check that two numbers are close to each other.
     */
    private static boolean almostEqual(long exp, long actual) {
        double rel = (double)(actual - exp) / exp;

        return Math.abs(rel) < 0.05;
    }
}
