public IgniteCacheSourceTask(Map<String, String> props, Ignite igniteNode) {
	Objects.requireNonNull(igniteNode);
	sourceNode = igniteNode;
	cacheName = props.get(IgniteSourceConstants.CACHE_NAME);

	if (props.containsKey(IgniteSourceConstants.INTL_BUF_SIZE)) {
		evtBufSize = Integer.parseInt(props.get(IgniteSourceConstants.INTL_BUF_SIZE));
		evtBuf = new LinkedBlockingQueue<>(evtBufSize);
	}

	if (props.containsKey(IgniteSourceConstants.INTL_BATCH_SIZE))
		evtBatchSize = Integer.parseInt(props.get(IgniteSourceConstants.INTL_BATCH_SIZE));

	TaskRemoteFilter rmtLsnr = new TaskRemoteFilter(cacheName);
	filter = startSourceCacheListeners(props);
	try {
		rmtLsnrId = sourceNode.events(sourceNode.cluster().forCacheNodes(cacheName))
				.remoteListen(locLsnr, rmtLsnr, EventType.EVT_CACHE_OBJECT_PUT);
	} catch (Exception e) {
		log.error("Failed to register event listener!", e);
		throw new IllegalStateException(e);
	} finally {
		stopped.set(false);
	}

}
 

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

    cfg.setDeploymentMode(depMode);
    cfg.setNetworkTimeout(10000);

    if (igniteInstanceName.contains("testCacheRedeployVersionChangeContinuousMode")) {
        CacheConfiguration cacheCfg = new CacheConfiguration(DEFAULT_CACHE_NAME);

        cacheCfg.setCacheMode(CacheMode.REPLICATED);

        cfg.setCacheConfiguration(cacheCfg);
    }
    else
        cfg.setCacheConfiguration();

    cfg.setIncludeEventTypes(EventType.EVTS_ALL);

    return cfg;
}
 

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

    List<CacheConfiguration> ccfgs = new ArrayList<>();

    ccfgs.add(buildCacheConfiguration(CACHE_ORG));
    ccfgs.add(buildCacheConfiguration(CACHE_PERSON));
    ccfgs.add(buildCacheConfiguration(CACHE_POSITION));

    c.setCacheConfiguration(ccfgs.toArray(new CacheConfiguration[ccfgs.size()]));
    c.setSqlConfiguration(new SqlConfiguration().setLongQueryWarningTimeout(10000));
    c.setIncludeEventTypes(EventType.EVTS_ALL);

    return c;
}
 

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

    cfg.setMarshaller(new BinaryMarshaller());
    cfg.setIncludeEventTypes(EventType.EVTS_CACHE);

    if (getTestIgniteInstanceName(CLIENT_ID).equals(igniteInstanceName)) {
        CacheConfiguration ccfg = getCacheConfiguration();

        cfg.setCacheConfiguration(ccfg);
    }

    if (joinTimeout != 0 && getTestIgniteInstanceName(1).equals(igniteInstanceName))
        ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setJoinTimeout(joinTimeout);

    return cfg;
}
 

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

    c.setConsistentId(igniteInstanceName);

    CacheConfiguration<?, ?> cc = defaultCacheConfiguration();

    cc.setCacheMode(PARTITIONED);
    cc.setBackups(1);
    cc.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
    cc.setAtomicityMode(TRANSACTIONAL);
    cc.setRebalanceMode(SYNC);
    cc.setAffinity(new RendezvousAffinityFunction(false, 15));
    cc.setIndexedTypes(
        Integer.class, Integer.class
    );

    c.setCacheConfiguration(cc);
    c.setIncludeEventTypes(EventType.EVTS_ALL);

    return c;
}
 

/**
 * @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(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());
        }
    );
}
 

/**
 * Generate events for created file or directory.
 *
 * @param createdPaths Created paths.
 * @param file Whether file was created.
 */
