ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ivasilin...@apache.org
Subject [02/23] incubator-ignite git commit: #ignite-1087: affinity run runs several times on primary node.
Date Mon, 20 Jul 2015 15:05:18 GMT
#ignite-1087: affinity run runs several times on primary node.


Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/355e33db
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/355e33db
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/355e33db

Branch: refs/heads/ignite-1087
Commit: 355e33db0b0551759a7ce976f6af53bc7ad5a36a
Parents: d8f5b6f
Author: ivasilinets <ivasilinets@gridgain.com>
Authored: Fri Jul 3 14:37:55 2015 +0300
Committer: ivasilinets <ivasilinets@gridgain.com>
Committed: Fri Jul 3 14:37:55 2015 +0300

----------------------------------------------------------------------
 .../closure/GridClosureProcessor.java           |  4 ++
 .../task/GridTaskThreadContextKey.java          |  6 +++
 .../processors/task/GridTaskWorker.java         | 16 ++++++
 .../spi/failover/always/AlwaysFailoverSpi.java  | 37 +++++++++++++
 .../cache/GridCacheAffinityRoutingSelfTest.java | 56 +++++++++++++++++++-
 5 files changed, 118 insertions(+), 1 deletion(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/355e33db/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..1eb53ee 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
@@ -414,6 +414,8 @@ public class GridClosureProcessor extends GridProcessorAdapter {
             final ClusterNode node = ctx.affinity().mapKeyToNode(cacheName, affKey0);
 
             ctx.task().setThreadContext(TC_SUBGRID, nodes);
+            ctx.task().setThreadContext(TC_AFFINITY_KEY, affKey0);
+            ctx.task().setThreadContext(TC_AFFINITY_CACHE, cacheName);
 
             return ctx.task().execute(new T5(node, job), null, false);
         }
@@ -446,6 +448,8 @@ public class GridClosureProcessor extends GridProcessorAdapter {
             final ClusterNode node = ctx.affinity().mapKeyToNode(cacheName, affKey0);
 
             ctx.task().setThreadContext(TC_SUBGRID, nodes);
+            ctx.task().setThreadContext(TC_AFFINITY_KEY, affKey0);
+            ctx.task().setThreadContext(TC_AFFINITY_CACHE, cacheName);
 
             return ctx.task().execute(new T4(node, job), null, false);
         }

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/355e33db/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java
index df706cf..0bfac62 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/task/GridTaskThreadContextKey.java
@@ -27,6 +27,12 @@ public enum GridTaskThreadContextKey {
     /** No failover flag. */
     TC_NO_FAILOVER,
 
+    /** Affinity key. */
+    TC_AFFINITY_KEY,
+
+    /** Affinity cache. */
+    TC_AFFINITY_CACHE,
+
     /** Projection for the task. */
     TC_SUBGRID,
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/355e33db/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 eb5fa77..61bb62a 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
@@ -34,6 +34,7 @@ import org.apache.ignite.internal.visor.util.*;
 import org.apache.ignite.lang.*;
 import org.apache.ignite.marshaller.*;
 import org.apache.ignite.resources.*;
+import org.apache.ignite.spi.failover.always.*;
 import org.jetbrains.annotations.*;
 import org.jsr166.*;
 
@@ -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,9 @@ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject
{
         Boolean noFailover = getThreadContext(TC_NO_FAILOVER);
 
         this.noFailover = noFailover != null ? noFailover : false;
+
+        this.affKey = getThreadContext(TC_AFFINITY_KEY);
+        this.affCache = getThreadContext(TC_AFFINITY_CACHE);
     }
 
     /**
@@ -559,6 +569,9 @@ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject
{
 
         Collection<? extends ClusterNode> subgrid = top != null ? ctx.discovery().nodes(top)
: ctx.discovery().allNodes();
 
+        if (affKey != null)
+            return Collections.singletonList(ctx.affinity().mapKeyToNode(affCache, affKey));
+
         int size = subgrid.size();
 
         if (size == 0)
@@ -769,6 +782,9 @@ class GridTaskWorker<T, R> extends GridWorker implements GridTimeoutObject
{
                         }
 
                         case FAILOVER: {
+                            if (jobRes != null && affKey != null)
+                                jobRes.getJobContext().setAttribute(AlwaysFailoverSpi.AFFINITY_CALL_FLAG,
true);
+
                             if (!failover(res, jobRes, getTaskTopology()))
                                 plc = null;
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/355e33db/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..c5c0f18 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,21 @@ public class AlwaysFailoverSpi extends IgniteSpiAdapter implements FailoverSpi,
      */
     public static final String FAILED_NODE_LIST_ATTR = "gg:failover:failednodelist";
 
+    /**
+     * Name of job context attribute containing affinity call flag for affinity call.
+     *
+     * @see org.apache.ignite.compute.ComputeJobContext
+     */
+    public static final String AFFINITY_CALL_ATTEMPT = "ignite:failover:affinitycallattempt";
+
+    /**
+     * Name of job context attribute containing affinity call flag for affinity call.
+     *
+     * @see org.apache.ignite.compute.ComputeJobContext
+     */
+    public static final String AFFINITY_CALL_FLAG = "ignite:failover:affinitycall";
+
+
     /** Maximum attempts attribute key should be the same on all nodes. */
     public static final String MAX_FAILOVER_ATTEMPT_ATTR = "gg:failover:maxattempts";
 
@@ -175,6 +190,28 @@ public class AlwaysFailoverSpi extends IgniteSpiAdapter implements FailoverSpi,
 
         Collection<UUID> failedNodes = ctx.getJobResult().getJobContext().getAttribute(FAILED_NODE_LIST_ATTR);
 
+        Boolean affCall = ctx.getJobResult().getJobContext().getAttribute(AFFINITY_CALL_FLAG);
+
+        if (affCall != null && affCall) {
+            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 top.get(0);
+            }
+        }
+
         if (failedNodes == null)
             failedNodes = U.newHashSet(1);
 

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/355e33db/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..38de9a1 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,20 @@ 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.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.*;
 
@@ -129,6 +132,18 @@ public class GridCacheAffinityRoutingSelfTest extends GridCommonAbstractTest
{
     }
 
     /**
+     * @throws Exception If failed.
+     */
+    public void testAffinityRunRestart() 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"));
+                return null;
+            }
+        }, ClusterTopologyException.class, "Failed to failover a job to another node");
+    }
+
+    /**
      * JUnit.
      *
      * @throws Exception If failed.
@@ -224,6 +239,45 @@ public class GridCacheAffinityRoutingSelfTest extends GridCommonAbstractTest
{
     }
 
     /**
+     * Test runnable.
+     */
+    private static class FailedRunnable extends CAX {
+        /** */
+        @IgniteInstanceResource
+        private Ignite ignite;
+
+        /** */
+        @JobContextResource
+        private ComputeJobContext jobCtx;
+
+        /** Key. */
+        private final Object key;
+
+        /**
+         * @param key Key.
+         */
+        public FailedRunnable(Object key) {
+            this.key = key;
+        }
+
+        /** {@inheritDoc} */
+        @Override public void applyx() throws IgniteCheckedException {
+            Integer attempt = jobCtx.getAttribute("Attempt");
+
+            if (attempt == null)
+                attempt = 0;
+
+            System.out.println("Attempt=" + attempt);
+
+            assertEquals(ignite.affinity(NON_DFLT_CACHE_NAME).mapKeyToNode(key), ignite.cluster().localNode());
+
+            jobCtx.setAttribute("Attempt", attempt + 1);
+
+            throw new ComputeJobFailoverException("Failover exception.");
+        }
+    }
+
+    /**
      * Test callable.
      */
     private static class CheckCallable implements IgniteCallable<Object> {


Mime
View raw message