ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [1/3] incubator-ignite git commit: #ignite-784: fix compatibility suite.
Date Fri, 24 Apr 2015 12:13:13 GMT
Repository: incubator-ignite
Updated Branches:
  refs/heads/ignite-784-1 [created] b5ed4f20f


#ignite-784: fix compatibility suite.


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

Branch: refs/heads/ignite-784-1
Commit: 6bd55a853750625e5bff790eeb5ca001e47ac75f
Parents: 381fd1a
Author: ivasilinets <ivasilinets@gridgain.com>
Authored: Fri Apr 24 14:36:44 2015 +0300
Committer: ivasilinets <ivasilinets@gridgain.com>
Committed: Fri Apr 24 14:36:44 2015 +0300

----------------------------------------------------------------------
 .../datastreamer/DataStreamProcessor.java       | 48 +++++++++++++-------
 1 file changed, 31 insertions(+), 17 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6bd55a85/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
index 9e53bb5..3e6f207 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java
@@ -27,6 +27,7 @@ import org.apache.ignite.internal.util.*;
 import org.apache.ignite.internal.util.typedef.*;
 import org.apache.ignite.internal.util.typedef.internal.*;
 import org.apache.ignite.internal.util.worker.*;
+import org.apache.ignite.lang.*;
 import org.apache.ignite.marshaller.*;
 import org.apache.ignite.stream.*;
 import org.apache.ignite.thread.*;
@@ -36,12 +37,16 @@ import java.util.*;
 import java.util.concurrent.*;
 
 import static org.apache.ignite.internal.GridTopic.*;
+import static org.apache.ignite.internal.IgniteNodeAttributes.*;
 import static org.apache.ignite.internal.managers.communication.GridIoPolicy.*;
 
 /**
  *
  */
 public class DataStreamProcessor<K, V> extends GridProcessorAdapter {
+    /** The lowest version of ignite that is compatible with current version. */
+    private static IgniteProductVersion COMPATIBLE_VERSION_SINCE = IgniteProductVersion.fromString("1.0.4");
+
     /** Loaders map (access is not supposed to be highly concurrent). */
     private Collection<DataStreamerImpl> ldrs = new GridConcurrentHashSet<>();
 
@@ -189,25 +194,34 @@ public class DataStreamProcessor<K, V> extends GridProcessorAdapter
{
             AffinityTopologyVersion locAffVer = ctx.cache().context().exchange().readyAffinityVersion();
             AffinityTopologyVersion rmtAffVer = req.topologyVersion();
 
-            if (locAffVer.compareTo(rmtAffVer) < 0) {
-                if (log.isDebugEnabled())
-                    log.debug("Received request has higher affinity topology version [request="
+ req +
-                        ", locTopVer=" + locAffVer + ", rmtTopVer=" + rmtAffVer + ']');
-
-                IgniteInternalFuture<?> fut = ctx.cache().context().exchange().affinityReadyFuture(rmtAffVer);
+            if (rmtAffVer == null) {
+                IgniteProductVersion rmtVer = ctx.discovery().node(nodeId).version();
 
-                if (fut != null && !fut.isDone()) {
-                    fut.listen(new CI1<IgniteInternalFuture<?>>() {
-                        @Override public void apply(IgniteInternalFuture<?> t) {
-                            ctx.closure().runLocalSafe(new Runnable() {
-                                @Override public void run() {
-                                    processRequest(nodeId, req);
-                                }
-                            }, false);
-                        }
-                    });
+                assert rmtVer.compareTo(COMPATIBLE_VERSION_SINCE) < 0;
+            }
+            else {
+                if (locAffVer.compareTo(rmtAffVer) < 0) {
+                    if (log.isDebugEnabled())
+                        log.debug("Received request has higher affinity topology version
[request=" + req +
+                            ", locTopVer=" + locAffVer + ", rmtTopVer=" + rmtAffVer + ']');
+
+                    IgniteInternalFuture<?> fut = ctx.cache().context().exchange().affinityReadyFuture(rmtAffVer);
+
+                    if (fut != null && !fut.isDone()) {
+                        fut.listen(new CI1<IgniteInternalFuture<?>>() {
+                            @Override
+                            public void apply(IgniteInternalFuture<?> t) {
+                                ctx.closure().runLocalSafe(new Runnable() {
+                                    @Override
+                                    public void run() {
+                                        processRequest(nodeId, req);
+                                    }
+                                }, false);
+                            }
+                        });
 
-                    return;
+                        return;
+                    }
                 }
             }
 


Mime
View raw message