private void generateCreateEvents(List<IgfsPath> createdPaths, boolean file) {
    if (evts.isRecordable(EventType.EVT_IGFS_DIR_CREATED)) {
        for (int i = 0; i < createdPaths.size() - 1; i++)
            IgfsUtils.sendEvents(igfsCtx.kernalContext(), createdPaths.get(i),
                EventType.EVT_IGFS_DIR_CREATED);
    }

    IgfsPath leafPath = createdPaths.get(createdPaths.size() - 1);

    if (file) {
        IgfsUtils.sendEvents(igfsCtx.kernalContext(), leafPath, EventType.EVT_IGFS_FILE_CREATED);
        IgfsUtils.sendEvents(igfsCtx.kernalContext(), leafPath, EventType.EVT_IGFS_FILE_OPENED_WRITE);
    }
    else
        IgfsUtils.sendEvents(igfsCtx.kernalContext(), leafPath, EventType.EVT_IGFS_DIR_CREATED);
}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testReconnect() throws Exception {
    Ignite srv = startGrid("server");

    IgniteEvents evts = srv.events();

    evts.enableLocal(EventType.EVTS_DISCOVERY_ALL);
    evts.localListen(new IgnitePredicate<Event>() {
        @Override public boolean apply(Event evt) {
            ClusterNode node = ((DiscoveryEvent)evt).eventNode();

            rejoinAttr = node.attribute("test");

            return true;
        }
    }, EventType.EVT_NODE_JOINED);

    Ignite client = startClientGrid("client");

    reconnectClientNode(log, client, srv, null);

    assertEquals("2", rejoinAttr);
}
 

/**
 * @param ignite Client node.
 * @return Client reconnect state.
 */
private AtomicBoolean getClientReconnectState(Ignite ignite) {
    final AtomicBoolean reconnectState = new AtomicBoolean(false);

    ignite.events().localListen(
        new IgnitePredicate<Event>() {
            @Override public boolean apply(Event evt) {
                if (evt.type() == EventType.EVT_CLIENT_NODE_RECONNECTED)
                    reconnectState.set(true);

                return true;
            }
        },
        EventType.EVT_CLIENT_NODE_RECONNECTED
    );

    return reconnectState;
}
 

/** {@inheritDoc} */
@Override public IgniteFuture<?> onDiscovery(
    int type,
    long topVer,
    ClusterNode node,
    Collection<ClusterNode> topSnapshot,
    @Nullable Map<Long, Collection<ClusterNode>> topHist,
    @Nullable DiscoverySpiCustomMessage spiCustomMsg
) {
    if (EventType.EVT_NODE_METRICS_UPDATED == type) {
        log.info("Metrics update message catched from node " + node);

        assertFalse(locNode.isClient());

        if (node.isClient())
            clientMetricsUpdateCnt++;
    }

    if (delegate != null)
        return delegate.onDiscovery(type, topVer, node, topSnapshot, topHist, spiCustomMsg);

    return new IgniteFinishedFutureImpl<>();
}
 

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

    int idx = getTestIgniteInstanceIndex(igniteInstanceName);

    if (idx == 0) {
        Map<IgnitePredicate<? extends Event>, int[]> lsnrs = new HashMap<>();

        lsnrs.put(new IgnitePredicate<Event>() {
            @Override public boolean apply(Event evt) {
                fired.countDown();

                return true;
            }
        }, new int[] { EventType.EVT_NODE_JOINED } );

        cfg.setLocalEventListeners(lsnrs);
    }

    return cfg;
}
 

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

    // Override P2P configuration to exclude Task and Job classes
    cfg.setPeerClassLoadingLocalClassPathExclude(GridDeploymentTestJob.class.getName(),
        GridDeploymentTestTask.class.getName());

    // Following tests makes sense in ISOLATED modes (they redeploy tasks
    // and don't change task version. The different tasks with the same version from the same node
    // executed in parallel - this does not work in share mode.)
    cfg.setDeploymentMode(DeploymentMode.ISOLATED);

    cfg.setPeerClassLoadingLocalClassPathExclude(
        "org.apache.ignite.internal.GridMultipleVersionsDeploymentSelfTest*");

    cfg.setIncludeEventTypes(EventType.EVTS_ALL);

    return cfg;
}
 

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

    cfg.setCacheConfiguration(
        cacheConfig(firstCacheMode, 1, FIRST_CACHE_NAME),
        cacheConfig(secondCacheMode, 2, SECOND_CACHE_NAME));

    Map<IgnitePredicate<? extends Event>, int[]> listeners = new HashMap<>();

    listeners.put(new IgnitePredicate<CacheRebalancingEvent>() {
        @Override public boolean apply(CacheRebalancingEvent evt) {
            times.get(gridIdx(evt)).putIfAbsent(evt.cacheName(), evt.timestamp());
            return true;
        }
    }, new int[]{EventType.EVT_CACHE_REBALANCE_STOPPED});

    cfg.setLocalEventListeners(listeners);

    cfg.setIncludeEventTypes(EventType.EVTS_ALL);

    return cfg;
}
 

/**
 * @param disconnectedLatch Disconnect latch. Will be fired when client disconnect event is received.
 * @param reconnectedLatch Reconnect latch. Will be fired when cilent reconnect event is receoved.
 */
