You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by nt...@apache.org on 2015/09/07 22:47:06 UTC
[39/50] [abbrv] ignite git commit: #ignite-1087: AffinityRun runs job
on not primary nodes.
#ignite-1087: AffinityRun runs job on not primary nodes.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/f9d2a2ef
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/f9d2a2ef
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/f9d2a2ef
Branch: refs/heads/ignite-788-dev
Commit: f9d2a2ef4424bf166912594791b6414b9a9e8457
Parents: ede9612
Author: ivasilinets <iv...@gridgain.com>
Authored: Mon Jul 20 11:25:00 2015 +0300
Committer: ivasilinets <iv...@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/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/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/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/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/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/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/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/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/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/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/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/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);