modules/core/src/test/java/org/apache/ignite/internal/GridJobStealingSelfTest.java - ignite - Git at Google

 /*
  * 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 java.io.Serializable;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLClassLoader;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.IgniteLogger;
 import org.apache.ignite.cluster.ClusterGroup;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.compute.ComputeJob;
 import org.apache.ignite.compute.ComputeJobAdapter;
 import org.apache.ignite.compute.ComputeJobResult;
 import org.apache.ignite.compute.ComputeTaskAdapter;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.util.typedef.P1;
 import org.apache.ignite.lang.IgnitePredicate;
 import org.apache.ignite.resources.IgniteInstanceResource;
 import org.apache.ignite.resources.LoggerResource;
 import org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi;
 import org.apache.ignite.spi.failover.jobstealing.JobStealingFailoverSpi;
 import org.apache.ignite.testframework.config.GridTestProperties;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.apache.ignite.testframework.junits.common.GridCommonTest;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import org.junit.Ignore;
 import org.junit.Test;

 /**
  * 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.
      */
     @Test
     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.
      */
     @Test
     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.
      */
     @Test
     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.
      */
     @Test
     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.
      */
     @Test
     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.
      */
     @Test
     public void testProjectionPredicateInternalStealing() throws Exception {
         final Ignite ignite3 = startGrid(3);

         waitForTopology(3);

         final UUID node1 = ignite1.cluster().localNode().id();
         final UUID node3 = ignite3.cluster().localNode().id();

         IgnitePredicate<ClusterNode> p = new P1<ClusterNode>() {
             @Override public boolean apply(ClusterNode e) {
                 return node1.equals(e.id()) ||
                     node3.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.
      */
     @Test
     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.
      */
     @Test
     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.
      */
     @Test
     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")
     @Test
     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.apache.ignite.tests.p2p.JobStealingTask");
         Class nodeFilterCls = ldr1.loadClass("org.apache.ignite.tests.p2p.ExcludeNodeFilter");

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

     /**
      * @throws Exception If fatiled.
      */
     @Ignore("https://issues.apache.org/jira/browse/IGNITE-12629")
     @Test
     public void testJobStealingMbeanValidity() throws Exception {
         String[] beansToValidate = new String[] {
             "org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi$JobStealingCollisionSpiMBeanImpl",
             "org.apache.ignite.spi.failover.jobstealing.JobStealingFailoverSpi$JobStealingFailoverSpiMBeanImpl"};

         validateMbeans(ignite1, beansToValidate);
     }

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

         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);

         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. */
         @LoggerResource
         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} */
         @NotNull @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
             @Nullable Object arg) {
             //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} */
         @Override public Object reduce(List<ComputeJobResult> results) {
             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} */
         @NotNull @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Object arg) {
             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. */
         @LoggerResource
         private IgniteLogger log;

         /**
          * @param arg Job argument.
          */
         GridJobStealingJob(Long arg) {
             super(arg);
         }

         /** {@inheritDoc} */
         @Override public Serializable execute() {
             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 IgniteException("Job got interrupted.", e);
             }
             finally {
                 log.info("Job finished on node: " + ignite.cluster().localNode().id());
             }

             return ignite.cluster().localNode().id();
         }
     }
 }
	/*
	* 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 java.io.Serializable;
	import java.net.MalformedURLException;
	import java.net.URL;
	import java.net.URLClassLoader;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.Iterator;
	import java.util.List;
	import java.util.Map;
	import java.util.UUID;
	import org.apache.ignite.Ignite;
	import org.apache.ignite.IgniteCheckedException;
	import org.apache.ignite.IgniteException;
	import org.apache.ignite.IgniteLogger;
	import org.apache.ignite.cluster.ClusterGroup;
	import org.apache.ignite.cluster.ClusterNode;
	import org.apache.ignite.compute.ComputeJob;
	import org.apache.ignite.compute.ComputeJobAdapter;
	import org.apache.ignite.compute.ComputeJobResult;
	import org.apache.ignite.compute.ComputeTaskAdapter;
	import org.apache.ignite.configuration.IgniteConfiguration;
	import org.apache.ignite.internal.util.typedef.P1;
	import org.apache.ignite.lang.IgnitePredicate;
	import org.apache.ignite.resources.IgniteInstanceResource;
	import org.apache.ignite.resources.LoggerResource;
	import org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi;
	import org.apache.ignite.spi.failover.jobstealing.JobStealingFailoverSpi;
	import org.apache.ignite.testframework.config.GridTestProperties;
	import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
	import org.apache.ignite.testframework.junits.common.GridCommonTest;
	import org.jetbrains.annotations.NotNull;
	import org.jetbrains.annotations.Nullable;
	import org.junit.Ignore;
	import org.junit.Test;

	/**
	* 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.
	*/
	@Test
	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.
	*/
	@Test
	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.
	*/
	@Test
	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.
	*/
	@Test
	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.
	*/
	@Test
	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.
	*/
	@Test
	public void testProjectionPredicateInternalStealing() throws Exception {
	final Ignite ignite3 = startGrid(3);

	waitForTopology(3);

	final UUID node1 = ignite1.cluster().localNode().id();
	final UUID node3 = ignite3.cluster().localNode().id();

	IgnitePredicate<ClusterNode> p = new P1<ClusterNode>() {
	@Override public boolean apply(ClusterNode e) {
	return node1.equals(e.id()) \|\|
	node3.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.
	*/
	@Test
	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.
	*/
	@Test
	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.
	*/
	@Test
	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")
	@Test
	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.apache.ignite.tests.p2p.JobStealingTask");
	Class nodeFilterCls = ldr1.loadClass("org.apache.ignite.tests.p2p.ExcludeNodeFilter");

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

	/**
	* @throws Exception If fatiled.
	*/
	@Ignore("https://issues.apache.org/jira/browse/IGNITE-12629")
	@Test
	public void testJobStealingMbeanValidity() throws Exception {
	String[] beansToValidate = new String[] {
	"org.apache.ignite.spi.collision.jobstealing.JobStealingCollisionSpi$JobStealingCollisionSpiMBeanImpl",
	"org.apache.ignite.spi.failover.jobstealing.JobStealingFailoverSpi$JobStealingFailoverSpiMBeanImpl"};

	validateMbeans(ignite1, beansToValidate);
	}

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

	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);

	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. */
	@LoggerResource
	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} */
	@NotNull @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
	@Nullable Object arg) {
	//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} */
	@Override public Object reduce(List<ComputeJobResult> results) {
	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} */
	@NotNull @Override public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, @Nullable Object arg) {
	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. */
	@LoggerResource
	private IgniteLogger log;

	/**
	* @param arg Job argument.
	*/
	GridJobStealingJob(Long arg) {
	super(arg);
	}

	/** {@inheritDoc} */
	@Override public Serializable execute() {
	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 IgniteException("Job got interrupted.", e);
	}
	finally {
	log.info("Job finished on node: " + ignite.cluster().localNode().id());
	}

	return ignite.cluster().localNode().id();
	}
	}
	}