protected void addDisconnectListener(
    final CountDownLatch disconnectedLatch,
    final CountDownLatch reconnectedLatch
) {
    grid(nodeClient).events().localListen(new IgnitePredicate<Event>() {
        @Override public boolean apply(Event event) {
            switch (event.type()) {
                case EventType.EVT_CLIENT_NODE_DISCONNECTED:
                    info("Client disconnected");

                    disconnectedLatch.countDown();

                    break;
                case EventType.EVT_CLIENT_NODE_RECONNECTED:
                    info("Client reconnected");

                    reconnectedLatch.countDown();
            }

            return true;
        }
    }, EventType.EVT_CLIENT_NODE_DISCONNECTED, EventType.EVT_CLIENT_NODE_RECONNECTED);
}
 

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

    cfg.setActiveOnStart(false);
    cfg.setIncludeEventTypes(EventType.EVT_CACHE_REBALANCE_PART_DATA_LOST);

    cfg.setCacheConfiguration(
        new CacheConfiguration<>(DEFAULT_CACHE_NAME)
            .setCacheMode(CacheMode.PARTITIONED)
            .setBackups(0)
            .setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
            .setAffinity(new RendezvousAffinityFunction(false, 50))
            .setPartitionLossPolicy(PartitionLossPolicy.READ_WRITE_SAFE)
    );

    return cfg;
}
 

public void addListener(MessageListener listener) {
    IgniteTableListener messageListener = new IgniteTableListener(listener);
    IgniteTopicListener topicListener = new IgniteTopicListener(listener);
    if(!ignite.cluster().localNode().isClient())ignite.events(ignite.cluster().forLocal()).remoteListen(messageListener,
            (IgnitePredicate<CacheEvent>) event -> event.cacheName().equals(IgniteMessageDao.CACHE_NAME),
            EventType.EVT_CACHE_OBJECT_PUT,
            EventType.EVT_CACHE_OBJECT_REMOVED);
    else ignite.message(ignite.cluster().forLocal()).remoteListen(EVENT_TOPIC,topicListener);
}
 

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

    cfg.setIncludeEventTypes(EventType.EVTS_ALL);

    if (indexingDisabled)
        GridQueryProcessor.idxCls = DummyQueryIndexing.class;

    return cfg;
}
 

/**
 * @param topVer Topology version.
 * @return Expected event instance.
 */
static DiscoveryEvent joinEvent(long topVer) {
    DiscoveryEvent expEvt = new DiscoveryEvent(null, null, EventType.EVT_NODE_JOINED, null);

    expEvt.topologySnapshot(topVer, null);

    return expEvt;
}
 

/**
 * @param topVer Topology version.
 * @return Expected event instance.
 */
static DiscoveryEvent failEvent(long topVer) {
    DiscoveryEvent expEvt = new DiscoveryEvent(null, null, EventType.EVT_NODE_FAILED, null);

    expEvt.topologySnapshot(topVer, null);

    return expEvt;
}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testSegmentation2() throws Exception {
    sesTimeout = 2000;

    Ignite node0 = startGrid(0);

    final CountDownLatch l = new CountDownLatch(1);

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

            return false;
        }
    }, EventType.EVT_NODE_SEGMENTED);

    try {
        zkCluster.close();

        assertTrue(l.await(10, TimeUnit.SECONDS));
    }
    finally {
        zkCluster = ZookeeperDiscoverySpiTestUtil.createTestingCluster(ZK_SRVS);

        zkCluster.start();
    }
}
 

/**
 * @throws Exception If failed.
 */
@Test
public void testSegmentation3() throws Exception {
    sesTimeout = 5000;

    Ignite node0 = startGrid(0);

    final CountDownLatch l = new CountDownLatch(1);

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

            return false;
        }
    }, EventType.EVT_NODE_SEGMENTED);

    List<TestingZooKeeperServer> srvs = zkCluster.getServers();

    assertEquals(3, srvs.size());

    try {
        srvs.get(0).stop();
        srvs.get(1).stop();

        QuorumPeer qp = srvs.get(2).getQuorumPeer();

        // Zookeeper's socket timeout [tickTime * initLimit] + 5 additional seconds for other logic
        assertTrue(l.await(qp.getTickTime() * qp.getInitLimit() + 5000, TimeUnit.MILLISECONDS));
    }
    finally {
        zkCluster.close();

        zkCluster = ZookeeperDiscoverySpiTestUtil.createTestingCluster(ZK_SRVS);

        zkCluster.start();
    }
}
 

/**
 * @param ignite Node.
 * @param joinEvtCnt Expected events number.
 * @return Events latch.
 */
