#ignite-1087: AffinityRun runs job on not primary nodes.
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/f9d2a2ef Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/f9d2a2ef Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/f9d2a2ef Branch: refs/heads/ignite-1121 Commit: f9d2a2ef4424bf166912594791b6414b9a9e8457 Parents: ede9612 Author: ivasilinets <ivasilin...@gridgain.com> Authored: Mon Jul 20 11:25:00 2015 +0300 Committer: ivasilinets <ivasilin...@gridgain.com> Committed: Mon Jul 20 11:25:00 2015 +0300 ---------------------------------------------------------------------- .../ignite/compute/ComputeJobResultPolicy.java | 3 +- .../failover/GridFailoverContextImpl.java | 28 ++- .../managers/failover/GridFailoverManager.java | 13 +- .../processors/closure/AffinityTask.java | 35 ++++ .../closure/GridClosureProcessor.java | 63 ++++++- .../processors/task/GridTaskWorker.java | 24 ++- .../ignite/spi/failover/FailoverContext.java | 18 ++ .../spi/failover/always/AlwaysFailoverSpi.java | 25 +++ .../cache/CacheAffinityCallSelfTest.java | 172 +++++++++++++++++++ .../cache/GridCacheAffinityRoutingSelfTest.java | 157 ++++++++++++++++- .../spi/failover/GridFailoverTestContext.java | 10 ++ .../ignite/testsuites/IgniteCacheTestSuite.java | 1 + 12 files changed, 533 insertions(+), 16 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobResultPolicy.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobResultPolicy.java b/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobResultPolicy.java index 37aba91..26eb542 100644 --- a/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobResultPolicy.java +++ b/modules/core/src/main/java/org/apache/ignite/compute/ComputeJobResultPolicy.java @@ -50,8 +50,7 @@ public enum ComputeJobResultPolicy { * @param ord Ordinal value. * @return Enumerated value. */ - @Nullable - public static ComputeJobResultPolicy fromOrdinal(byte ord) { + @Nullable public static ComputeJobResultPolicy fromOrdinal(byte ord) { return ord >= 0 && ord < VALS.length ? VALS[ord] : null; } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverContextImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverContextImpl.java index a3f8e44..c2b104e 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverContextImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverContextImpl.java @@ -24,6 +24,7 @@ import org.apache.ignite.internal.managers.loadbalancer.*; import org.apache.ignite.internal.util.tostring.*; import org.apache.ignite.internal.util.typedef.internal.*; import org.apache.ignite.spi.failover.*; +import org.jetbrains.annotations.*; import java.util.*; @@ -41,15 +42,26 @@ public class GridFailoverContextImpl implements FailoverContext { @GridToStringExclude private final GridLoadBalancerManager loadMgr; + /** Affinity key for affinityCall. */ + private final Object affKey; + + /** Affinity cache name for affinityCall. */ + private final String affCacheName; + /** * Initializes failover context. * * @param taskSes Grid task session. * @param jobRes Failed job result. * @param loadMgr Load manager. + * @param affKey Affinity key. + * @param affCacheName Affinity cache name. */ - public GridFailoverContextImpl(GridTaskSessionImpl taskSes, ComputeJobResult jobRes, - GridLoadBalancerManager loadMgr) { + public GridFailoverContextImpl(GridTaskSessionImpl taskSes, + ComputeJobResult jobRes, + GridLoadBalancerManager loadMgr, + @Nullable Object affKey, + @Nullable String affCacheName) { assert taskSes != null; assert jobRes != null; assert loadMgr != null; @@ -57,6 +69,8 @@ public class GridFailoverContextImpl implements FailoverContext { this.taskSes = taskSes; this.jobRes = jobRes; this.loadMgr = loadMgr; + this.affKey = affKey; + this.affCacheName = affCacheName; } /** {@inheritDoc} */ @@ -75,6 +89,16 @@ public class GridFailoverContextImpl implements FailoverContext { } /** {@inheritDoc} */ + @Nullable @Override public Object affinityKey() { + return affKey; + } + + /** {@inheritDoc} */ + @Nullable @Override public String affinityCacheName() { + return affCacheName; + } + + /** {@inheritDoc} */ @Override public String toString() { return S.toString(GridFailoverContextImpl.class, this); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverManager.java index 714cccb..dffc965 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/failover/GridFailoverManager.java @@ -23,6 +23,7 @@ import org.apache.ignite.compute.*; import org.apache.ignite.internal.*; import org.apache.ignite.internal.managers.*; import org.apache.ignite.spi.failover.*; +import org.jetbrains.annotations.*; import java.util.*; @@ -56,11 +57,17 @@ public class GridFailoverManager extends GridManagerAdapter<FailoverSpi> { /** * @param taskSes Task session. * @param jobRes Job result. - * @param top Collection of all top nodes that does not include the failed node. + * @param top Collection of all topology nodes. + * @param affKey Affinity key. + * @param affCacheName Affinity cache name. * @return New node to route this job to. */ - public ClusterNode failover(GridTaskSessionImpl taskSes, ComputeJobResult jobRes, List<ClusterNode> top) { + public ClusterNode failover(GridTaskSessionImpl taskSes, + ComputeJobResult jobRes, + List<ClusterNode> top, + @Nullable Object affKey, + @Nullable String affCacheName) { return getSpi(taskSes.getFailoverSpi()).failover(new GridFailoverContextImpl(taskSes, jobRes, - ctx.loadBalancing()), top); + ctx.loadBalancing(), affKey, affCacheName), top); } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/AffinityTask.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/AffinityTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/AffinityTask.java new file mode 100644 index 0000000..1b32444 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/AffinityTask.java @@ -0,0 +1,35 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.closure; + +import org.jetbrains.annotations.*; + +/** + * Affinity mapped task. + */ +public interface AffinityTask { + /** + * @return Affinity key. + */ + public Object affinityKey(); + + /** + * @return Affinity cache name. + */ + @Nullable public String affinityCacheName(); +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java index 658557e..21bfc11 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/closure/GridClosureProcessor.java @@ -413,9 +413,12 @@ public class GridClosureProcessor extends GridProcessorAdapter { final ClusterNode node = ctx.affinity().mapKeyToNode(cacheName, affKey0); + if (node == null) + return ComputeTaskInternalFuture.finishedFuture(ctx, T5.class, U.emptyTopologyException()); + ctx.task().setThreadContext(TC_SUBGRID, nodes); - return ctx.task().execute(new T5(node, job), null, false); + return ctx.task().execute(new T5(node, job, affKey0, cacheName), null, false); } catch (IgniteCheckedException e) { return ComputeTaskInternalFuture.finishedFuture(ctx, T5.class, e); @@ -445,9 +448,12 @@ public class GridClosureProcessor extends GridProcessorAdapter { final ClusterNode node = ctx.affinity().mapKeyToNode(cacheName, affKey0); + if (node == null) + return ComputeTaskInternalFuture.finishedFuture(ctx, T4.class, U.emptyTopologyException()); + ctx.task().setThreadContext(TC_SUBGRID, nodes); - return ctx.task().execute(new T4(node, job), null, false); + return ctx.task().execute(new T4(node, job, affKey0, cacheName), null, false); } catch (IgniteCheckedException e) { return ComputeTaskInternalFuture.finishedFuture(ctx, T4.class, e); @@ -1223,7 +1229,7 @@ public class GridClosureProcessor extends GridProcessorAdapter { /** */ - private static class T4 extends TaskNoReduceAdapter<Void> implements GridNoImplicitInjection { + private static class T4 extends TaskNoReduceAdapter<Void> implements GridNoImplicitInjection, AffinityTask { /** */ private static final long serialVersionUID = 0L; @@ -1233,15 +1239,27 @@ public class GridClosureProcessor extends GridProcessorAdapter { /** */ private Runnable job; + /** */ + private Object affKey; + + /** */ + private String affCacheName; + /** * @param node Cluster node. * @param job Job. + * @param affKey Affinity key. + * @param affCacheName Affinity cache name. */ - private T4(ClusterNode node, Runnable job) { + private T4(ClusterNode node, Runnable job, Object affKey, String affCacheName) { super(U.peerDeployAware0(job)); + assert affKey != null; + this.node = node; this.job = job; + this.affKey = affKey; + this.affCacheName = affCacheName; } /** {@inheritDoc} */ @@ -1250,11 +1268,22 @@ public class GridClosureProcessor extends GridProcessorAdapter { return Collections.singletonMap(job, node); } + + /** {@inheritDoc} */ + @Override public Object affinityKey() { + return affKey; + } + + /** {@inheritDoc} */ + @Nullable @Override public String affinityCacheName() { + return affCacheName; + } } /** */ - private static class T5<R> extends GridPeerDeployAwareTaskAdapter<Void, R> implements GridNoImplicitInjection { + private static class T5<R> extends GridPeerDeployAwareTaskAdapter<Void, R> implements + GridNoImplicitInjection, AffinityTask { /** */ private static final long serialVersionUID = 0L; @@ -1264,15 +1293,27 @@ public class GridClosureProcessor extends GridProcessorAdapter { /** */ private Callable<R> job; + /** */ + private Object affKey; + + /** */ + private String affCacheName; + /** * @param node Cluster node. * @param job Job. + * @param affKey Affinity key. + * @param affCacheName Affinity cache name. */ - private T5(ClusterNode node, Callable<R> job) { + private T5(ClusterNode node, Callable<R> job, Object affKey, String affCacheName) { super(U.peerDeployAware0(job)); + assert affKey != null; + this.node = node; this.job = job; + this.affKey = affKey; + this.affCacheName = affCacheName; } /** {@inheritDoc} */ @@ -1291,6 +1332,16 @@ public class GridClosureProcessor extends GridProcessorAdapter { throw new IgniteException("Failed to find successful job result: " + res); } + + /** {@inheritDoc} */ + @Override public Object affinityKey() { + return affKey; + } + + /** {@inheritDoc} */ + @Nullable @Override public String affinityCacheName() { + return affCacheName; + } } /** http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java index 133a31f..f241bcc 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskWorker.java @@ -26,6 +26,7 @@ import org.apache.ignite.internal.*; import org.apache.ignite.internal.cluster.*; import org.apache.ignite.internal.compute.*; import org.apache.ignite.internal.managers.deployment.*; +import org.apache.ignite.internal.processors.closure.*; import org.apache.ignite.internal.processors.timeout.*; import org.apache.ignite.internal.util.typedef.*; import org.apache.ignite.internal.util.typedef.internal.*; @@ -136,6 +137,12 @@ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject { private final boolean noFailover; /** */ + private final Object affKey; + + /** */ + private final String affCache; + + /** */ private final UUID subjId; /** Continuous mapper. */ @@ -245,6 +252,17 @@ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject { Boolean noFailover = getThreadContext(TC_NO_FAILOVER); this.noFailover = noFailover != null ? noFailover : false; + + if (task instanceof AffinityTask) { + AffinityTask affTask = (AffinityTask)task; + + affKey = affTask.affinityKey(); + affCache = affTask.affinityCacheName(); + } + else { + affKey = null; + affCache = null; + } } /** @@ -397,7 +415,9 @@ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject { ses.setClassLoader(dep.classLoader()); - final List<ClusterNode> shuffledNodes = getTaskTopology(); + // Nodes are ignored by affinity tasks. + final List<ClusterNode> shuffledNodes = + affKey == null ? getTaskTopology() : Collections.<ClusterNode>emptyList(); // Load balancer. ComputeLoadBalancer balancer = ctx.loadBalancing().getLoadBalancer(ses, shuffledNodes); @@ -968,7 +988,7 @@ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject { ctx.resource().invokeAnnotated(dep, jobRes.getJob(), ComputeJobBeforeFailover.class); // Map to a new node. - ClusterNode node = ctx.failover().failover(ses, jobRes, new ArrayList<>(top)); + ClusterNode node = ctx.failover().failover(ses, jobRes, new ArrayList<>(top), affKey, affCache); if (node == null) { String msg = "Failed to failover a job to another node (failover SPI returned null) [job=" + http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java index b0cae92..865f1a2 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/FailoverContext.java @@ -20,6 +20,8 @@ package org.apache.ignite.spi.failover; import org.apache.ignite.*; import org.apache.ignite.cluster.*; import org.apache.ignite.compute.*; +import org.apache.ignite.lang.*; +import org.jetbrains.annotations.*; import java.util.*; @@ -52,4 +54,20 @@ public interface FailoverContext { * @throws IgniteException If anything failed. */ public ClusterNode getBalancedNode(List<ClusterNode> top) throws IgniteException; + + /** + * Gets affinity key for {@link IgniteCompute#affinityRun(String, Object, IgniteRunnable)} + * and {@link IgniteCompute#affinityCall(String, Object, IgniteCallable)}. + * + * @return Affinity key. + */ + @Nullable public Object affinityKey(); + + /** + * Returns affinity cache name {@link IgniteCompute#affinityRun(String, Object, IgniteRunnable)} + * and {@link IgniteCompute#affinityCall(String, Object, IgniteCallable)}. + * + * @return Cache name. + */ + @Nullable public String affinityCacheName(); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java index e075d3e..e925995 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/failover/always/AlwaysFailoverSpi.java @@ -92,6 +92,11 @@ public class AlwaysFailoverSpi extends IgniteSpiAdapter implements FailoverSpi, */ public static final String FAILED_NODE_LIST_ATTR = "gg:failover:failednodelist"; + /** + * Name of job context attribute containing number of affinity call attempts. + */ + public static final String AFFINITY_CALL_ATTEMPT = "ignite:failover:affinitycallattempt"; + /** Maximum attempts attribute key should be the same on all nodes. */ public static final String MAX_FAILOVER_ATTEMPT_ATTR = "gg:failover:maxattempts"; @@ -173,6 +178,26 @@ public class AlwaysFailoverSpi extends IgniteSpiAdapter implements FailoverSpi, return null; } + if (ctx.affinityKey() != null) { + Integer affCallAttempt = ctx.getJobResult().getJobContext().getAttribute(AFFINITY_CALL_ATTEMPT); + + if (affCallAttempt == null) + affCallAttempt = 1; + + if (maxFailoverAttempts <= affCallAttempt) { + U.warn(log, "Job failover failed because number of maximum failover attempts for affinity call" + + " is exceeded [failedJob=" + ctx.getJobResult().getJob() + ", maxFailoverAttempts=" + + maxFailoverAttempts + ']'); + + return null; + } + else { + ctx.getJobResult().getJobContext().setAttribute(AFFINITY_CALL_ATTEMPT, affCallAttempt + 1); + + return ignite.affinity(ctx.affinityCacheName()).mapKeyToNode(ctx.affinityKey()); + } + } + Collection<UUID> failedNodes = ctx.getJobResult().getJobContext().getAttribute(FAILED_NODE_LIST_ATTR); if (failedNodes == null) http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheAffinityCallSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheAffinityCallSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheAffinityCallSelfTest.java new file mode 100644 index 0000000..c4436ca --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheAffinityCallSelfTest.java @@ -0,0 +1,172 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.cache; + +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.internal.util.typedef.internal.*; +import org.apache.ignite.lang.*; +import org.apache.ignite.resources.*; +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.spi.failover.always.*; +import org.apache.ignite.testframework.*; +import org.apache.ignite.testframework.junits.common.*; + +import java.util.concurrent.*; + +import static org.apache.ignite.cache.CacheMode.*; + +/** + * Test for {@link IgniteCompute#affinityCall(String, Object, IgniteCallable)} and + * {@link IgniteCompute#affinityRun(String, Object, IgniteRunnable)}. + */ +public class CacheAffinityCallSelfTest extends GridCommonAbstractTest { + /** */ + private static final String CACHE_NAME = "myCache"; + + /** */ + private static final int MAX_FAILOVER_ATTEMPTS = 5; + + /** */ + private static final int SERVERS_COUNT = 4; + + /** */ + private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true); + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); + + TcpDiscoverySpi spi = new TcpDiscoverySpi(); + + spi.setIpFinder(IP_FINDER); + + cfg.setDiscoverySpi(spi); + + AlwaysFailoverSpi failSpi = new AlwaysFailoverSpi(); + failSpi.setMaximumFailoverAttempts(MAX_FAILOVER_ATTEMPTS); + cfg.setFailoverSpi(failSpi); + + CacheConfiguration ccfg = defaultCacheConfiguration(); + ccfg.setName(CACHE_NAME); + ccfg.setCacheMode(PARTITIONED); + ccfg.setBackups(1); + + cfg.setCacheConfiguration(ccfg); + + if (gridName.equals(getTestGridName(SERVERS_COUNT))) + cfg.setClientMode(true); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + } + + /** + * @throws Exception If failed. + */ + public void testAffinityCallRestartNode() throws Exception { + startGrids(4); + + Integer key = primaryKey(grid(0).cache(CACHE_NAME)); + + IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() { + @Override public Object call() throws Exception { + U.sleep(500); + stopGrid(0); + + return null; + } + }); + + while (!fut.isDone()) + grid(1).compute().affinityCall(CACHE_NAME, key, new CheckCallable(key)); + + stopAllGrids(); + } + + /** + * @throws Exception If failed. + */ + public void testAffinityCallNoServerNode() throws Exception { + startGrids(SERVERS_COUNT + 1); + + final Integer key = 1; + + final Ignite client = grid(SERVERS_COUNT); + + final IgniteInternalFuture<Object> fut = GridTestUtils.runAsync(new Callable<Object>() { + @Override public Object call() throws Exception { + for (int i = 0; i < SERVERS_COUNT; ++i) + stopGrid(i); + + return null; + } + }); + + try { + while (!fut.isDone()) + client.compute().affinityCall(CACHE_NAME, key, new CheckCallable(key)); + } + catch (ComputeTaskCancelledException e) { + assertTrue(e.getMessage().contains("stopping")); + } + catch(ClusterGroupEmptyException e) { + assertTrue(e.getMessage().contains("Topology projection is empty")); + } + catch(IgniteException e) { + assertTrue(e.getMessage().contains("cache (or node) is stopping")); + } + + stopGrid(SERVERS_COUNT); + } + + /** + * Test callable. + */ + public static class CheckCallable implements IgniteCallable<Object> { + /** Key. */ + private final Object key; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** + * @param key Key. + */ + public CheckCallable(Object key) { + this.key = key; + } + + /** {@inheritDoc} */ + @Override public Object call() throws IgniteCheckedException { + assert ignite.cluster().localNode().id().equals(ignite.cluster().mapKeyToNode(CACHE_NAME, key).id()); + + return null; + } + } +} http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityRoutingSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityRoutingSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityRoutingSelfTest.java index 78ecf08..a56ab9f 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityRoutingSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAffinityRoutingSelfTest.java @@ -19,17 +19,21 @@ package org.apache.ignite.internal.processors.cache; import org.apache.ignite.*; 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.util.typedef.*; import org.apache.ignite.lang.*; -import org.apache.ignite.marshaller.optimized.*; import org.apache.ignite.resources.*; 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.spi.failover.always.*; +import org.apache.ignite.testframework.*; import org.apache.ignite.testframework.junits.common.*; +import java.util.concurrent.*; + import static org.apache.ignite.cache.CacheMode.*; import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*; @@ -47,6 +51,9 @@ public class GridCacheAffinityRoutingSelfTest extends GridCommonAbstractTest { private static final int KEY_CNT = 50; /** */ + private static final int MAX_FAILOVER_ATTEMPTS = 5; + + /** */ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true); /** @@ -66,6 +73,10 @@ public class GridCacheAffinityRoutingSelfTest extends GridCommonAbstractTest { cfg.setDiscoverySpi(spi); + AlwaysFailoverSpi failSpi = new AlwaysFailoverSpi(); + failSpi.setMaximumFailoverAttempts(MAX_FAILOVER_ATTEMPTS); + cfg.setFailoverSpi(failSpi); + if (!gridName.equals(getTestGridName(GRID_CNT))) { // Default cache configuration. CacheConfiguration dfltCacheCfg = defaultCacheConfiguration(); @@ -129,6 +140,48 @@ public class GridCacheAffinityRoutingSelfTest extends GridCommonAbstractTest { } /** + * @throws Exception If failed. + */ + public void testAffinityCallRestartFails() throws Exception { + GridTestUtils.assertThrows(log, new Callable<Object>() { + @Override public Object call() throws Exception { + grid(0).compute().affinityCall(NON_DFLT_CACHE_NAME, "key", + new FailedCallable("key", MAX_FAILOVER_ATTEMPTS + 1)); + return null; + } + }, ClusterTopologyException.class, "Failed to failover a job to another node"); + } + + /** + * @throws Exception If failed. + */ + public void testAffinityCallRestart() throws Exception { + assertEquals(MAX_FAILOVER_ATTEMPTS, + grid(0).compute().affinityCall(NON_DFLT_CACHE_NAME, "key", + new FailedCallable("key", MAX_FAILOVER_ATTEMPTS))); + } + + /** + * @throws Exception If failed. + */ + public void testAffinityRunRestartFails() throws Exception { + GridTestUtils.assertThrows(log, new Callable<Object>() { + @Override public Object call() throws Exception { + grid(0).compute().affinityRun(NON_DFLT_CACHE_NAME, "key", + new FailedRunnable("key", MAX_FAILOVER_ATTEMPTS + 1)); + return null; + } + }, ClusterTopologyException.class, "Failed to failover a job to another node"); + } + + /** + * @throws Exception If failed. + */ + public void testAffinityRunRestart() throws Exception { + grid(0).compute().affinityRun(NON_DFLT_CACHE_NAME, "key", new FailedRunnable("key", MAX_FAILOVER_ATTEMPTS)); + } + + /** * JUnit. * * @throws Exception If failed. @@ -224,6 +277,108 @@ public class GridCacheAffinityRoutingSelfTest extends GridCommonAbstractTest { } /** + * Test runnable. + */ + private static class FailedCallable implements IgniteCallable<Object> { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + private static final String ATTR_ATTEMPT = "Attempt"; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** */ + @JobContextResource + private ComputeJobContext jobCtx; + + /** Key. */ + private final Object key; + + /** Call attempts. */ + private final Integer callAttempt; + + /** + * @param key Key. + * @param callAttempt Call attempts. + */ + public FailedCallable(Object key, Integer callAttempt) { + this.key = key; + this.callAttempt = callAttempt; + } + + /** {@inheritDoc} */ + @Override public Object call() throws IgniteCheckedException { + Integer attempt = jobCtx.getAttribute(ATTR_ATTEMPT); + + if (attempt == null) + attempt = 1; + + assertEquals(ignite.affinity(NON_DFLT_CACHE_NAME).mapKeyToNode(key), ignite.cluster().localNode()); + + jobCtx.setAttribute(ATTR_ATTEMPT, attempt + 1); + + if (attempt < callAttempt) + throw new ComputeJobFailoverException("Failover exception."); + else + return attempt; + } + } + + /** + * Test runnable. + */ + private static class FailedRunnable implements IgniteRunnable { + /** */ + private static final long serialVersionUID = 0L; + + /** */ + private static final String ATTR_ATTEMPT = "Attempt"; + + /** */ + @IgniteInstanceResource + private Ignite ignite; + + /** */ + @JobContextResource + private ComputeJobContext jobCtx; + + /** Key. */ + private final Object key; + + /** Call attempts. */ + private final Integer callAttempt; + + /** + * @param key Key. + * @param callAttempt Call attempts. + */ + public FailedRunnable(Object key, Integer callAttempt) { + this.key = key; + this.callAttempt = callAttempt; + } + + /** {@inheritDoc} */ + @Override public void run() { + Integer attempt = jobCtx.getAttribute(ATTR_ATTEMPT); + + if (attempt == null) + attempt = 1; + + assertEquals(ignite.affinity(NON_DFLT_CACHE_NAME).mapKeyToNode(key), ignite.cluster().localNode()); + + jobCtx.setAttribute(ATTR_ATTEMPT, attempt + 1); + + if (attempt < callAttempt) + throw new ComputeJobFailoverException("Failover exception."); + else + assertEquals(callAttempt, attempt); + } + } + + /** * Test callable. */ private static class CheckCallable implements IgniteCallable<Object> { http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/test/java/org/apache/ignite/spi/failover/GridFailoverTestContext.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/spi/failover/GridFailoverTestContext.java b/modules/core/src/test/java/org/apache/ignite/spi/failover/GridFailoverTestContext.java index db64475..bfca83d 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/failover/GridFailoverTestContext.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/failover/GridFailoverTestContext.java @@ -66,4 +66,14 @@ public class GridFailoverTestContext implements FailoverContext { @Override public ClusterNode getBalancedNode(List<ClusterNode> grid) { return grid.get(RAND.nextInt(grid.size())); } + + /** {@inheritDoc} */ + @Override public Object affinityKey() { + return null; + } + + /** {@inheritDoc} */ + @Override public String affinityCacheName() { + return null; + } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f9d2a2ef/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java index 39702a3..bafdfef 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite.java @@ -109,6 +109,7 @@ public class IgniteCacheTestSuite extends TestSuite { // Common tests. suite.addTestSuite(GridCacheConcurrentMapSelfTest.class); suite.addTestSuite(GridCacheAffinityMapperSelfTest.class); + suite.addTestSuite(CacheAffinityCallSelfTest.class); GridTestUtils.addTestIfNeeded(suite, GridCacheAffinityRoutingSelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, GridCacheMvccSelfTest.class, ignoredTests); suite.addTestSuite(GridCacheMvccPartitionedSelfTest.class);
