ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [13/56] [abbrv] incubator-ignite git commit: # ignite-63
Date Fri, 23 Jan 2015 09:36:32 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridFailoverTaskWithPredicateSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridFailoverTaskWithPredicateSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridFailoverTaskWithPredicateSelfTest.java
new file mode 100644
index 0000000..d41c325
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridFailoverTaskWithPredicateSelfTest.java
@@ -0,0 +1,252 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.marshaller.optimized.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.failover.*;
+import org.apache.ignite.spi.failover.always.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import java.io.*;
+import java.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Test failover of a task with Node filter predicate.
+ */
+@GridCommonTest(group = "Kernal Self")
+public class GridFailoverTaskWithPredicateSelfTest extends GridCommonAbstractTest {
+    /** First node's name. */
+    private static final String NODE1 = "NODE1";
+
+    /** Second node's name. */
+    private static final String NODE2 = "NODE2";
+
+    /** Third node's name. */
+    private static final String NODE3 = "NODE3";
+
+    /** Predicate to exclude the second node from topology */
+    private final IgnitePredicate<ClusterNode> p = new IgnitePredicate<ClusterNode>() {
+        @Override
+        public boolean apply(ClusterNode e) {
+            return !NODE2.equals(e.attribute(GridNodeAttributes.ATTR_GRID_NAME));
+        }
+    };
+
+    /** Whether delegating fail over node was found or not. */
+    private final AtomicBoolean routed = new AtomicBoolean();
+
+    /** Whether job execution failed with exception. */
+    private final AtomicBoolean failed = new AtomicBoolean();
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        cfg.setFailoverSpi(new AlwaysFailoverSpi() {
+            /** {@inheritDoc} */
+            @Override public ClusterNode failover(FailoverContext ctx, List<ClusterNode> grid) {
+                ClusterNode failoverNode = super.failover(ctx, grid);
+
+                if (failoverNode != null)
+                    routed.set(true);
+                else
+                    routed.set(false);
+
+                return failoverNode;
+            }
+        });
+
+        cfg.setMarshaller(new IgniteOptimizedMarshaller(false));
+
+        return cfg;
+    }
+
+    /**
+     * Tests that failover doesn't happen on two-node grid when the Task is applicable only for the first node
+     * and fails on it.
+     *
+     * @throws Exception If failed.
+     */
+    public void testJobNotFailedOver() throws Exception {
+        failed.set(false);
+        routed.set(false);
+
+        try {
+            Ignite ignite1 = startGrid(NODE1);
+            Ignite ignite2 = startGrid(NODE2);
+
+            assert ignite1 != null;
+            assert ignite2 != null;
+
+            compute(ignite1.cluster().forPredicate(p)).withTimeout(10000).execute(JobFailTask.class.getName(), "1");
+        }
+        catch (ClusterTopologyException ignored) {
+            failed.set(true);
+        }
+        finally {
+            assertTrue(failed.get());
+            assertFalse(routed.get());
+
+            stopGrid(NODE1);
+            stopGrid(NODE2);
+        }
+    }
+
+    /**
+     * Tests that failover happens on three-node grid when the Task is applicable for the first node
+     * and fails on it, but is also applicable on another node.
+     *
+     * @throws Exception If failed.
+     */
+    public void testJobFailedOver() throws Exception {
+        failed.set(false);
+        routed.set(false);
+
+        try {
+            Ignite ignite1 = startGrid(NODE1);
+            Ignite ignite2 = startGrid(NODE2);
+            Ignite ignite3 = startGrid(NODE3);
+
+            assert ignite1 != null;
+            assert ignite2 != null;
+            assert ignite3 != null;
+
+            Integer res = (Integer)compute(ignite1.cluster().forPredicate(p)).withTimeout(10000).
+                execute(JobFailTask.class.getName(), "1");
+
+            assert res == 1;
+        }
+        catch (ClusterTopologyException ignored) {
+            failed.set(true);
+        }
+        finally {
+            assertFalse(failed.get());
+            assertTrue(routed.get());
+
+            stopGrid(NODE1);
+            stopGrid(NODE2);
+            stopGrid(NODE3);
+        }
+    }
+
+    /**
+     * Tests that in case of failover our predicate is intersected with projection
+     * (logical AND is performed).
+     *
+     * @throws Exception If error happens.
+     */
+    public void testJobNotFailedOverWithStaticProjection() throws Exception {
+        failed.set(false);
+        routed.set(false);
+
+        try {
+            Ignite ignite1 = startGrid(NODE1);
+            Ignite ignite2 = startGrid(NODE2);
+            Ignite ignite3 = startGrid(NODE3);
+
+            assert ignite1 != null;
+            assert ignite2 != null;
+            assert ignite3 != null;
+
+            // Get projection only for first 2 nodes.
+            ClusterGroup nodes = ignite1.cluster().forNodeIds(Arrays.asList(
+                ignite1.cluster().localNode().id(),
+                ignite2.cluster().localNode().id()));
+
+            // On failover NODE3 shouldn't be taken into account.
+            Integer res = (Integer)compute(nodes.forPredicate(p)).withTimeout(10000).
+                execute(JobFailTask.class.getName(), "1");
+
+            assert res == 1;
+        }
+        catch (ClusterTopologyException ignored) {
+            failed.set(true);
+        }
+        finally {
+            assertTrue(failed.get());
+            assertFalse(routed.get());
+
+            stopGrid(NODE1);
+            stopGrid(NODE2);
+            stopGrid(NODE3);
+        }
+    }
+
+    /** */
+    @ComputeTaskSessionFullSupport
+    private static class JobFailTask implements ComputeTask<String, Object> {
+        /** */
+        @IgniteTaskSessionResource
+        private ComputeTaskSession ses;
+
+        /** {@inheritDoc} */
+        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, String arg) throws IgniteCheckedException {
+            ses.setAttribute("fail", true);
+
+            return Collections.singletonMap(new ComputeJobAdapter(arg) {
+                /** {@inheritDoc} */
+                @SuppressWarnings({"RedundantTypeArguments"})
+                @Override
+                public Serializable execute() throws IgniteCheckedException {
+                    boolean fail;
+
+                    try {
+                        fail = ses.<String, Boolean>waitForAttribute("fail", 0);
+                    }
+                    catch (InterruptedException e) {
+                        throw new IgniteCheckedException("Got interrupted while waiting for attribute to be set.", e);
+                    }
+
+                    if (fail) {
+                        ses.setAttribute("fail", false);
+
+                        throw new IgniteCheckedException("Job exception.");
+                    }
+
+                    // This job does not return any result.
+                    return Integer.parseInt(this.<String>argument(0));
+                }
+            }, subgrid.get(0));
+        }
+
+        /** {@inheritDoc} */
+        @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> received)
+                throws IgniteCheckedException {
+            if (res.getException() != null && !(res.getException() instanceof ComputeUserUndeclaredException))
+                return ComputeJobResultPolicy.FAILOVER;
+
+            return ComputeJobResultPolicy.REDUCE;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object reduce(List<ComputeJobResult> results) throws IgniteCheckedException {
+            assert results.size() == 1;
+
+            return results.get(0).getData();
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridFailoverTopologySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridFailoverTopologySelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridFailoverTopologySelfTest.java
new file mode 100644
index 0000000..3a99f4e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridFailoverTopologySelfTest.java
@@ -0,0 +1,160 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.failover.*;
+import org.apache.ignite.spi.failover.always.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import java.io.*;
+import java.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Test failover and topology. It don't pick local node if it has been excluded from topology.
+ */
+@GridCommonTest(group = "Kernal Self")
+public class GridFailoverTopologySelfTest extends GridCommonAbstractTest {
+    /** */
+    private final AtomicBoolean failed = new AtomicBoolean(false);
+
+    /** */
+    public GridFailoverTopologySelfTest() {
+        super(/*start Grid*/false);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        cfg.setNodeId(null);
+
+        cfg.setFailoverSpi(new AlwaysFailoverSpi() {
+            /** Ignite instance. */
+            @IgniteInstanceResource
+            private Ignite ignite;
+
+            /** {@inheritDoc} */
+            @Override public ClusterNode failover(FailoverContext ctx, List<ClusterNode> grid) {
+                if (grid.size() != 1) {
+                    failed.set(true);
+
+                    error("Unexpected grid size [expected=1, grid=" + grid + ']');
+                }
+
+                UUID locNodeId = ignite.configuration().getNodeId();
+
+                for (ClusterNode node : grid) {
+                    if (node.id().equals(locNodeId)) {
+                        failed.set(true);
+
+                        error("Grid shouldn't contain local node [localNodeId=" + locNodeId + ", grid=" + grid + ']');
+                    }
+                }
+
+                return super.failover(ctx, grid);
+            }
+        });
+
+        return cfg;
+    }
+
+    /**
+     * Tests that failover don't pick local node if it has been excluded from topology.
+     *
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings("unchecked")
+    public void testFailoverTopology() throws Exception {
+        try {
+            Ignite ignite1 = startGrid(1);
+
+            startGrid(2);
+
+            ignite1.compute().localDeployTask(JobFailTask.class, JobFailTask.class.getClassLoader());
+
+            try {
+                compute(ignite1.cluster().forRemotes()).execute(JobFailTask.class, null);
+            }
+            catch (IgniteCheckedException e) {
+                info("Got expected grid exception: " + e);
+            }
+
+            assert !failed.get();
+        }
+        finally {
+            stopGrid(1);
+            stopGrid(2);
+        }
+    }
+
+    /** */
+    private static class JobFailTask implements ComputeTask<String, Object> {
+        /** Ignite instance. */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
+        /** */
+        private boolean jobFailedOver;
+
+        /** {@inheritDoc} */
+        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, String arg) throws IgniteCheckedException {
+            assert ignite != null;
+
+            UUID locNodeId = ignite.configuration().getNodeId();
+
+            assert locNodeId != null;
+
+            ClusterNode remoteNode = null;
+
+            for (ClusterNode node : subgrid) {
+                if (!node.id().equals(locNodeId))
+                    remoteNode = node;
+            }
+
+            return Collections.singletonMap(new ComputeJobAdapter(arg) {
+                @Override public Serializable execute() throws IgniteCheckedException {
+                    throw new IgniteCheckedException("Job exception.");
+                }
+            }, remoteNode);
+        }
+
+        /** {@inheritDoc} */
+        @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> received) throws IgniteCheckedException {
+            if (res.getException() != null && !jobFailedOver) {
+                jobFailedOver = true;
+
+                return ComputeJobResultPolicy.FAILOVER;
+            }
+
+            return ComputeJobResultPolicy.REDUCE;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object reduce(List<ComputeJobResult> results) throws IgniteCheckedException {
+            assert results.size() == 1;
+
+            return results.get(0).getData();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridHomePathSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridHomePathSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridHomePathSelfTest.java
new file mode 100644
index 0000000..f3cf621
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridHomePathSelfTest.java
@@ -0,0 +1,75 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import static org.apache.ignite.IgniteSystemProperties.*;
+
+/**
+ *
+ */
+public class GridHomePathSelfTest extends GridCommonAbstractTest {
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        cfg.setLocalHost(getTestResources().getLocalHost());
+
+        return cfg;
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testHomeOverride() throws Exception {
+        try {
+            startGrid(0);
+
+            // Test home override.
+            IgniteConfiguration c = getConfiguration(getTestGridName(1));
+
+            c.setGridGainHome("/new/path");
+
+            try {
+                G.start(c);
+
+                assert false : "Exception should have been thrown.";
+            }
+            catch (Exception e) {
+                if (X.hasCause(e, IgniteException.class))
+                    info("Caught expected exception: " + e);
+                else
+                    throw e;
+            }
+
+            // Test no override.
+            IgniteConfiguration c1 = getConfiguration(getTestGridName(1));
+
+            c1.setGridGainHome(System.getProperty(GG_HOME));
+
+            G.start(c1);
+        }
+        finally {
+            stopAllGrids();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridJobCheckpointCleanupSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridJobCheckpointCleanupSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridJobCheckpointCleanupSelfTest.java
new file mode 100644
index 0000000..ea2d4bc
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridJobCheckpointCleanupSelfTest.java
@@ -0,0 +1,164 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.*;
+import org.apache.ignite.spi.checkpoint.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.jetbrains.annotations.*;
+
+import java.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * Test for checkpoint cleanup.
+ */
+public class GridJobCheckpointCleanupSelfTest extends GridCommonAbstractTest {
+    /** Number of currently alive checkpoints. */
+    private final AtomicInteger cntr = new AtomicInteger();
+
+    /** Checkpoint. */
+    private CheckpointSpi checkpointSpi;
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration c = super.getConfiguration(gridName);
+
+        c.setCheckpointSpi(checkpointSpi);
+
+        return c;
+    }
+
+    /**
+     * Spawns one job on the node other than task node and
+     * ensures that all checkpoints were removed after task completion.
+     *
+     * @throws Exception if failed.
+     */
+    public void testCheckpointCleanup() throws Exception {
+        try {
+            checkpointSpi = new TestCheckpointSpi("task-checkpoints", cntr);
+
+            Ignite taskIgnite = startGrid(0);
+
+            checkpointSpi = new TestCheckpointSpi("job-checkpoints", cntr);
+
+            Ignite jobIgnite = startGrid(1);
+
+            taskIgnite.compute().execute(new CheckpointCountingTestTask(), jobIgnite.cluster().localNode());
+        }
+        finally {
+            stopAllGrids();
+        }
+
+        assertEquals(cntr.get(), 0);
+    }
+
+    /**
+     * Test checkpoint SPI.
+     */
+    @IgniteSpiMultipleInstancesSupport(true)
+    private static class TestCheckpointSpi extends IgniteSpiAdapter implements CheckpointSpi {
+        /** Counter. */
+        private final AtomicInteger cntr;
+
+        /**
+         * @param name Name.
+         * @param cntr Counter.
+         */
+        TestCheckpointSpi(String name, AtomicInteger cntr) {
+            setName(name);
+
+            this.cntr = cntr;
+        }
+
+        /** {@inheritDoc} */
+        @Override public byte[] loadCheckpoint(String key) throws IgniteSpiException {
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean saveCheckpoint(String key, byte[] state, long timeout, boolean overwrite)
+            throws IgniteSpiException {
+            cntr.incrementAndGet();
+
+            return true;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean removeCheckpoint(String key) {
+            cntr.decrementAndGet();
+
+            return true;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void setCheckpointListener(CheckpointListener lsnr) {
+            // No-op.
+        }
+
+        /** {@inheritDoc} */
+        @Override public void spiStart(@Nullable String gridName) throws IgniteSpiException {
+            // No-op.
+        }
+
+        /** {@inheritDoc} */
+        @Override public void spiStop() throws IgniteSpiException {
+            // No-op.
+        }
+    }
+
+    /**
+     *
+     */
+    @ComputeTaskSessionFullSupport
+    private static class CheckpointCountingTestTask extends ComputeTaskAdapter<ClusterNode, Object> {
+        /** {@inheritDoc} */
+        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable ClusterNode arg)
+            throws IgniteCheckedException {
+            for (ClusterNode node : subgrid) {
+                if (node.id().equals(arg.id()))
+                    return Collections.singletonMap(new ComputeJobAdapter() {
+                        @IgniteTaskSessionResource
+                        private ComputeTaskSession ses;
+
+                        @Nullable @Override public Object execute() throws IgniteCheckedException {
+                            ses.saveCheckpoint("checkpoint-key", "checkpoint-value");
+
+                            return null;
+                        }
+                    }, node);
+            }
+
+            assert false : "Expected node wasn't found in grid";
+
+            // Never accessible.
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object reduce(List<ComputeJobResult> results) throws IgniteCheckedException {
+            return null;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridJobCollisionCancelSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridJobCollisionCancelSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridJobCollisionCancelSelfTest.java
new file mode 100644
index 0000000..bf967a6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridJobCollisionCancelSelfTest.java
@@ -0,0 +1,276 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.*;
+import org.apache.ignite.spi.collision.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import java.io.*;
+import java.util.*;
+
+/**
+ * Job collision cancel test.
+ */
+@SuppressWarnings( {"PublicInnerClass"})
+@GridCommonTest(group = "Kernal Self")
+public class GridJobCollisionCancelSelfTest extends GridCommonAbstractTest {
+    /** */
+    private static final Object mux = new Object();
+
+    /** */
+    private static final int SPLIT_COUNT = 2;
+
+    /** */
+    private static final long maxJobExecTime = 10000;
+
+    /** */
+    private static int cancelCnt;
+
+    /** */
+    private static int execCnt;
+
+    /** */
+    private static int colResolutionCnt;
+
+    /** */
+    public GridJobCollisionCancelSelfTest() {
+        super(true);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings( {"AssignmentToCatchBlockParameter"})
+    public void testCancel() throws Exception {
+        Ignite ignite = G.ignite(getTestGridName());
+
+        ignite.compute().localDeployTask(GridCancelTestTask.class, GridCancelTestTask.class.getClassLoader());
+
+        ComputeTaskFuture<?> res0 =
+            executeAsync(ignite.compute().withTimeout(maxJobExecTime * 2), GridCancelTestTask.class.getName(), null);
+
+        try {
+            Object res = res0.get();
+
+            info("Cancel test result: " + res);
+
+            synchronized (mux) {
+                // Every execute must be called.
+                assert execCnt <= SPLIT_COUNT : "Invalid execute count: " + execCnt;
+
+                // Job returns 1 if was cancelled.
+                assert (Integer)res <= SPLIT_COUNT  : "Invalid task result: " + res;
+
+                // Should be exactly the same as Jobs number.
+                assert cancelCnt <= SPLIT_COUNT : "Invalid cancel count: " + cancelCnt;
+
+                // One per start and one per stop and some that come with heartbeats.
+                assert colResolutionCnt > SPLIT_COUNT + 1:
+                    "Invalid collision resolution count: " + colResolutionCnt;
+            }
+        }
+        catch (ComputeTaskTimeoutException e) {
+            error("Task execution got timed out.", e);
+        }
+        catch (Exception e) {
+            assert e.getCause() != null;
+
+            if (e.getCause() instanceof IgniteCheckedException)
+                e = (Exception)e.getCause();
+
+            if (e.getCause() instanceof IOException)
+                e = (Exception)e.getCause();
+
+            assert e.getCause() instanceof InterruptedException : "Invalid exception cause: " + e.getCause();
+        }
+    }
+
+    /**
+     * @return Configuration.
+     * @throws Exception If failed.
+     */
+    @Override protected IgniteConfiguration getConfiguration() throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration();
+
+        cfg.setCollisionSpi(new GridTestCollision());
+
+        return cfg;
+    }
+
+    /**
+     *
+     */
+    public static class GridCancelTestTask extends ComputeTaskSplitAdapter<Serializable, Object> {
+        /** */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** {@inheritDoc} */
+        @Override public Collection<? extends ComputeJob> split(int gridSize, Serializable arg) {
+            if (log.isInfoEnabled())
+                log.info("Splitting task [task=" + this + ", gridSize=" + gridSize + ", arg=" + arg + ']');
+
+            Collection<GridCancelTestJob> jobs = new ArrayList<>(SPLIT_COUNT);
+
+            for (int i = 0; i < SPLIT_COUNT; i++)
+                jobs.add(new GridCancelTestJob());
+
+            return jobs;
+        }
+
+
+        /** {@inheritDoc} */
+        @Override public Object reduce(List<ComputeJobResult> results) {
+            if (log.isInfoEnabled())
+                log.info("Aggregating job [job=" + this + ", results=" + results + ']');
+
+            int res = 0;
+
+            for (ComputeJobResult result : results) {
+                assert result != null;
+
+                if (result.getData() != null)
+                    res += (Integer)result.getData();
+            }
+
+            return res;
+        }
+    }
+
+    /**
+     * Test job.
+     */
+    public static class GridCancelTestJob extends ComputeJobAdapter {
+        /** */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** */
+        @IgniteJobContextResource
+        private ComputeJobContext jobCtx;
+
+        /** */
+        @SuppressWarnings( {"FieldAccessedSynchronizedAndUnsynchronized"})
+        private boolean isCancelled;
+
+        /** */
+        private final long thresholdTime;
+
+        /** */
+        public GridCancelTestJob() {
+            thresholdTime = System.currentTimeMillis() + maxJobExecTime;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Serializable execute() {
+            synchronized (mux) {
+                execCnt++;
+            }
+
+            if (log.isInfoEnabled())
+                log.info("Executing job: " + jobCtx.getJobId());
+
+            long now = System.currentTimeMillis();
+
+            while (!isCancelled && now < thresholdTime) {
+                synchronized (mux) {
+                    try {
+                        mux.wait(thresholdTime - now);
+                    }
+                    catch (InterruptedException ignored) {
+                        // No-op.
+                    }
+                }
+
+                now = System.currentTimeMillis();
+            }
+
+            synchronized (mux) {
+                return isCancelled ? 1 : 0;
+            }
+        }
+
+        /** {@inheritDoc} */
+        @Override public void cancel() {
+            synchronized (mux) {
+                isCancelled = true;
+
+                cancelCnt++;
+
+                mux.notifyAll();
+            }
+
+            log.warning("Job cancelled: " + jobCtx.getJobId());
+        }
+    }
+
+
+    /**
+     * Test collision SPI.
+     */
+    @IgniteSpiMultipleInstancesSupport(true)
+    public static class GridTestCollision extends IgniteSpiAdapter implements CollisionSpi {
+        /** */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** {@inheritDoc} */
+        @Override public void onCollision(CollisionContext ctx) {
+            Collection<CollisionJobContext> activeJobs = ctx.activeJobs();
+            Collection<CollisionJobContext> waitJobs = ctx.waitingJobs();
+
+            synchronized (mux) {
+                colResolutionCnt++;
+            }
+
+            for (CollisionJobContext job : waitJobs)
+                job.activate();
+
+            for (CollisionJobContext job : activeJobs)
+                job.cancel();
+        }
+
+        /** {@inheritDoc} */
+        @Override public void spiStart(String gridName) throws IgniteSpiException {
+            // Start SPI start stopwatch.
+            startStopwatch();
+
+            // Ack start.
+            if (log.isInfoEnabled())
+                log.info(startInfo());
+        }
+
+        /** {@inheritDoc} */
+        @Override public void spiStop() throws IgniteSpiException {
+            // Ack stop.
+            if (log.isInfoEnabled())
+                log.info(stopInfo());
+        }
+
+        /** {@inheritDoc} */
+        @Override public void setExternalCollisionListener(CollisionExternalListener lsnr) {
+            // No-op.
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridJobContextSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridJobContextSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridJobContextSelfTest.java
new file mode 100644
index 0000000..d0c1a70
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridJobContextSelfTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.testframework.junits.common.*;
+
+import java.io.*;
+import java.util.*;
+
+/**
+ * Job context test.
+ */
+@GridCommonTest(group = "Kernal Self")
+public class GridJobContextSelfTest extends GridCommonAbstractTest {
+    /**
+     * @throws Exception If anything failed.
+     */
+    public void testJobContext() throws Exception {
+        Ignite ignite = startGrid(1);
+
+        try {
+            startGrid(2);
+
+            try {
+                ignite.compute().execute(JobContextTask.class, null);
+            }
+            finally {
+                stopGrid(2);
+            }
+        }
+        finally{
+            stopGrid(1);
+        }
+    }
+
+    /** */
+    @SuppressWarnings("PublicInnerClass")
+    public static class JobContextTask extends ComputeTaskSplitAdapter<Object, Object> {
+        @Override protected Collection<? extends ComputeJob> split(int gridSize, Object arg) throws IgniteCheckedException {
+            Collection<ComputeJobAdapter> jobs = new ArrayList<>(gridSize);
+
+            for (int i = 0; i < gridSize; i++) {
+                jobs.add(new ComputeJobAdapter() {
+                    /** */
+                    @IgniteJobContextResource
+                    private ComputeJobContext jobCtx;
+
+                    /** Ignite instance. */
+                    @IgniteInstanceResource
+                    private Ignite ignite;
+
+                    /** {@inheritDoc} */
+                    @Override public Serializable execute() {
+                        UUID locNodeId = ignite.configuration().getNodeId();
+
+                        jobCtx.setAttribute("nodeId", locNodeId);
+                        jobCtx.setAttribute("jobId", jobCtx.getJobId());
+
+                        Map<String, String> attrs = new HashMap<>(10);
+
+                        for (int i = 0; i < 10; i++) {
+                            String s = jobCtx.getJobId().toString() + i;
+
+                            attrs.put(s, s);
+                        }
+
+                        jobCtx.setAttributes(attrs);
+
+                        assert jobCtx.getAttribute("nodeId").equals(locNodeId);
+                        assert jobCtx.getAttributes().get("nodeId").equals(locNodeId);
+                        assert jobCtx.getAttributes().keySet().containsAll(attrs.keySet());
+                        assert jobCtx.getAttributes().values().containsAll(attrs.values());
+
+                        return null;
+                    }
+                });
+            }
+
+            return jobs;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object reduce(List<ComputeJobResult> results) throws IgniteCheckedException {
+            for (ComputeJobResult res : results) {
+                ComputeJobContext jobCtx = res.getJobContext();
+
+                assert jobCtx.getAttribute("nodeId").equals(res.getNode().id());
+                assert jobCtx.getAttributes().get("nodeId").equals(res.getNode().id());
+
+                assert jobCtx.getAttribute("jobId").equals(jobCtx.getJobId());
+
+                for (int i = 0; i < 10; i++) {
+                    String s = jobCtx.getJobId().toString() + i;
+
+                    assert jobCtx.getAttribute(s).equals(s);
+                    assert jobCtx.getAttributes().get(s).equals(s);
+                }
+            }
+
+            return null;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridJobMasterLeaveAwareSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridJobMasterLeaveAwareSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridJobMasterLeaveAwareSelfTest.java
new file mode 100644
index 0000000..cb51453
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridJobMasterLeaveAwareSelfTest.java
@@ -0,0 +1,802 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cache.*;
+import org.apache.ignite.cache.affinity.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.marshaller.optimized.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.*;
+import org.apache.ignite.internal.managers.communication.*;
+import org.apache.ignite.spi.communication.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+import org.apache.ignite.internal.util.direct.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.jetbrains.annotations.*;
+
+import java.util.*;
+import java.util.concurrent.*;
+
+import static java.util.concurrent.TimeUnit.*;
+import static org.apache.ignite.cache.GridCacheMode.*;
+
+/**
+ * Test behavior of jobs when master node has failed, but job class implements {@link org.apache.ignite.compute.ComputeJobMasterLeaveAware}
+ * interface.
+ */
+@GridCommonTest(group = "Task Session")
+public class GridJobMasterLeaveAwareSelfTest extends GridCommonAbstractTest {
+    /** Total grid count within the cloud. */
+    private static final int GRID_CNT = 2;
+
+    /** Default IP finder for single-JVM cloud grid. */
+    private static final TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+    /** Counts how many times master-leave interface implementation was called. */
+    private static volatile CountDownLatch invokeLatch;
+
+    /** Latch which blocks job execution until main thread has sent node fail signal. */
+    private static volatile CountDownLatch latch;
+
+    /** Latch which blocks main thread until all jobs start their execution. */
+    private static volatile CountDownLatch jobLatch;
+
+    /** Should job wait for callback. */
+    private static volatile boolean awaitMasterLeaveCallback = true;
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        stopAllGrids();
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        awaitMasterLeaveCallback = true;
+        latch = new CountDownLatch(1);
+        jobLatch = new CountDownLatch(GRID_CNT - 1);
+        invokeLatch  = new CountDownLatch(GRID_CNT - 1);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+        discoSpi.setIpFinder(ipFinder);
+
+        cfg.setDiscoverySpi(discoSpi);
+
+        cfg.setCommunicationSpi(new CommunicationSpi());
+        cfg.setMarshaller(new IgniteOptimizedMarshaller(false));
+
+        CacheConfiguration ccfg = defaultCacheConfiguration();
+
+        ccfg.setCacheMode(PARTITIONED);
+
+        cfg.setCacheConfiguration(ccfg);
+
+        return cfg;
+    }
+
+    /**
+     * Get predicate which allows task execution on all nodes except the last one.
+     *
+     * @return Predicate.
+     */
+    private IgnitePredicate<ClusterNode> excludeLastPredicate() {
+        return new IgnitePredicate<ClusterNode>() {
+            @Override public boolean apply(ClusterNode e) {
+                return !e.id().equals(grid(GRID_CNT - 1).localNode().id());
+            }
+        };
+    }
+
+    /**
+     * Constructor.
+     */
+    public GridJobMasterLeaveAwareSelfTest() {
+        super(/* don't start grid */ false);
+    }
+
+    /**
+     * Ensure that {@link org.apache.ignite.compute.ComputeJobMasterLeaveAware} callback is invoked on job which is initiated by
+     * master and is currently running on it.
+     *
+     * @throws Exception If failed.
+     */
+    public void testLocalJobOnMaster() throws Exception {
+        invokeLatch  = new CountDownLatch(1);
+        jobLatch = new CountDownLatch(1);
+
+        Ignite g = startGrid(0);
+
+        g.compute().enableAsync().execute(new TestTask(1), null);
+
+        jobLatch.await();
+
+        // Count down the latch in a separate thread.
+        new Thread(new Runnable() {
+            @Override public void run() {
+                try {
+                    U.sleep(500);
+                }
+                catch (IgniteInterruptedException ignore) {
+                    // No-op.
+                }
+
+                latch.countDown();
+            }
+        }).start();
+
+        stopGrid(0, true);
+
+        latch.countDown();
+
+        assert invokeLatch.await(5000, MILLISECONDS);
+    }
+
+    /**
+     * Ensure that {@link org.apache.ignite.compute.ComputeJobMasterLeaveAware} callback is invoked when master node leaves topology normally.
+     *
+     * @throws Exception If failed.
+     */
+    public void testMasterStoppedNormally() throws Exception {
+        // Start grids.
+        for (int i = 0; i < GRID_CNT; i++)
+            startGrid(i);
+
+        int lastGridIdx = GRID_CNT - 1;
+
+        compute(grid(lastGridIdx).forPredicate(excludeLastPredicate())).enableAsync().
+            execute(new TestTask(GRID_CNT - 1), null);
+
+        jobLatch.await();
+
+        stopGrid(lastGridIdx, true);
+
+        latch.countDown();
+
+        assert invokeLatch.await(5000, MILLISECONDS);
+    }
+
+    /**
+     * Ensure that {@link org.apache.ignite.compute.ComputeJobMasterLeaveAware} callback is invoked when master node leaves topology
+     * abruptly (e.g. due to a network failure or immediate node shutdown).
+     *
+     * @throws Exception If failed.
+     */
+    public void testMasterStoppedAbruptly() throws Exception {
+        // Start grids.
+        for (int i = 0; i < GRID_CNT; i++)
+            startGrid(i);
+
+        int lastGridIdx = GRID_CNT - 1;
+
+        compute(grid(lastGridIdx).forPredicate(excludeLastPredicate())).enableAsync().
+            execute(new TestTask(GRID_CNT - 1), null);
+
+        jobLatch.await();
+
+        ((CommunicationSpi)grid(lastGridIdx).configuration().getCommunicationSpi()).blockMessages();
+
+        stopGrid(lastGridIdx, true);
+
+        latch.countDown();
+
+        assert invokeLatch.await(5000, MILLISECONDS);
+    }
+
+    /**
+     * Ensure that {@link org.apache.ignite.compute.ComputeJobMasterLeaveAware} callback is invoked when fails to send
+     * {@link GridJobExecuteResponse} to master node.
+     *
+     * @throws Exception If failed.
+     */
+    public void testCannotSendJobExecuteResponse() throws Exception {
+        awaitMasterLeaveCallback = false;
+
+        // Start grids.
+        for (int i = 0; i < GRID_CNT; i++)
+            startGrid(i);
+
+        int lastGridIdx = GRID_CNT - 1;
+
+        compute(grid(lastGridIdx).forPredicate(excludeLastPredicate())).enableAsync().
+            execute(new TestTask(GRID_CNT - 1), null);
+
+        jobLatch.await();
+
+        for (int i = 0; i < lastGridIdx; i++)
+            ((CommunicationSpi)grid(i).configuration().getCommunicationSpi()).waitLatch();
+
+        latch.countDown();
+
+        // Ensure that all worker nodes has already started job response sending.
+        for (int i = 0; i < lastGridIdx; i++)
+            ((CommunicationSpi)grid(i).configuration().getCommunicationSpi()).awaitResponse();
+
+        // Now we stop master grid.
+        stopGrid(lastGridIdx, true);
+
+        // Release communication SPI wait latches. As master node is stopped, job worker will receive and exception.
+        for (int i = 0; i < lastGridIdx; i++)
+            ((CommunicationSpi)grid(i).configuration().getCommunicationSpi()).releaseWaitLatch();
+
+        assert invokeLatch.await(5000, MILLISECONDS);
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testApply1() throws Exception {
+        testMasterLeaveAwareCallback(1, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup grid) throws IgniteCheckedException {
+                IgniteCompute comp = compute(grid).enableAsync();
+
+                comp.apply(new TestClosure(), "arg");
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testApply2() throws Exception {
+        testMasterLeaveAwareCallback(2, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup grid) throws IgniteCheckedException {
+                IgniteCompute comp = compute(grid).enableAsync();
+
+                comp.apply(new TestClosure(), Arrays.asList("arg1", "arg2"));
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testApply3() throws Exception {
+        testMasterLeaveAwareCallback(2, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup grid) throws IgniteCheckedException {
+                IgniteCompute comp = compute(grid).enableAsync();
+
+                comp.apply(new TestClosure(),
+                    Arrays.asList("arg1", "arg2"),
+                    new IgniteReducer<Void, Object>() {
+                        @Override public boolean collect(@Nullable Void aVoid) {
+                            return true;
+                        }
+
+                        @Override public Object reduce() {
+                            return null;
+                        }
+                    });
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testRun1() throws Exception {
+        testMasterLeaveAwareCallback(1, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                comp.run(new TestRunnable());
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testRun2() throws Exception {
+        testMasterLeaveAwareCallback(2, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                comp.run(Arrays.asList(new TestRunnable(), new TestRunnable()));
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testCall1() throws Exception {
+        testMasterLeaveAwareCallback(1, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                comp.call(new TestCallable());
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testCall2() throws Exception {
+        testMasterLeaveAwareCallback(2, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                comp.call(Arrays.asList(new TestCallable(), new TestCallable()));
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testCall3() throws Exception {
+        testMasterLeaveAwareCallback(2, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                comp.call(
+                    Arrays.asList(new TestCallable(), new TestCallable()),
+                    new IgniteReducer<Void, Object>() {
+                        @Override public boolean collect(@Nullable Void aVoid) {
+                            return true;
+                        }
+
+                        @Override public Object reduce() {
+                            return null;
+                        }
+                    });
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testBroadcast1() throws Exception {
+        testMasterLeaveAwareCallback(1, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                comp.broadcast(new TestRunnable());
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testBroadcast2() throws Exception {
+        testMasterLeaveAwareCallback(1, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                comp.broadcast(new TestCallable());
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testBroadcast3() throws Exception {
+        testMasterLeaveAwareCallback(1, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                comp.broadcast(new TestClosure(), "arg");
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAffinityRun() throws Exception {
+        testMasterLeaveAwareCallback(1, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                GridCacheAffinity<Object> aff = prj.ignite().cache(null).affinity();
+
+                ClusterNode node = F.first(prj.nodes());
+
+                comp.affinityRun(null, keyForNode(aff, node), new TestRunnable());
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @throws Exception If failed.
+     */
+    public void testAffinityCall() throws Exception {
+        testMasterLeaveAwareCallback(1, new CX1<ClusterGroup, IgniteFuture<?>>() {
+            @Override public IgniteFuture<?> applyx(ClusterGroup prj) throws IgniteCheckedException {
+                IgniteCompute comp = compute(prj).enableAsync();
+
+                GridCacheAffinity<Object> aff = prj.ignite().cache(null).affinity();
+
+                ClusterNode node = F.first(prj.nodes());
+
+                comp.affinityCall(null, keyForNode(aff, node), new TestCallable());
+
+                return comp.future();
+            }
+        });
+    }
+
+    /**
+     * @param aff Cache affinity.
+     * @param node Node.
+     * @return Finds some cache key for which given node is primary.
+     */
+    private Object keyForNode(GridCacheAffinity<Object> aff, ClusterNode node) {
+        assertNotNull(node);
+
+        Object key = null;
+
+        for (int i = 0; i < 1000; i++) {
+            if (aff.isPrimary(node, i)) {
+                key = i;
+
+                break;
+            }
+        }
+
+        assertNotNull(key);
+
+        return key;
+    }
+
+    /**
+     * @param expJobs Expected jobs number.
+     * @param taskStarter Task started.
+     * @throws Exception If failed.
+     */
+    private void testMasterLeaveAwareCallback(int expJobs, IgniteClosure<ClusterGroup, IgniteFuture<?>> taskStarter)
+        throws Exception {
+        jobLatch = new CountDownLatch(expJobs);
+        invokeLatch  = new CountDownLatch(expJobs);
+
+        for (int i = 0; i < GRID_CNT; i++)
+            startGrid(i);
+
+        int lastGridIdx = GRID_CNT - 1;
+
+        IgniteFuture<?> fut = taskStarter.apply(grid(lastGridIdx).forPredicate(excludeLastPredicate()));
+
+        jobLatch.await();
+
+        stopGrid(lastGridIdx, true);
+
+        latch.countDown();
+
+        assert invokeLatch.await(5000, MILLISECONDS);
+
+        try {
+            fut.get();
+        }
+        catch (IgniteCheckedException e) {
+            log.debug("Task failed: " + e);
+        }
+    }
+
+    /**
+     */
+    private static class TestMasterLeaveAware {
+        /** */
+        private final CountDownLatch latch0 = new CountDownLatch(1);
+
+        /**
+         * @param log Logger.
+         */
+        private void execute(IgniteLogger log) {
+            try {
+                log.info("Started execute.");
+
+                // Countdown shared job latch so that the main thread know that all jobs are
+                // inside the "execute" routine.
+                jobLatch.countDown();
+
+                log.info("After job latch.");
+
+                // Await for the main thread to allow jobs to proceed.
+                latch.await();
+
+                log.info("After latch.");
+
+                if (awaitMasterLeaveCallback) {
+                    latch0.await();
+
+                    log.info("After latch0.");
+                }
+                else
+                    log.info("Latch 0 skipped.");
+            }
+            catch (InterruptedException e) {
+                // We do not expect any interruptions here, hence this statement.
+                fail("Unexpected exception: " + e);
+            }
+        }
+
+        /**
+         * @param log Logger.
+         * @param job Actual job.
+         */
+        private void onMasterLeave(IgniteLogger log, Object job) {
+            log.info("Callback executed: " + job);
+
+            latch0.countDown();
+
+            invokeLatch.countDown();
+        }
+    }
+
+    /**
+     * Master leave aware callable.
+     */
+    private static class TestCallable implements Callable<Void>, ComputeJobMasterLeaveAware {
+        /** Task session. */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** */
+        private TestMasterLeaveAware masterLeaveAware = new TestMasterLeaveAware();
+
+        /** {@inheritDoc} */
+        @Override public Void call() throws Exception {
+            masterLeaveAware.execute(log);
+
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException {
+            masterLeaveAware.onMasterLeave(log, this);
+        }
+    }
+
+    /**
+     * Master leave aware runnable.
+     */
+    private static class TestRunnable implements Runnable, ComputeJobMasterLeaveAware {
+        /** Task session. */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** */
+        private TestMasterLeaveAware masterLeaveAware = new TestMasterLeaveAware();
+
+        /** {@inheritDoc} */
+        @Override public void run() {
+            masterLeaveAware.execute(log);
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException {
+            masterLeaveAware.onMasterLeave(log, this);
+        }
+    }
+
+    /**
+     * Master leave aware closure.
+     */
+    private static class TestClosure implements IgniteClosure<String, Void>, ComputeJobMasterLeaveAware {
+        /** Task session. */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** */
+        private TestMasterLeaveAware masterLeaveAware = new TestMasterLeaveAware();
+
+        /** {@inheritDoc} */
+        @Override public Void apply(String arg) {
+            masterLeaveAware.execute(log);
+
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException {
+            masterLeaveAware.onMasterLeave(log, this);
+        }
+    }
+
+    /**
+     * Base implementation of dummy task which produces predefined amount of test jobs on split stage.
+     */
+    private static class TestTask extends ComputeTaskSplitAdapter<String, Integer> {
+        /** How many jobs to produce. */
+        private int jobCnt;
+
+        /** */
+        @IgniteTaskSessionResource
+        private ComputeTaskSession taskSes;
+
+        /**
+         * Constructor.
+         *
+         * @param jobCnt How many jobs to produce on split stage.
+         */
+        private TestTask(int jobCnt) {
+            this.jobCnt = jobCnt;
+        }
+
+        /** {@inheritDoc} */
+        @Override protected Collection<? extends ComputeJob> split(int gridSize, String arg) throws IgniteCheckedException {
+            Collection<ComputeJobAdapter> jobs = new ArrayList<>(jobCnt);
+
+            for (int i = 0; i < jobCnt; i++)
+                jobs.add(new TestJob());
+
+            return jobs;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Integer reduce(List<ComputeJobResult> results) throws IgniteCheckedException {
+            return null;
+        }
+    }
+
+    /**
+     * Base implementation of dummy test job.
+     */
+    private static class TestJob extends ComputeJobAdapter implements ComputeJobMasterLeaveAware {
+        /** Task session. */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** */
+        private TestMasterLeaveAware masterLeaveAware = new TestMasterLeaveAware();
+
+        /**
+         * Constructor
+         */
+        private TestJob() {
+            super(new Object());
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object execute() throws IgniteCheckedException {
+            masterLeaveAware.execute(log);
+
+            return null;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void onMasterNodeLeft(ComputeTaskSession ses) throws IgniteCheckedException {
+            masterLeaveAware.onMasterLeave(log, this);
+        }
+    }
+
+    /**
+     * Communication SPI which could optionally block outgoing messages.
+     */
+    private static class CommunicationSpi extends TcpCommunicationSpi {
+        /** Whether to block all outgoing messages. */
+        private volatile boolean block;
+
+        /** Job execution response latch. */
+        private CountDownLatch respLatch = new CountDownLatch(1);
+
+        /** Whether to wait for a wait latch before returning. */
+        private volatile boolean wait;
+
+        /** Wait latch. */
+        private CountDownLatch waitLatch = new CountDownLatch(1);
+
+        /** {@inheritDoc} */
+        @Override public void sendMessage(ClusterNode node, GridTcpCommunicationMessageAdapter msg)
+            throws IgniteSpiException {
+            sendMessage0(node, msg);
+        }
+
+        /**
+         * Send message optionally either blocking it or throwing an exception if it is of
+         * {@link GridJobExecuteResponse} type.
+         *
+         * @param node Destination node.
+         * @param msg Message to be sent.
+         * @throws org.apache.ignite.spi.IgniteSpiException If failed.
+         */
+        private void sendMessage0(ClusterNode node, GridTcpCommunicationMessageAdapter msg) throws IgniteSpiException {
+            if (msg instanceof GridIoMessage) {
+                GridIoMessage msg0 = (GridIoMessage)msg;
+
+                if (msg0.message() instanceof GridJobExecuteResponse) {
+                    respLatch.countDown();
+
+                    if (wait) {
+                        try {
+                            U.await(waitLatch);
+                        }
+                        catch (IgniteInterruptedException ignore) {
+                            // No-op.
+                        }
+                    }
+                }
+            }
+
+            if (!block)
+                super.sendMessage(node, msg);
+        }
+
+        /**
+         * Block all outgoing message.
+         */
+        void blockMessages() {
+            block = true;
+        }
+
+        /**
+         * Whether to block on a wait latch.
+         */
+        private void waitLatch() {
+            wait = true;
+        }
+
+        /**
+         * Count down wait latch.
+         */
+        private void releaseWaitLatch() {
+            waitLatch.countDown();
+        }
+
+        /**
+         * Await for job execution response to come.
+         *
+         * @throws org.apache.ignite.IgniteInterruptedException If interrupted.
+         */
+        private void awaitResponse() throws IgniteInterruptedException {
+            U.await(respLatch);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingSelfTest.java
new file mode 100644
index 0000000..d918ccb
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingSelfTest.java
@@ -0,0 +1,439 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.lang.*;
+import org.apache.ignite.marshaller.optimized.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.collision.jobstealing.*;
+import org.apache.ignite.spi.failover.jobstealing.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.testframework.config.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+
+/**
+ * Job stealing test.
+ */
+@SuppressWarnings("unchecked")
+@GridCommonTest(group = "Kernal Self")
+public class GridJobStealingSelfTest extends GridCommonAbstractTest {
+    /** Task execution timeout in milliseconds. */
+    private static final int TASK_EXEC_TIMEOUT_MS = 50000;
+
+    /** */
+    private Ignite ignite1;
+
+    /** */
+    private Ignite ignite2;
+
+    /** Job distribution map. Records which job has run on which node. */
+    private static Map<UUID, Collection<ComputeJob>> jobDistrMap = new HashMap<>();
+
+    /** */
+    public GridJobStealingSelfTest() {
+        super(false /* don't start grid*/);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        jobDistrMap.clear();
+
+        ignite1 = startGrid(1);
+
+        ignite2 = startGrid(2);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        stopAllGrids();
+
+        ignite1 = null;
+        ignite2 = null;
+    }
+
+    /**
+     * Test 2 jobs on 1 node.
+     *
+     * @throws IgniteCheckedException If test failed.
+     */
+    public void testTwoJobs() throws IgniteCheckedException {
+        executeAsync(ignite1.compute(), new JobStealingSingleNodeTask(2), null).get(TASK_EXEC_TIMEOUT_MS);
+
+        // Verify that 1 job was stolen by second node.
+        assertEquals(2, jobDistrMap.keySet().size());
+        assertEquals(1, jobDistrMap.get(ignite1.cluster().localNode().id()).size());
+        assertEquals(1, jobDistrMap.get(ignite2.cluster().localNode().id()).size());
+    }
+
+    /**
+     * Test 2 jobs on 1 node with null predicate.
+     *
+     * @throws IgniteCheckedException If test failed.
+     */
+    @SuppressWarnings("NullArgumentToVariableArgMethod")
+    public void testTwoJobsNullPredicate() throws IgniteCheckedException {
+        executeAsync(ignite1.compute(), new JobStealingSingleNodeTask(2), null).get(TASK_EXEC_TIMEOUT_MS);
+
+        // Verify that 1 job was stolen by second node.
+        assertEquals(2, jobDistrMap.keySet().size());
+        assertEquals(1, jobDistrMap.get(ignite1.cluster().localNode().id()).size());
+        assertEquals(1, jobDistrMap.get(ignite2.cluster().localNode().id()).size());
+    }
+
+    /**
+     * Test 2 jobs on 1 node with null predicate using string task name.
+     *
+     * @throws IgniteCheckedException If test failed.
+     */
+    @SuppressWarnings("NullArgumentToVariableArgMethod")
+    public void testTwoJobsTaskNameNullPredicate() throws IgniteCheckedException {
+        executeAsync(ignite1.compute(), JobStealingSingleNodeTask.class.getName(), null).get(TASK_EXEC_TIMEOUT_MS);
+
+        // Verify that 1 job was stolen by second node.
+        assertEquals(2, jobDistrMap.keySet().size());
+        assertEquals(1, jobDistrMap.get(ignite1.cluster().localNode().id()).size());
+        assertEquals(1, jobDistrMap.get(ignite2.cluster().localNode().id()).size());
+    }
+
+    /**
+     * Test 2 jobs on 1 node when one of the predicates is null.
+     *
+     * @throws IgniteCheckedException If test failed.
+     */
+    @SuppressWarnings("unchecked")
+    public void testTwoJobsPartiallyNullPredicate() throws IgniteCheckedException {
+        IgnitePredicate<ClusterNode> topPred =  new IgnitePredicate<ClusterNode>() {
+                @Override public boolean apply(ClusterNode e) {
+                    return ignite2.cluster().localNode().id().equals(e.id()); // Limit projection with only grid2.
+                }
+            };
+
+        executeAsync(compute(ignite1.cluster().forPredicate(topPred)).withTimeout(TASK_EXEC_TIMEOUT_MS),
+            new JobStealingSpreadTask(2), null).get(TASK_EXEC_TIMEOUT_MS);
+
+        assertEquals(1, jobDistrMap.keySet().size());
+        assertEquals(2, jobDistrMap.get(ignite2.cluster().localNode().id()).size());
+        assertFalse(jobDistrMap.containsKey(ignite1.cluster().localNode().id()));
+    }
+
+    /**
+     * Tests that projection predicate is taken into account by Stealing SPI.
+     *
+     * @throws Exception If failed.
+     */
+    public void testProjectionPredicate() throws Exception {
+        final Ignite ignite3 = startGrid(3);
+
+        executeAsync(compute(ignite1.cluster().forPredicate(new P1<ClusterNode>() {
+            @Override public boolean apply(ClusterNode e) {
+                return ignite1.cluster().localNode().id().equals(e.id()) ||
+                    ignite3.cluster().localNode().id().equals(e.id()); // Limit projection with only grid1 or grid3 node.
+            }
+        })), new JobStealingSpreadTask(4), null).get(TASK_EXEC_TIMEOUT_MS);
+
+        // Verify that jobs were run only on grid1 and grid3 (not on grid2)
+        assertEquals(2, jobDistrMap.keySet().size());
+        assertEquals(2, jobDistrMap.get(ignite1.cluster().localNode().id()).size());
+        assertEquals(2, jobDistrMap.get(ignite3.cluster().localNode().id()).size());
+        assertFalse(jobDistrMap.containsKey(ignite2.cluster().localNode().id()));
+    }
+
+    /**
+     * Tests that projection predicate is taken into account by Stealing SPI,
+     * and that jobs in projection can steal tasks from each other.
+     *
+     * @throws Exception If failed.
+     */
+    public void testProjectionPredicateInternalStealing() throws Exception {
+        final Ignite ignite3 = startGrid(3);
+
+        IgnitePredicate<ClusterNode> p = new P1<ClusterNode>() {
+            @Override public boolean apply(ClusterNode e) {
+                return ignite1.cluster().localNode().id().equals(e.id()) ||
+                    ignite3.cluster().localNode().id().equals(e.id()); // Limit projection with only grid1 or grid3 node.
+            }
+        };
+
+        executeAsync(compute(ignite1.cluster().forPredicate(p)), new JobStealingSingleNodeTask(4), null).get(TASK_EXEC_TIMEOUT_MS);
+
+        // Verify that jobs were run only on grid1 and grid3 (not on grid2)
+        assertEquals(2, jobDistrMap.keySet().size());
+        assertFalse(jobDistrMap.containsKey(ignite2.cluster().localNode().id()));
+    }
+
+    /**
+     * Tests that a job is not cancelled if there are no
+     * available thief nodes in topology.
+     *
+     * @throws Exception If failed.
+     */
+    public void testSingleNodeTopology() throws Exception {
+        IgnitePredicate<ClusterNode> p = new IgnitePredicate<ClusterNode>() {
+            @Override public boolean apply(ClusterNode e) {
+                return ignite1.cluster().localNode().id().equals(e.id()); // Limit projection with only grid1 node.
+            }
+        };
+
+        executeAsync(compute(ignite1.cluster().forPredicate(p)), new JobStealingSpreadTask(2), null).
+            get(TASK_EXEC_TIMEOUT_MS);
+
+        assertEquals(1, jobDistrMap.keySet().size());
+        assertEquals(2, jobDistrMap.get(ignite1.cluster().localNode().id()).size());
+    }
+
+    /**
+     * Tests that a job is not cancelled if there are no
+     * available thief nodes in projection.
+     *
+     * @throws Exception If failed.
+     */
+    public void testSingleNodeProjection() throws Exception {
+        ClusterGroup prj = ignite1.cluster().forNodeIds(Collections.singleton(ignite1.cluster().localNode().id()));
+
+        executeAsync(compute(prj), new JobStealingSpreadTask(2), null).get(TASK_EXEC_TIMEOUT_MS);
+
+        assertEquals(1, jobDistrMap.keySet().size());
+        assertEquals(2, jobDistrMap.get(ignite1.cluster().localNode().id()).size());
+    }
+
+    /**
+     * Tests that a job is not cancelled if there are no
+     * available thief nodes in projection. Uses null predicate.
+     *
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings("NullArgumentToVariableArgMethod")
+    public void testSingleNodeProjectionNullPredicate() throws Exception {
+        ClusterGroup prj = ignite1.cluster().forNodeIds(Collections.singleton(ignite1.cluster().localNode().id()));
+
+        executeAsync(compute(prj).withTimeout(TASK_EXEC_TIMEOUT_MS), new JobStealingSpreadTask(2), null).
+            get(TASK_EXEC_TIMEOUT_MS);
+
+        assertEquals(1, jobDistrMap.keySet().size());
+        assertEquals(2, jobDistrMap.get(ignite1.cluster().localNode().id()).size());
+    }
+
+    /**
+     * Tests job stealing with peer deployment and different class loaders.
+     *
+     * @throws Exception If failed.
+     */
+    @SuppressWarnings("unchecked")
+    public void testProjectionPredicateDifferentClassLoaders() throws Exception {
+        final Ignite ignite3 = startGrid(3);
+
+        URL[] clsLdrUrls;
+        try {
+            clsLdrUrls = new URL[] {new URL(GridTestProperties.getProperty("p2p.uri.cls"))};
+        }
+        catch (MalformedURLException e) {
+            throw new RuntimeException("Define property p2p.uri.cls", e);
+        }
+
+        ClassLoader ldr1 = new URLClassLoader(clsLdrUrls, getClass().getClassLoader());
+
+        Class taskCls = ldr1.loadClass("org.gridgain.grid.tests.p2p.JobStealingTask");
+        Class nodeFilterCls = ldr1.loadClass("org.gridgain.grid.tests.p2p.GridExcludeNodeFilter");
+
+        IgnitePredicate<ClusterNode> nodeFilter = (IgnitePredicate<ClusterNode>)nodeFilterCls
+            .getConstructor(UUID.class).newInstance(ignite2.cluster().localNode().id());
+
+        Map<UUID, Integer> ret = (Map<UUID, Integer>)executeAsync(compute(ignite1.cluster().forPredicate(nodeFilter)),
+            taskCls, null).get(TASK_EXEC_TIMEOUT_MS);
+
+        assert ret != null;
+        assert ret.get(ignite1.cluster().localNode().id()) != null && ret.get(ignite1.cluster().localNode().id()) == 2 :
+            ret.get(ignite1.cluster().localNode().id());
+        assert ret.get(ignite3.cluster().localNode().id()) != null && ret.get(ignite3.cluster().localNode().id()) == 2 :
+            ret.get(ignite3.cluster().localNode().id());
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        JobStealingCollisionSpi colSpi = new JobStealingCollisionSpi();
+
+        // One job at a time.
+        colSpi.setActiveJobsThreshold(1);
+        colSpi.setWaitJobsThreshold(0);
+
+        JobStealingFailoverSpi failSpi = new JobStealingFailoverSpi();
+
+        // Verify defaults.
+        assert failSpi.getMaximumFailoverAttempts() == JobStealingFailoverSpi.DFLT_MAX_FAILOVER_ATTEMPTS;
+
+        cfg.setCollisionSpi(colSpi);
+        cfg.setFailoverSpi(failSpi);
+
+        cfg.setMarshaller(new IgniteOptimizedMarshaller(false));
+
+        return cfg;
+    }
+
+    /**
+     * Job stealing task, that spreads jobs equally over the grid.
+     */
+    private static class JobStealingSpreadTask extends ComputeTaskAdapter<Object, Object> {
+        /** Grid. */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
+        /** Logger. */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** Number of jobs to spawn from task. */
+        protected final int nJobs;
+
+        /**
+         * Constructs a new task instance.
+         *
+         * @param nJobs Number of jobs to spawn from this task.
+         */
+        JobStealingSpreadTask(int nJobs) {
+            this.nJobs = nJobs;
+        }
+
+        /** {@inheritDoc} */
+        @SuppressWarnings("ForLoopReplaceableByForEach")
+        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
+            @Nullable Object arg) throws IgniteCheckedException {
+            //assert subgrid.size() == 2 : "Invalid subgrid size: " + subgrid.size();
+
+            Map<ComputeJobAdapter, ClusterNode> map = new HashMap<>(subgrid.size());
+
+            Iterator<ClusterNode> subIter = subgrid.iterator();
+
+            // Spread jobs over subgrid.
+            for (int i = 0; i < nJobs; i++) {
+                if (!subIter.hasNext())
+                    subIter = subgrid.iterator(); // wrap around
+
+                map.put(new GridJobStealingJob(5000L), subIter.next());
+            }
+
+            return map;
+        }
+
+        /** {@inheritDoc} */
+        @SuppressWarnings("SuspiciousMethodCalls")
+        @Override public Object reduce(List<ComputeJobResult> results) throws IgniteCheckedException {
+            for (ComputeJobResult res : results) {
+                log.info("Job result: " + res.getData());
+            }
+
+            return null;
+        }
+    }
+
+    /**
+     * Job stealing task, that puts all jobs onto one node.
+     */
+    private static class JobStealingSingleNodeTask extends JobStealingSpreadTask {
+        /** {@inheritDoc} */
+        JobStealingSingleNodeTask(int nJobs) {
+            super(nJobs);
+        }
+
+        /**
+         * Default constructor.
+         *
+         * Uses 2 jobs.
+         */
+        JobStealingSingleNodeTask() {
+            super(2);
+        }
+
+        /** {@inheritDoc} */
+        @SuppressWarnings("ForLoopReplaceableByForEach")
+        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
+            @Nullable Object arg) throws IgniteCheckedException {
+            assert subgrid.size() > 1 : "Invalid subgrid size: " + subgrid.size();
+
+            Map<ComputeJobAdapter, ClusterNode> map = new HashMap<>(subgrid.size());
+
+            // Put all jobs onto one node.
+            for (int i = 0; i < nJobs; i++)
+                map.put(new GridJobStealingJob(5000L), subgrid.get(0));
+
+            return map;
+        }
+    }
+
+    /**
+     * Job stealing job.
+     */
+    private static final class GridJobStealingJob extends ComputeJobAdapter {
+        /** Injected grid. */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
+        /** Logger. */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /**
+         * @param arg Job argument.
+         */
+        GridJobStealingJob(Long arg) {
+            super(arg);
+        }
+
+        /** {@inheritDoc} */
+        @Override public Serializable execute() throws IgniteCheckedException {
+            log.info("Started job on node: " + ignite.cluster().localNode().id());
+
+            if (!jobDistrMap.containsKey(ignite.cluster().localNode().id())) {
+                Collection<ComputeJob> jobs = new ArrayList<>();
+                jobs.add(this);
+
+                jobDistrMap.put(ignite.cluster().localNode().id(), jobs);
+            }
+            else
+                jobDistrMap.get(ignite.cluster().localNode().id()).add(this);
+
+            try {
+                Long sleep = argument(0);
+
+                assert sleep != null;
+
+                Thread.sleep(sleep);
+            }
+            catch (InterruptedException e) {
+                log.info("Job got interrupted on node: " + ignite.cluster().localNode().id());
+
+                throw new IgniteCheckedException("Job got interrupted.", e);
+            }
+            finally {
+                log.info("Job finished on node: " + ignite.cluster().localNode().id());
+            }
+
+            return ignite.cluster().localNode().id();
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/537f631a/modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingZeroActiveJobsSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingZeroActiveJobsSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingZeroActiveJobsSelfTest.java
new file mode 100644
index 0000000..de0d669
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingZeroActiveJobsSelfTest.java
@@ -0,0 +1,169 @@
+/*
+ * 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;
+
+import org.apache.ignite.*;
+import org.apache.ignite.cluster.*;
+import org.apache.ignite.compute.*;
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.collision.jobstealing.*;
+import org.apache.ignite.spi.failover.jobstealing.*;
+import org.apache.ignite.testframework.junits.common.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.util.*;
+
+/**
+ * Job stealing test.
+ */
+@GridCommonTest(group = "Kernal Self")
+public class GridJobStealingZeroActiveJobsSelfTest extends GridCommonAbstractTest {
+    /** */
+    private static Ignite ignite1;
+
+    /** */
+    private static Ignite ignite2;
+
+    /** */
+    public GridJobStealingZeroActiveJobsSelfTest() {
+        super(false /* don't start grid*/);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void beforeTest() throws Exception {
+        ignite1 = startGrid(1);
+        ignite2 = startGrid(2);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void afterTest() throws Exception {
+        ignite1 = null;
+
+        stopGrid(1);
+        stopGrid(2);
+    }
+
+    /**
+     * Test 2 jobs on 2 nodes.
+     *
+     * @throws IgniteCheckedException If test failed.
+     */
+    public void testTwoJobs() throws IgniteCheckedException {
+        ignite1.compute().execute(JobStealingTask.class, null);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+        IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+        JobStealingCollisionSpi colSpi = new JobStealingCollisionSpi();
+
+        // One job at a time.
+        colSpi.setActiveJobsThreshold(gridName.endsWith("1") ? 0 : 2);
+        colSpi.setWaitJobsThreshold(0);
+
+        JobStealingFailoverSpi failSpi = new JobStealingFailoverSpi();
+
+        // Verify defaults.
+        assert failSpi.getMaximumFailoverAttempts() == JobStealingFailoverSpi.DFLT_MAX_FAILOVER_ATTEMPTS;
+
+        cfg.setCollisionSpi(colSpi);
+        cfg.setFailoverSpi(failSpi);
+
+        return cfg;
+    }
+
+    /** */
+    @SuppressWarnings({"PublicInnerClass"})
+    public static class JobStealingTask extends ComputeTaskAdapter<Object, Object> {
+        /** Grid. */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
+        /** Logger. */
+        @IgniteLoggerResource
+        private IgniteLogger log;
+
+        /** {@inheritDoc} */
+        @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Object arg) throws IgniteCheckedException {
+            assert subgrid.size() == 2 : "Invalid subgrid size: " + subgrid.size();
+
+            Map<ComputeJobAdapter, ClusterNode> map = new HashMap<>(subgrid.size());
+
+            // Put all jobs onto local node.
+            for (Iterator iter = subgrid.iterator(); iter.hasNext(); iter.next())
+                map.put(new GridJobStealingJob(5000L), ignite.cluster().localNode());
+
+            return map;
+        }
+
+        /** {@inheritDoc} */
+        @Override public Object reduce(List<ComputeJobResult> results) throws IgniteCheckedException {
+            assert results.size() == 2;
+
+            for (ComputeJobResult res : results) {
+                log.info("Job result: " + res.getData());
+            }
+
+            String name1 = results.get(0).getData();
+            String name2 = results.get(1).getData();
+
+            assert name1.equals(name2);
+
+            assert !name1.equals(ignite1.name());
+            assert name1.equals(ignite2.name());
+
+            return null;
+        }
+    }
+
+    /**
+     *
+     */
+    @SuppressWarnings({"PublicInnerClass"})
+    public static final class GridJobStealingJob extends ComputeJobAdapter {
+        /** Injected grid. */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
+        /**
+         * @param arg Job argument.
+         */
+        GridJobStealingJob(Long arg) {
+            super(arg);
+        }
+
+        /** {@inheritDoc} */
+        @Override public Serializable execute() throws IgniteCheckedException {
+            try {
+                Long sleep = argument(0);
+
+                assert sleep != null;
+
+                Thread.sleep(sleep);
+            }
+            catch (InterruptedException e) {
+                throw new IgniteCheckedException("Job got interrupted.", e);
+            }
+
+            return ignite.name();
+        }
+    }
+}


Mime
View raw message