private CountDownLatch expectJoinEvents(Ignite ignite, int joinEvtCnt) {
    final CountDownLatch latch = new CountDownLatch(joinEvtCnt);

    ignite.events().remoteListen(new IgniteBiPredicate<UUID, Event>() {
        @Override public boolean apply(UUID uuid, Event evt) {
            latch.countDown();
            return true;
        }
    }, null, EventType.EVT_NODE_JOINED);

    return latch;
}
 

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

    cfg.setCacheConfiguration(new CacheConfiguration()
        .setName(DEFAULT_CACHE_NAME)
        .setCacheMode(CacheMode.PARTITIONED)
        .setBackups(0)
    );

    cfg.setIncludeEventTypes(EventType.EVTS_ALL);

    return cfg;
}
 

/**
 * Test OOME in event listener.
 */
@Test
public void testEventListenerOomError() throws Exception {
    IgniteEx ignite0 = startGrid(0);
    IgniteEx ignite1 = startGrid(1);

    IgniteCache<Integer, Integer> cache0 = ignite0.getOrCreateCache(DEFAULT_CACHE_NAME);
    IgniteCache<Integer, Integer> cache1 = ignite1.getOrCreateCache(DEFAULT_CACHE_NAME);

    awaitPartitionMapExchange();

    ignite1.events().localListen(new IgnitePredicate<Event>() {
        @Override public boolean apply(Event evt) {
            throw new OutOfMemoryError();
        }
    }, EventType.EVT_CACHE_OBJECT_PUT);

    Integer key = primaryKey(cache1);

    try {
        cache0.put(key, key);
    }
    catch (Throwable ignore) {
        // Expected.
    }

    assertFailureState(ignite0, ignite1);
}
 

/**
 * Tests node is stopped after triggering StopNodeFailureHandler.
 *
 * @throws Exception If failed.
 */
@Test
public void testNodeStopped() throws Exception {
    try {
        IgniteEx ignite0 = startGrid(0);
        IgniteEx ignite1 = startGrid(1);

        final CountDownLatch latch = new CountDownLatch(1);

        ignite0.events().localListen(new PE() {
            @Override public boolean apply(Event evt) {
                latch.countDown();

                return true;
            }
        }, EventType.EVT_NODE_LEFT);

        ignite1.context().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, null));

        assertTrue(latch.await(2000, TimeUnit.MILLISECONDS));

        Thread.sleep(1000);

        assertEquals(IgnitionEx.state(ignite0.name()), IgniteState.STARTED);
        assertEquals(IgnitionEx.state(ignite1.name()), IgniteState.STOPPED_ON_FAILURE);
    }
    finally {
        stopAllGrids();
    }
}
 

/**
 * Tests failed node's JVM is halted after triggering StopNodeOrHaltFailureHandler.
 */
@Test
public void testJvmHalted() throws Exception {
    IgniteEx g = grid(0);
    IgniteEx rmt1 = grid(1);
    IgniteEx rmt2 = grid(2);

    assertTrue(isMultiJvmObject(rmt1));
    assertTrue(isMultiJvmObject(rmt2));

    assertTrue(g.cluster().nodes().size() == NODES_CNT);

    final CountDownLatch latch = new CountDownLatch(1);

    g.events().localListen(new PE() {
        @Override public boolean apply(Event evt) {
            latch.countDown();

            return true;
        }
    }, EventType.EVT_NODE_LEFT, EventType.EVT_NODE_FAILED);

    g.compute().broadcast(new CA() {
        @IgniteInstanceResource
        private Ignite ignite;

        @Override public void apply() {
            ((IgniteEx)ignite).context().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, null));
        }
    });

    assertTrue(latch.await(2000, TimeUnit.MILLISECONDS));

    Thread.sleep(1000);

    assertTrue(((IgniteProcessProxy)rmt1).getProcess().getProcess().isAlive());
    assertFalse(((IgniteProcessProxy)rmt2).getProcess().getProcess().isAlive());
}
 

/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
    return super.getConfiguration(igniteInstanceName)
        .setCacheConfiguration(new CacheConfiguration(DEFAULT_CACHE_NAME))
        .setClusterStateOnStart(INACTIVE)
        .setIncludeEventTypes(EventType.EVTS_ALL);
}
 

/**
 * Starts temporary nodes.
 *
 * @throws Exception If failed.
 */
private void startTempNodes() throws Exception {
    for (int j = 0; j < TMP_NODES_CNT; j++) {
        Ignite newNode = startGrid(gridCntr++);

        info("New node started: " + newNode.name());

        alive.add(newNode);

        newNode.events().localListen(lsnr, EventType.EVT_NODE_LEFT, EventType.EVT_NODE_FAILED);
    }
}
 

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

    cfg.setCacheConfiguration();
    cfg.setIncludeProperties();
    cfg.setIncludeEventTypes(EventType.EVTS_ALL);

    return cfg;
}