/*
 * 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.spi.discovery.tcp;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Callable;
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 org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteIllegalStateException;
import org.apache.ignite.Ignition;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.events.Event;
import org.apache.ignite.events.EventType;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.processors.continuous.StartRoutineAckDiscoveryMessage;
import org.apache.ignite.internal.processors.port.GridPortRecord;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
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.lang.IgniteInClosure;
import org.apache.ignite.lang.IgnitePredicate;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
import org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder;
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.TcpDiscoveryCustomEventMessage;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessage;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddedMessage;
import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeFailedMessage;
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.eclipse.jetty.util.ConcurrentHashSet;
import org.jetbrains.annotations.Nullable;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.ignite.events.EventType.EVT_JOB_MAPPED;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static org.apache.ignite.events.EventType.EVT_NODE_METRICS_UPDATED;
import static org.apache.ignite.events.EventType.EVT_TASK_FAILED;
import static org.apache.ignite.events.EventType.EVT_TASK_FINISHED;
import static org.apache.ignite.spi.IgnitePortProtocol.UDP;

/**
 * Test for {@link TcpDiscoverySpi}.
 */
public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
    /** */
    private TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);

    /** */
    private Map<String, TcpDiscoverySpi> discoMap = new HashMap<>();

    /** */
    private UUID nodeId;

    /** */
    private static ThreadLocal<TcpDiscoverySpi> nodeSpi = new ThreadLocal<>();

    /**
     * @throws Exception If fails.
     */
    public TcpDiscoverySelfTest() throws Exception {
        super(false);
    }

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

        TcpDiscoverySpi spi = nodeSpi.get();

        if (spi == null) {
            spi = gridName.contains("testPingInterruptedOnNodeFailedFailingNode") ?
                new TestTcpDiscoverySpi() : new TcpDiscoverySpi();
        }
        else
            nodeSpi.set(null);

        discoMap.put(gridName, spi);

        spi.setIpFinder(ipFinder);

        spi.setNetworkTimeout(2500);

        spi.setHeartbeatFrequency(1000);

        spi.setMaxMissedHeartbeats(3);

        spi.setIpFinderCleanFrequency(5000);

        spi.setJoinTimeout(5000);

        cfg.setDiscoverySpi(spi);

        cfg.setCacheConfiguration();

        cfg.setIncludeEventTypes(EVT_TASK_FAILED, EVT_TASK_FINISHED, EVT_JOB_MAPPED);

        cfg.setIncludeProperties();

        if (!gridName.contains("LoopbackProblemTest"))
            cfg.setLocalHost("127.0.0.1");

        if (gridName.contains("testFailureDetectionOnNodePing")) {
            spi.setReconnectCount(1); // To make test faster: on Windows 1 connect takes 1 second.
            spi.setHeartbeatFrequency(40000);
        }

        cfg.setConnectorConfiguration(null);

        if (nodeId != null)
            cfg.setNodeId(nodeId);

        if (gridName.contains("NonSharedIpFinder")) {
            TcpDiscoveryVmIpFinder finder = new TcpDiscoveryVmIpFinder();

            finder.setAddresses(Arrays.asList("127.0.0.1:47501"));

            spi.setIpFinder(finder);
        }
        else if (gridName.contains("MulticastIpFinder")) {
            TcpDiscoveryMulticastIpFinder finder = new TcpDiscoveryMulticastIpFinder();

            finder.setAddressRequestAttempts(10);
            finder.setMulticastGroup(GridTestUtils.getNextMulticastGroup(getClass()));
            finder.setMulticastPort(GridTestUtils.getNextMulticastPort(getClass()));

            spi.setIpFinder(finder);

            // Loopback multicast discovery is not working on Mac OS
            // (possibly due to http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=7122846).
            if (U.isMacOs())
                spi.setLocalAddress(F.first(U.allLocalIps()));
        }
        else if (gridName.contains("testPingInterruptedOnNodeFailedPingingNode"))
            cfg.setFailureDetectionTimeout(30_000);

        return cfg;
    }

    /** {@inheritDoc} */
    @Override protected void afterTest() throws Exception {
        discoMap = null;

        super.afterTest();
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testSingleNodeStartStop() throws Exception {
        try {
            startGrid(1);
        }
        finally {
            stopGrid(1);
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testThreeNodesStartStop() throws Exception {
        try {
            IgniteEx ignite1 = startGrid(1);
            IgniteEx ignite2 = startGrid(2);
            IgniteEx ignite3 = startGrid(3);

            TcpDiscoverySpi spi1 = (TcpDiscoverySpi)ignite1.configuration().getDiscoverySpi();
            TcpDiscoverySpi spi2 = (TcpDiscoverySpi)ignite2.configuration().getDiscoverySpi();
            TcpDiscoverySpi spi3 = (TcpDiscoverySpi)ignite3.configuration().getDiscoverySpi();

            TcpDiscoveryNode node = (TcpDiscoveryNode)spi1.getNode(ignite2.localNode().id());

            assertNotNull(node);
            assertNotNull(node.lastSuccessfulAddress());

            LinkedHashSet<InetSocketAddress> addrs = spi1.getNodeAddresses(node);

            assertEquals(addrs.iterator().next(), node.lastSuccessfulAddress());

            assertTrue(spi1.pingNode(ignite3.localNode().id()));

            node = (TcpDiscoveryNode)spi1.getNode(ignite3.localNode().id());

            assertNotNull(node);
            assertNotNull(node.lastSuccessfulAddress());

            addrs = spi1.getNodeAddresses(node);
            assertEquals(addrs.iterator().next(), node.lastSuccessfulAddress());

            node = (TcpDiscoveryNode)spi2.getNode(ignite1.localNode().id());

            assertNotNull(node);
            assertNotNull(node.lastSuccessfulAddress());

            node = (TcpDiscoveryNode)spi2.getNode(ignite3.localNode().id());

            assertNotNull(node);
            assertNotNull(node.lastSuccessfulAddress());

            node = (TcpDiscoveryNode)spi3.getNode(ignite1.localNode().id());

            assertNotNull(node);
            assertNotNull(node.lastSuccessfulAddress());
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any errors occur.
     */
    public void testNodeConnectMessageSize() throws Exception {
        try {
            Ignite g1 = startGrid(1);

            final AtomicInteger gridNameIdx = new AtomicInteger(1);

            GridTestUtils.runMultiThreaded(new Callable<Object>() {
                @Nullable @Override public Object call() throws Exception {
                    startGrid(gridNameIdx.incrementAndGet());

                    return null;
                }
            }, 4, "grid-starter");

            Collection<TcpDiscoveryNode> nodes = ((ServerImpl)discoMap.get(g1.name()).impl).ring().allNodes();

            ByteArrayOutputStream bos = new ByteArrayOutputStream();

            g1.configuration().getMarshaller().marshal(nodes, bos);

            info(">>> Approximate node connect message size [topSize=" + nodes.size() +
                ", msgSize=" + bos.size() / 1024.0 + "KB]");
        }
        finally {
            stopAllGrids(false);
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testPing() throws Exception {
        try {
            startGrid(1);
            startGrid(2);
            startGrid(3);

            info("Nodes were started");

            for (Map.Entry<String, TcpDiscoverySpi> e : discoMap.entrySet()) {
                DiscoverySpi spi = e.getValue();

                for (Ignite g : G.allGrids()) {
                    boolean res = spi.pingNode(g.cluster().localNode().id());

                    assert res : e.getKey() + " failed to ping " + g.cluster().localNode().id() + " of " + g.name();

                    info(e.getKey() + " pinged " + g.cluster().localNode().id() + " of " + g.name());
                }
            }

            info("All nodes pinged successfully.");
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testFailureDetectionOnNodePing1() throws Exception {
        try {
            Ignite g1 = startGrid("testFailureDetectionOnNodePingCoordinator");
            startGrid("testFailureDetectionOnNodePing2");
            Ignite g3 = startGrid("testFailureDetectionOnNodePing3");

            testFailureDetectionOnNodePing(g1, g3);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testFailureDetectionOnNodePing2() throws Exception {
        try {
            startGrid("testFailureDetectionOnNodePingCoordinator");
            Ignite g2 = startGrid("testFailureDetectionOnNodePing2");
            Ignite g3 = startGrid("testFailureDetectionOnNodePing3");

            testFailureDetectionOnNodePing(g3, g2);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testFailureDetectionOnNodePing3() throws Exception {
        try {
            Ignite g1 = startGrid("testFailureDetectionOnNodePingCoordinator");
            Ignite g2 = startGrid("testFailureDetectionOnNodePing2");
            startGrid("testFailureDetectionOnNodePing3");

            testFailureDetectionOnNodePing(g2, g1);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    private void testFailureDetectionOnNodePing(Ignite pingingNode, Ignite failedNode) throws Exception {
        final CountDownLatch cnt = new CountDownLatch(1);

        final UUID failedNodeId = failedNode.cluster().localNode().id();

        pingingNode.events().localListen(
            new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    cnt.countDown();

                    return true;
                }
            },
            EventType.EVT_NODE_FAILED
        );

        info("Nodes were started");

        discoMap.get(failedNode.name()).simulateNodeFailure();

        TcpDiscoverySpi spi = discoMap.get(pingingNode.name());

        boolean res = spi.pingNode(failedNodeId);

        assertFalse("Ping is ok for node " + failedNodeId + ", but had to fail.", res);

        // Heartbeat interval is 40 seconds, but we should detect node failure faster.
        assert cnt.await(7, SECONDS);
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testPingInterruptedOnNodeFailed() throws Exception {
        try {
            final Ignite pingingNode = startGrid("testPingInterruptedOnNodeFailedPingingNode");
            final Ignite failedNode = startGrid("testPingInterruptedOnNodeFailedFailingNode");
            startGrid("testPingInterruptedOnNodeFailedSimpleNode");

            ((TestTcpDiscoverySpi)failedNode.configuration().getDiscoverySpi()).ignorePingResponse = true;

            final UUID failedNodeId = failedNode.cluster().localNode().id();

            final CountDownLatch pingLatch = new CountDownLatch(1);

            final CountDownLatch eventLatch = new CountDownLatch(1);

            final AtomicBoolean pingRes = new AtomicBoolean(true);

            final AtomicBoolean failRes = new AtomicBoolean(false);

            long startTs = System.currentTimeMillis();

            pingingNode.events().localListen(
                new IgnitePredicate<Event>() {
                    @Override public boolean apply(Event event) {
                        if (((DiscoveryEvent)event).eventNode().id().equals(failedNodeId)) {
                            failRes.set(true);
                            eventLatch.countDown();
                        }

                        return true;
                    }
                },
                EventType.EVT_NODE_FAILED);

            IgniteInternalFuture<?> pingFut = multithreadedAsync(
                new Callable<Object>() {
                    @Override public Object call() throws Exception {
                        pingLatch.countDown();

                        pingRes.set(pingingNode.configuration().getDiscoverySpi().pingNode(
                            failedNodeId));

                        return null;
                    }
                }, 1);

            IgniteInternalFuture<?> failingFut = multithreadedAsync(
                new Callable<Object>() {
                    @Override public Object call() throws Exception {
                        pingLatch.await();

                        Thread.sleep(3000);

                        ((TestTcpDiscoverySpi)failedNode.configuration().getDiscoverySpi()).simulateNodeFailure();

                        return null;
                    }
                }, 1);

            failingFut.get();
            pingFut.get();

            assertFalse(pingRes.get());

            assertTrue(System.currentTimeMillis() - startTs <
                pingingNode.configuration().getFailureDetectionTimeout() / 2);

            assertTrue(eventLatch.await(7, TimeUnit.SECONDS));
            assertTrue(failRes.get());
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testPingInterruptedOnNodeLeft() throws Exception {
        try {
            final Ignite pingingNode = startGrid("testPingInterruptedOnNodeFailedPingingNode");
            final Ignite leftNode = startGrid("testPingInterruptedOnNodeFailedFailingNode");
            startGrid("testPingInterruptedOnNodeFailedSimpleNode");

            ((TestTcpDiscoverySpi)leftNode.configuration().getDiscoverySpi()).ignorePingResponse = true;

            final CountDownLatch pingLatch = new CountDownLatch(1);

            final AtomicBoolean pingRes = new AtomicBoolean(true);

            long startTs = System.currentTimeMillis();

            IgniteInternalFuture<?> pingFut = multithreadedAsync(
                new Callable<Object>() {
                    @Override public Object call() throws Exception {
                        pingLatch.countDown();

                        pingRes.set(pingingNode.configuration().getDiscoverySpi().pingNode(
                            leftNode.cluster().localNode().id()));

                        return null;
                    }
                }, 1);

            IgniteInternalFuture<?> stoppingFut = multithreadedAsync(
                new Callable<Object>() {
                    @Override public Object call() throws Exception {
                        pingLatch.await();

                        Thread.sleep(3000);

                        stopGrid("testPingInterruptedOnNodeFailedFailingNode");

                        return null;
                    }
                }, 1);

            stoppingFut.get();
            pingFut.get();

            assertFalse(pingRes.get());

            assertTrue(System.currentTimeMillis() - startTs <
                pingingNode.configuration().getFailureDetectionTimeout() / 2);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testNodeAdded() throws Exception {
        try {
            final Ignite g1 = startGrid(1);

            final CountDownLatch cnt = new CountDownLatch(2);

            g1.events().localListen(
                new IgnitePredicate<Event>() {
                    @Override public boolean apply(Event evt) {
                        info("Node joined: " + evt.message());

                        DiscoveryEvent discoEvt = (DiscoveryEvent)evt;

                        TcpDiscoveryNode node = ((TcpDiscoveryNode)discoMap.get(g1.name()).
                            getNode(discoEvt.eventNode().id()));

                        assert node != null && node.visible();

                        cnt.countDown();

                        return true;
                    }
                },
                EventType.EVT_NODE_JOINED
            );

            startGrid(2);
            startGrid(3);

            info("Nodes were started");

            assert cnt.await(1, SECONDS);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testOrdinaryNodeLeave() throws Exception {
        try {
            Ignite g1 = startGrid(1);
            startGrid(2);
            startGrid(3);

            final CountDownLatch cnt = new CountDownLatch(2);

            g1.events().localListen(
                new IgnitePredicate<Event>() {
                    @Override public boolean apply(Event evt) {
                        cnt.countDown();

                        return true;
                    }
                },
                EVT_NODE_LEFT, EVT_NODE_FAILED);

            info("Nodes were started");

            stopGrid(3);
            stopGrid(2);

            boolean res = cnt.await(1, SECONDS);

            assert res;
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testCoordinatorNodeLeave() throws Exception {
        try {
            startGrid(1);
            Ignite g2 = startGrid(2);

            final CountDownLatch cnt = new CountDownLatch(1);

            g2.events().localListen(new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    cnt.countDown();

                    return true;
                }
            }, EVT_NODE_LEFT, EVT_NODE_FAILED);

            info("Nodes were started");

            stopGrid(1);

            assert cnt.await(1, SECONDS);

            // Start new grid, ensure that added to topology
            final CountDownLatch cnt2 = new CountDownLatch(1);

            g2.events().localListen(new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    cnt2.countDown();

                    return true;
                }
            }, EVT_NODE_JOINED);

            startGrid(3);

            assert cnt2.await(1, SECONDS);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testOrdinaryNodeFailure() throws Exception {
        try {
            Ignite g1 = startGrid(1);
            Ignite g2 = startGrid(2);
            Ignite g3 = startGrid(3);

            final CountDownLatch cnt = new CountDownLatch(2);

            g1.events().localListen(
                new IgnitePredicate<Event>() {
                    @Override public boolean apply(Event evt) {
                        cnt.countDown();

                        return true;
                    }
                },
                EventType.EVT_NODE_FAILED
            );

            info("Nodes were started");

            discoMap.get(g2.name()).simulateNodeFailure();
            discoMap.get(g3.name()).simulateNodeFailure();

            assert cnt.await(25, SECONDS);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testCoordinatorNodeFailure() throws Exception {
        try {
            Ignite g1 = startGrid(1);
            Ignite g2 = startGrid(2);

            final CountDownLatch cnt = new CountDownLatch(1);

            g2.events().localListen(new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    cnt.countDown();

                    return true;
                }
            }, EventType.EVT_NODE_FAILED);

            info("Nodes were started");

            discoMap.get(g1.name()).simulateNodeFailure();

            assert cnt.await(20, SECONDS);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testMetricsSending() throws Exception {
        final AtomicBoolean stopping = new AtomicBoolean();

        try {
            final CountDownLatch latch1 = new CountDownLatch(1);

            final Ignite g1 = startGrid(1);

            IgnitePredicate<Event> lsnr1 = new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    info(evt.message());

                    latch1.countDown();

                    return true;
                }
            };

            g1.events().localListen(lsnr1, EVT_NODE_METRICS_UPDATED);

            assert latch1.await(10, SECONDS);

            g1.events().stopLocalListen(lsnr1);

            final CountDownLatch latch1_1 = new CountDownLatch(1);
            final CountDownLatch latch1_2 = new CountDownLatch(1);
            final CountDownLatch latch2_1 = new CountDownLatch(1);
            final CountDownLatch latch2_2 = new CountDownLatch(1);

            final Ignite g2 = startGrid(2);

            g2.events().localListen(
                new IgnitePredicate<Event>() {
                    @Override public boolean apply(Event evt) {
                        if (stopping.get())
                            return true;

                        info(evt.message());

                        UUID id = ((DiscoveryEvent) evt).eventNode().id();

                        if (id.equals(g1.cluster().localNode().id()))
                            latch2_1.countDown();
                        else if (id.equals(g2.cluster().localNode().id()))
                            latch2_2.countDown();
                        else
                            assert false : "Event fired for unknown node.";

                        return true;
                    }
                },
                EVT_NODE_METRICS_UPDATED
            );

            g1.events().localListen(new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    if (stopping.get())
                        return true;

                    info(evt.message());

                    UUID id = ((DiscoveryEvent) evt).eventNode().id();

                    if (id.equals(g1.cluster().localNode().id()))
                        latch1_1.countDown();
                    else if (id.equals(g2.cluster().localNode().id()))
                        latch1_2.countDown();
                    else
                        assert false : "Event fired for unknown node.";

                    return true;
                }
            }, EVT_NODE_METRICS_UPDATED);

            assert latch1_1.await(10, SECONDS);
            assert latch1_2.await(10, SECONDS);
            assert latch2_1.await(10, SECONDS);
            assert latch2_2.await(10, SECONDS);
        }
        finally {
            stopping.set(true);

            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testFailBeforeNodeAddedSent() throws Exception {
        try {
            Ignite g1 = startGrid(1);

            final CountDownLatch joinCnt = new CountDownLatch(2);
            final CountDownLatch failCnt = new CountDownLatch(1);

            g1.events().localListen(new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    if (evt.type() == EVT_NODE_JOINED)
                        joinCnt.countDown();
                    else if (evt.type() == EVT_NODE_FAILED)
                        failCnt.countDown();
                    else
                        assert false : "Unexpected event type: " + evt;

                    return true;
                }
            }, EVT_NODE_JOINED, EVT_NODE_FAILED);

            final Ignite g = startGrid("FailBeforeNodeAddedSentSpi");

            discoMap.get(g.name()).addSendMessageListener(new IgniteInClosure<TcpDiscoveryAbstractMessage>() {
                @Override public void apply(TcpDiscoveryAbstractMessage msg) {
                    if (msg instanceof TcpDiscoveryNodeAddedMessage) {
                        discoMap.get(g.name()).simulateNodeFailure();

                        throw new RuntimeException("Avoid message sending: " + msg.getClass());
                    }
                }
            });

            startGrid(3);

            assert joinCnt.await(10, SECONDS);
            assert failCnt.await(10, SECONDS);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testFailBeforeNodeLeftSent() throws Exception {
        try {
            startGrid(1);
            startGrid(2);

            final Ignite g = startGrid("FailBeforeNodeLeftSentSpi");

            discoMap.get(g.name()).addSendMessageListener(new IgniteInClosure<TcpDiscoveryAbstractMessage>() {
                @Override public void apply(TcpDiscoveryAbstractMessage msg) {
                    if (msg instanceof TcpDiscoveryNodeLeftMessage) {
                        discoMap.get(g.name()).simulateNodeFailure();

                        throw new RuntimeException("Avoid message sending: " + msg.getClass());
                    }
                }
            });

            Ignite g3 = startGrid(3);

            final CountDownLatch cnt = new CountDownLatch(1);

            g3.events().localListen(new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    cnt.countDown();

                    return true;
                }
            }, EVT_NODE_FAILED);

            stopGrid(1);

            assert cnt.await(20, SECONDS);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testIpFinderCleaning() throws Exception {
        try {
            ipFinder.registerAddresses(Arrays.asList(new InetSocketAddress("1.1.1.1", 1024),
                new InetSocketAddress("1.1.1.2", 1024)));

            Ignite g1 = startGrid(1);

            long failureDetectTimeout = g1.configuration().getFailureDetectionTimeout();

            long timeout = (long)(discoMap.get(g1.name()).getIpFinderCleanFrequency() * 1.5) + failureDetectTimeout;

            GridTestUtils.waitForCondition(new GridAbsPredicate() {
                @Override public boolean apply() {
                    return ipFinder.getRegisteredAddresses().size() == 1;
                }
            }, timeout);

            if (ipFinder.getRegisteredAddresses().size() != 1) {
                log.error("Failed to wait for IP cleanup, will dump threads.");

                U.dumpThreads(log);
            }

            assert ipFinder.getRegisteredAddresses().size() == 1 : "ipFinder=" + ipFinder.getRegisteredAddresses();

            // Check that missing addresses are returned back.
            ipFinder.unregisterAddresses(ipFinder.getRegisteredAddresses()); // Unregister valid address.

            ipFinder.registerAddresses(Arrays.asList(new InetSocketAddress("1.1.1.1", 1024),
                new InetSocketAddress("1.1.1.2", 1024)));

            GridTestUtils.waitForCondition(new GridAbsPredicate() {
                @Override public boolean apply() {
                    return ipFinder.getRegisteredAddresses().size() == 1;
                }
            }, timeout);

            assert ipFinder.getRegisteredAddresses().size() == 1 : "ipFinder=" + ipFinder.getRegisteredAddresses();
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testNonSharedIpFinder() throws Exception {
        try {
            GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
                @Override public Object call() throws Exception {
                    Thread.sleep(4000);

                    return startGrid("NonSharedIpFinder-2");
                }
            }, 1, "grid-starter");

            // This node should wait until any node "from ipFinder" appears, see log messages.
            Ignite g = startGrid("NonSharedIpFinder-1");

            assert g.cluster().localNode().order() == 2;
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testMulticastIpFinder() throws Exception {
        try {
            for (int i = 0; i < 5; i++) {
                Ignite g = startGrid("MulticastIpFinder-" + i);

                assertEquals(i + 1, g.cluster().nodes().size());

                TcpDiscoverySpi spi = (TcpDiscoverySpi)g.configuration().getDiscoverySpi();

                TcpDiscoveryMulticastIpFinder ipFinder = (TcpDiscoveryMulticastIpFinder)spi.getIpFinder();

                boolean found = false;

                for (GridPortRecord rec : ((IgniteKernal) g).context().ports().records()) {
                    if ((rec.protocol() == UDP) && rec.port() == ipFinder.getMulticastPort()) {
                        found = true;

                        break;
                    }
                }

                assertTrue("TcpDiscoveryMulticastIpFinder should register port." , found);
            }
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testInvalidAddressIpFinder() throws Exception {
        ipFinder.setShared(false);

        ipFinder.setAddresses(Collections.singletonList("some-host"));

        try {
            GridTestUtils.assertThrows(
                log,
                new Callable<Object>() {
                    @Nullable @Override public Object call() throws Exception {
                        startGrid(1);

                        return null;
                    }
                },
                IgniteCheckedException.class,
                null);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testJoinTimeout() throws Exception {
        try {
            // This start will fail as expected.
            Throwable t = GridTestUtils.assertThrows(log, new Callable<Object>() {
                @Override public Object call() throws Exception {
                    startGrid("NonSharedIpFinder-1");

                    return null;
                }
            }, IgniteCheckedException.class, null);

            assert X.hasCause(t, IgniteSpiException.class) : "Unexpected exception: " + t;
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If failed.
     */
    public void testDirtyIpFinder() throws Exception {
        try {
            // Dirty IP finder
            for (int i = 47500; i < 47520; i++)
                ipFinder.registerAddresses(Arrays.asList(new InetSocketAddress("127.0.0.1", i),
                    new InetSocketAddress("unknown-host", i)));

            assert ipFinder.isShared();

            startGrid(1);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testDuplicateId() throws Exception {
        try {
            // Random ID.
            startGrid(1);

            nodeId = UUID.randomUUID();

            startGrid(2);

            // Duplicate ID.
            GridTestUtils.assertThrows(
                log,
                new Callable<Object>() {
                    @Nullable @Override public Object call() throws Exception {
                        // Exception will be thrown and output to log.
                        startGrid(3);

                        return null;
                    }
                },
                IgniteCheckedException.class,
                null);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testLoopbackProblemFirstNodeOnLoopback() throws Exception {
        // On Windows and Mac machines two nodes can reside on the same port
        // (if one node has localHost="127.0.0.1" and another has localHost="0.0.0.0").
        // So two nodes do not even discover each other.
        if (U.isWindows() || U.isMacOs())
            return;

        try {
            startGridNoOptimize(1);

            GridTestUtils.assertThrows(
                log,
                new Callable<Object>() {
                    @Nullable @Override public Object call() throws Exception {
                        // Exception will be thrown because we start node which does not use loopback address,
                        // but the first node does.
                        startGridNoOptimize("LoopbackProblemTest");

                        return null;
                    }
                },
                IgniteException.class,
                null);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testLoopbackProblemSecondNodeOnLoopback() throws Exception {
        if (U.isWindows() || U.isMacOs())
            return;

        try {
            startGridNoOptimize("LoopbackProblemTest");

            GridTestUtils.assertThrows(
                log,
                new Callable<Object>() {
                    @Nullable @Override public Object call() throws Exception {
                        // Exception will be thrown because we start node which uses loopback address,
                        // but the first node does not.
                        startGridNoOptimize(1);

                        return null;
                    }
                },
                IgniteException.class,
                null);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If any error occurs.
     */
    public void testGridStartTime() throws Exception {
        try {
            startGridsMultiThreaded(5);

            Long startTime = null;

            IgniteKernal firstGrid = null;

            Collection<IgniteKernal> grids = new ArrayList<>();

            for (int i = 0; i < 5 ; i++) {
                IgniteKernal grid = (IgniteKernal)grid(i);

                assertTrue(grid.context().discovery().gridStartTime() > 0);

                if (i > 0)
                    assertEquals(startTime, (Long)grid.context().discovery().gridStartTime());
                else
                    startTime = grid.context().discovery().gridStartTime();

                if (grid.localNode().order() == 1)
                    firstGrid = grid;
                else
                    grids.add(grid);
            }

            assertNotNull(firstGrid);

            stopGrid(firstGrid.name());

            for (IgniteKernal grid : grids)
                assertEquals(startTime, (Long)grid.context().discovery().gridStartTime());

            grids.add((IgniteKernal)startGrid(5));

            for (IgniteKernal grid : grids)
                assertEquals(startTime, (Long)grid.context().discovery().gridStartTime());
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If failed
     */
    public void testCustomEventRace1_1() throws Exception {
        try {
            customEventRace1(true, false);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If failed
     */
    public void testCustomEventRace1_2() throws Exception {
        try {
            customEventRace1(false, false);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If failed
     */
    public void testCustomEventRace1_3() throws Exception {
        try {
            customEventRace1(true, true);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @param cacheStartFrom1 If {code true} starts cache from node1.
     * @param stopCrd If {@code true} stops coordinator.
     * @throws Exception If failed
     */
    private void customEventRace1(final boolean cacheStartFrom1, boolean stopCrd) throws Exception {
        TestCustomEventRaceSpi spi0 = new TestCustomEventRaceSpi();

        nodeSpi.set(spi0);

        final Ignite ignite0 = startGrid(0);

        nodeSpi.set(new TestCustomEventRaceSpi());

        final Ignite ignite1 = startGrid(1);

        CountDownLatch latch1 = new CountDownLatch(1);
        CountDownLatch latch2 = new CountDownLatch(1);

        spi0.nodeAdded1 = latch1;
        spi0.nodeAdded2 = latch2;
        spi0.debug = true;

        IgniteInternalFuture<?> fut1 = GridTestUtils.runAsync(new Callable<Void>() {
            @Override public Void call() throws Exception {
                log.info("Start 2");

                nodeSpi.set(new TestCustomEventRaceSpi());

                Ignite ignite2 = startGrid(2);

                return null;
            }
        });

        latch1.await();

        final String CACHE_NAME = "cache";

        IgniteInternalFuture<?> fut2 = GridTestUtils.runAsync(new Callable<Void>() {
            @Override public Void call() throws Exception {
                CacheConfiguration ccfg = new CacheConfiguration();

                ccfg.setName(CACHE_NAME);

                Ignite ignite = cacheStartFrom1 ? ignite1 : ignite0;

                ignite.createCache(ccfg);

                return null;
            }
        });

        if (stopCrd) {
            spi0.stop = true;

            latch2.countDown();

            ignite0.close();
        }
        else {
            U.sleep(500);

            latch2.countDown();
        }

        fut1.get();
        fut2.get();

        IgniteCache<Object, Object> cache = grid(2).cache(CACHE_NAME);

        assertNotNull(cache);

        cache.put(1, 1);

        assertEquals(1, cache.get(1));

        nodeSpi.set(new TestCustomEventRaceSpi());

        Ignite ignite = startGrid(3);

        cache = ignite.cache(CACHE_NAME);

        cache.put(2, 2);

        assertEquals(1, cache.get(1));
        assertEquals(2, cache.get(2));
    }

    /**
     * @throws Exception If failed
     */
    public void testCustomEventCoordinatorFailure1() throws Exception {
        try {
            customEventCoordinatorFailure(true);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If failed
     */
    public void testCustomEventCoordinatorFailure2() throws Exception {
        try {
            customEventCoordinatorFailure(false);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If failed
     */
    public void testNodeShutdownOnRingMessageWorkerFailure() throws Exception {
        try {
            TestMessageWorkerFailureSpi1 spi0 = new TestMessageWorkerFailureSpi1();

            nodeSpi.set(spi0);

            final Ignite ignite0 = startGrid(0);

            nodeSpi.set(new TcpDiscoverySpi());

            Ignite ignite1 = startGrid(1);

            final AtomicBoolean disconnected = new AtomicBoolean();

            final CountDownLatch latch = new CountDownLatch(1);

            final UUID failedNodeId = ignite0.cluster().localNode().id();

            ignite1.events().localListen(new IgnitePredicate<Event>() {
                @Override public boolean apply(Event evt) {
                    if (evt.type() == EventType.EVT_NODE_FAILED &&
                        failedNodeId.equals(((DiscoveryEvent)evt).eventNode().id()))
                        disconnected.set(true);

                    latch.countDown();

                    return false;
                }
            }, EventType.EVT_NODE_FAILED);

            spi0.stop = true;

            latch.await(15, TimeUnit.SECONDS);

            assertTrue(disconnected.get());

            try {
                ignite0.cluster().localNode().id();
            }
            catch (IllegalStateException e) {
                if (e.getMessage().contains("Grid is in invalid state to perform this operation"))
                    return;
            }

            fail();
        }
        finally {
            stopAllGrids();
        }
    }
    /**
     * @throws Exception If failed
     */
    public void testNodeShutdownOnRingMessageWorkerStartNotFinished() throws Exception {
        try {
            Ignite ignite0 = startGrid(0);

            TestMessageWorkerFailureSpi2 spi0 = new TestMessageWorkerFailureSpi2();

            nodeSpi.set(spi0);

            try {
                startGrid(1);

                fail();
            }
            catch (Exception e) {
                log.error("Expected error: " + e, e);
            }

            Ignite ignite1 = startGrid(1);

            assertEquals(2, ignite1.cluster().nodes().size());
            assertEquals(4, ignite1.cluster().topologyVersion());

            assertEquals(2, ignite0.cluster().nodes().size());
            assertEquals(4, ignite0.cluster().topologyVersion());
        }
        finally {
            stopAllGrids();
        }
    }


    /**
     * @param twoNodes If {@code true} starts two nodes, otherwise three.
     * @throws Exception If failed
     */
    private void customEventCoordinatorFailure(boolean twoNodes) throws Exception {
        TestCustomEventCoordinatorFailureSpi spi0 = new TestCustomEventCoordinatorFailureSpi();

        nodeSpi.set(spi0);

        Ignite ignite0 = startGrid(0);

        nodeSpi.set(new TestCustomEventCoordinatorFailureSpi());

        Ignite ignite1 = startGrid(1);

        nodeSpi.set(new TestCustomEventCoordinatorFailureSpi());

        Ignite ignite2 = twoNodes ? null : startGrid(2);

        final Ignite createCacheNode = ignite2 != null ? ignite2 : ignite1;

        CountDownLatch latch = new CountDownLatch(1);

        spi0.latch = latch;

        final String CACHE_NAME = "test-cache";

        IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new Callable<Void>() {
            @Override public Void call() throws Exception {
                log.info("Create test cache");

                CacheConfiguration ccfg = new CacheConfiguration();

                ccfg.setName(CACHE_NAME);

                createCacheNode.createCache(ccfg);

                return null;
            }
        }, "create-cache-thread");

        ((TcpCommunicationSpi)ignite0.configuration().getCommunicationSpi()).simulateNodeFailure();

        latch.await();

        ignite0.close();

        fut.get();

        IgniteCache<Object, Object> cache = grid(1).cache(CACHE_NAME);

        assertNotNull(cache);

        cache.put(1, 1);

        assertEquals(1, cache.get(1));

        log.info("Try start one more node.");

        nodeSpi.set(new TestCustomEventCoordinatorFailureSpi());

        Ignite ignite = startGrid(twoNodes ? 2 : 3);

        cache = ignite.cache(CACHE_NAME);

        assertNotNull(cache);

        cache.put(2, 2);

        assertEquals(1, cache.get(1));
        assertEquals(2, cache.get(2));
    }

    /**
     * Coordinator is added in failed list during node start.
     *
     * @throws Exception If failed.
     */
    public void testFailedNodes1() throws Exception {
        try {
            final int FAIL_ORDER = 3;

            nodeSpi.set(new TestFailedNodesSpi(FAIL_ORDER));

            final Ignite ignite0 = startGrid(0);

            nodeSpi.set(new TestFailedNodesSpi(FAIL_ORDER));

            startGrid(1);

            nodeSpi.set(new TestFailedNodesSpi(FAIL_ORDER));

            Ignite ignite2 = startGrid(2);

            assertEquals(2, ignite2.cluster().nodes().size());

            waitNodeStop(ignite0.name());

            tryCreateCache(2);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * Coordinator is added in failed list, concurrent nodes start.
     *
     * @throws Exception If failed.
     */
    public void testFailedNodes2() throws Exception {
        try {
            final int FAIL_ORDER = 3;

            nodeSpi.set(new TestFailedNodesSpi(FAIL_ORDER));

            Ignite ignite0 = startGrid(0);

            nodeSpi.set(new TestFailedNodesSpi(FAIL_ORDER));

            startGrid(1);

            final AtomicInteger nodeIdx = new AtomicInteger(1);

            GridTestUtils.runMultiThreaded(new Callable<Void>() {
                @Override public Void call() throws Exception {
                    int idx = nodeIdx.incrementAndGet();

                    nodeSpi.set(new TestFailedNodesSpi(FAIL_ORDER));

                    startGrid(idx);

                    return null;
                }
            }, 3, "start-node");

            Ignite ignite2 = ignite(2);

            waitForRemoteNodes(ignite2, 3);

            waitNodeStop(ignite0.name());

            tryCreateCache(4);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * Coordinator is added in failed list during node start, test with two nodes.
     *
     * @throws Exception If failed.
     */
    public void testFailedNodes3() throws Exception {
        try {
            nodeSpi.set(new TestFailedNodesSpi(-1));

            Ignite ignite0 = startGrid(0);

            nodeSpi.set(new TestFailedNodesSpi(2));

            Ignite ignite1 = startGrid(1);

            assertEquals(1, ignite1.cluster().nodes().size());

            waitNodeStop(ignite0.name());

            ignite1.getOrCreateCache(new CacheConfiguration<>()).put(1, 1);

            startGrid(2);

            assertEquals(2, ignite1.cluster().nodes().size());

            tryCreateCache(2);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * Coordinator is added in failed list during node start, but node detected failure dies before
     * sending {@link TcpDiscoveryNodeFailedMessage}.
     *
     * @throws Exception If failed.
     */
    public void testFailedNodes4() throws Exception {
        try {
            final int FAIL_ORDER = 3;

            nodeSpi.set(new TestFailedNodesSpi(FAIL_ORDER));

            final Ignite ignite0 = startGrid(0);

            nodeSpi.set(new TestFailedNodesSpi(FAIL_ORDER));

            Ignite ignite1 = startGrid(1);

            TestFailedNodesSpi spi = new TestFailedNodesSpi(FAIL_ORDER);

            spi.stopBeforeSndFail = true;

            nodeSpi.set(spi);

            Ignite ignite2 = startGrid(2);

            waitNodeStop(ignite2.name());

            log.info("Try start new node.");

            Ignite ignite3 = startGrid(3);

            waitNodeStop(ignite0.name());

            assertEquals(2, ignite1.cluster().nodes().size());
            assertEquals(2, ignite3.cluster().nodes().size());

            tryCreateCache(2);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * Adds some node in failed list after join process finished.
     *
     * @throws Exception If failed.
     */
    public void testFailedNodes5() throws Exception {
        try {
            ThreadLocalRandom rnd = ThreadLocalRandom.current();

            for (int iter = 0; iter < 3; iter++) {
                final int NODES = iter == 0 ? 2 : rnd.nextInt(3, 6);

                for (int i = 0; i < NODES; i++) {
                    nodeSpi.set(new TestFailedNodesSpi(-1));

                    startGrid(i);
                }

                Map<Long, Ignite> nodes = new HashMap<>();

                for (int i = 0; i < NODES; i++) {
                    Ignite ignite = ignite(i);

                    nodes.put(ignite.cluster().localNode().order(), ignite);
                }

                Ignite ignite = ignite(rnd.nextInt(NODES));

                log.info("Iteration [iter=" + iter + ", nodes=" + NODES + ", failFrom=" + ignite.name() + ']');

                TestFailedNodesSpi spi = (TestFailedNodesSpi)ignite.configuration().getDiscoverySpi();

                spi.failSingleMsg = true;

                long order = ignite.cluster().localNode().order();

                long nextOrder = order == NODES ? 1 : order + 1;

                Ignite failingNode = nodes.get(nextOrder);

                assertNotNull(failingNode);

                waitNodeStop(failingNode.name());

                Ignite newNode = startGrid(NODES);

                assertEquals(NODES, newNode.cluster().nodes().size());

                tryCreateCache(NODES);

                stopAllGrids();
            }
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If failed.
     */
    public void testCustomEventAckNotSend() throws Exception {
        try {
            TestCustomerEventAckSpi spi0 = new TestCustomerEventAckSpi();

            nodeSpi.set(spi0);

            Ignite ignite0 = startGrid(0);

            nodeSpi.set(new TestCustomerEventAckSpi());

            Ignite ignite1 = startGrid(1);

            spi0.stopBeforeSndAck = true;

            ignite1.message().remoteListen("test", new DummyPredicate());

            waitNodeStop(ignite0.name());

            startGrid(2);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @throws Exception If failed.
     */
    public void testDiscoveryEventsDiscard() throws Exception {
        try {
            TestEventDiscardSpi spi = new TestEventDiscardSpi();

            nodeSpi.set(spi);

            Ignite ignite0 = startGrid(0);

            startGrid(1);

            ignite0.createCache(new CacheConfiguration<>()); // Send custom message.

            ignite0.destroyCache(null); // Send custom message.

            stopGrid(1);

            log.info("Start new node.");

            spi.checkDuplicates = true;

            startGrid(1);

            spi.checkDuplicates = false;

            assertFalse(spi.failed);
        }
        finally {
            stopAllGrids();
        }
    }

    /**
     * @param nodeName Node name.
     * @throws Exception If failed.
     */
    private void waitNodeStop(final String nodeName) throws Exception {
        boolean wait = GridTestUtils.waitForCondition(new GridAbsPredicate() {
            @Override public boolean apply() {
                try {
                    Ignition.ignite(nodeName);

                    return false;
                }
                catch (IgniteIllegalStateException e) {
                    return true;
                }
            }
        }, 10_000);

        if (!wait)
            U.dumpThreads(log);

        assertTrue("Failed to wait for node stop.", wait);
    }

    /**
     * @param expNodes Expected nodes number.
     */
    private void tryCreateCache(int expNodes) {
        List<Ignite> allNodes = G.allGrids();

        assertEquals(expNodes, allNodes.size());

        int cntr = 0;

        for (Ignite ignite : allNodes) {
            CacheConfiguration<Object, Object> ccfg = new CacheConfiguration<>();

            ccfg.setName("cache-" + cntr++);

            log.info("Try create cache [node=" + ignite.name() + ", cache=" + ccfg.getName() + ']');

            ignite.getOrCreateCache(ccfg).put(1, 1);
        }
    }

    /**
     *
     */
    static class DummyPredicate implements IgniteBiPredicate<UUID, Object> {
        /** {@inheritDoc} */
        @Override public boolean apply(UUID uuid, Object o) {
            return true;
        }
    }


    /**
     *
     */
    private static class TestEventDiscardSpi extends TcpDiscoverySpi {
        /** */
        private ConcurrentHashSet<IgniteUuid> msgIds = new ConcurrentHashSet<>();

        /** */
        private volatile boolean checkDuplicates;

        /** */
        private volatile boolean failed;

        /** {@inheritDoc} */
        @Override protected void writeToSocket(Socket sock,
            TcpDiscoveryAbstractMessage msg,
            long timeout) throws IOException, IgniteCheckedException {
            boolean add = msgIds.add(msg.id());

            if (checkDuplicates && !add) {
                log.error("Send duplicated message: " + msg);

                failed = true;
            }

            super.writeToSocket(sock, msg, timeout);
        }
    }

    /**
     *
     */
    private static class TestCustomerEventAckSpi extends TcpDiscoverySpi {
        /** */
        private volatile boolean stopBeforeSndAck;

        /** {@inheritDoc} */
        @Override protected void writeToSocket(Socket sock,
            TcpDiscoveryAbstractMessage msg,
            long timeout) throws IOException, IgniteCheckedException {
            if (stopBeforeSndAck) {
                if (msg instanceof TcpDiscoveryCustomEventMessage) {
                    try {
                        DiscoveryCustomMessage custMsg = GridTestUtils.getFieldValue(
                            ((TcpDiscoveryCustomEventMessage)msg).message(marsh), "delegate");

                        if (custMsg instanceof StartRoutineAckDiscoveryMessage) {
                            log.info("Skip message send and stop node: " + msg);

                            sock.close();

                            GridTestUtils.runAsync(new Callable<Object>() {
                                @Override public Object call() throws Exception {
                                    ignite.close();

                                    return null;
                                }
                            }, "stop-node");

                            return;
                        }
                    }
                    catch (Throwable e) {
                        fail("Unexpected error: " + e);
                    }
                }
            }

            super.writeToSocket(sock, msg, timeout);
        }
    }

    /**
     * Simulate scenario when node detects node failure trying to send message, but node still alive.
     */
    private static class TestFailedNodesSpi extends TcpDiscoverySpi {
        /** */
        private AtomicBoolean failMsg = new AtomicBoolean();

        /** */
        private int failOrder;

        /** */
        private boolean stopBeforeSndFail;

        /** */
        private boolean stop;

        /** */
        private volatile boolean failSingleMsg;

        /**
         * @param failOrder Spi fails connection if local node order equals to this order.
         */
        TestFailedNodesSpi(int failOrder) {
            this.failOrder = failOrder;
        }

        /** {@inheritDoc} */
        @Override protected void writeToSocket(Socket sock,
            TcpDiscoveryAbstractMessage msg,
            long timeout) throws IOException, IgniteCheckedException {
            if (stop)
                return;

            if (failSingleMsg) {
                failSingleMsg = false;

                log.info("IO error on message send [locNode=" + locNode + ", msg=" + msg + ']');

                sock.close();

                throw new SocketTimeoutException();
            }

            if (locNode.internalOrder() == failOrder &&
                (msg instanceof TcpDiscoveryNodeAddedMessage) &&
                failMsg.compareAndSet(false, true)) {
                log.info("IO error on message send [locNode=" + locNode + ", msg=" + msg + ']');

                sock.close();

                throw new SocketTimeoutException();
            }

            if (stopBeforeSndFail &&
                locNode.internalOrder() == failOrder &&
                (msg instanceof TcpDiscoveryNodeFailedMessage)) {
                stop = true;

                log.info("Skip messages send and stop node [locNode=" + locNode + ", msg=" + msg + ']');

                sock.close();

                GridTestUtils.runAsync(new Callable<Object>() {
                    @Override public Object call() throws Exception {
                        ignite.close();

                        return null;
                    }
                }, "stop-node");

                return;
            }

            super.writeToSocket(sock, msg, timeout);
        }
    }

    /**
     *
     */
    private static class TestCustomEventCoordinatorFailureSpi extends TcpDiscoverySpi {
        /** */
        private volatile CountDownLatch latch;

        /** */
        private boolean stop;

        /** {@inheritDoc} */
        @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
            long timeout) throws IOException, IgniteCheckedException {
            if (msg instanceof TcpDiscoveryCustomEventMessage && latch != null) {
                log.info("Stop node on custom event: " + msg);

                latch.countDown();

                stop = true;
            }

            if (stop)
                return;

            super.writeToSocket(sock, msg, timeout);
        }
    }

    /**
     *
     */
    private static class TestCustomEventRaceSpi extends TcpDiscoverySpi {
        /** */
        private volatile CountDownLatch nodeAdded1;

        /** */
        private volatile CountDownLatch nodeAdded2;

        /** */
        private volatile boolean stop;

        /** */
        private boolean debug;

        /** {@inheritDoc} */
        @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
            long timeout) throws IOException, IgniteCheckedException {
            if (msg instanceof TcpDiscoveryNodeAddedMessage) {
                if (nodeAdded1 != null) {
                    nodeAdded1.countDown();

                    if (debug)
                        log.info("--- Wait node added: " + msg);

                    U.await(nodeAdded2);

                    nodeAdded1 = null;
                    nodeAdded2 = null;
                }

                if (stop)
                    return;

                if (debug)
                    log.info("--- Send node added: " + msg);
            }

            if (debug && msg instanceof TcpDiscoveryNodeAddFinishedMessage)
                log.info("--- Send node finished: " + msg);

            if (debug && msg instanceof TcpDiscoveryCustomEventMessage)
                log.info("--- Send custom event: " + msg);

            super.writeToSocket(sock, msg, timeout);
        }
    }

    /**
     *
     */
    private static class TestMessageWorkerFailureSpi1 extends TcpDiscoverySpi {
        /** */
        private volatile boolean stop;

        /** {@inheritDoc} */
        @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
            long timeout) throws IOException, IgniteCheckedException {

            if (stop)
                throw new RuntimeException("Failing ring message worker explicitly");

            super.writeToSocket(sock, msg, timeout);
        }
    }

    /**
     *
     */
    private static class TestMessageWorkerFailureSpi2 extends TcpDiscoverySpi {
        /** */
        private volatile boolean stop;

        /** {@inheritDoc} */
        @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg,
            long timeout) throws IOException, IgniteCheckedException {
            if (stop)
                throw new RuntimeException("Failing ring message worker explicitly");

            super.writeToSocket(sock, msg, timeout);

            if (msg instanceof TcpDiscoveryNodeAddedMessage)
                stop = true;
        }
    }

    /**
     * Starts new grid with given index. Method optimize is not invoked.
     *
     * @param idx Index of the grid to start.
     * @return Started grid.
     * @throws Exception If anything failed.
     */
    private Ignite startGridNoOptimize(int idx) throws Exception {
        return startGridNoOptimize(getTestGridName(idx));
    }

    /**
     * Starts new grid with given name. Method optimize is not invoked.
     *
     * @param gridName Grid name.
     * @return Started grid.
     * @throws Exception If failed.
     */
    private Ignite startGridNoOptimize(String gridName) throws Exception {
        return G.start(getConfiguration(gridName));
    }
}