Repository: ignite Updated Branches: refs/heads/ignite-1758-debug ca7eebde9 -> 1a1cb8ca6
ignite-1758 Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/1a1cb8ca Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/1a1cb8ca Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/1a1cb8ca Branch: refs/heads/ignite-1758-debug Commit: 1a1cb8ca607af77e7d96a51497d47a3c42c1f4a3 Parents: ca7eebd Author: sboikov <[email protected]> Authored: Mon Nov 16 18:42:58 2015 +0300 Committer: sboikov <[email protected]> Committed: Mon Nov 16 18:42:58 2015 +0300 ---------------------------------------------------------------------- .../apache/ignite/spi/discovery/tcp/ServerImpl.java | 15 ++++++++++++++- .../tcp/internal/TcpDiscoveryNodesRing.java | 16 ++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/1a1cb8ca/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java index 9a2c849..2356404 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java @@ -159,6 +159,10 @@ class ServerImpl extends TcpDiscoveryImpl { private static final int ENSURED_MSG_HIST_SIZE = getInteger(IGNITE_DISCOVERY_HISTORY_SIZE, 1024 * 10); /** */ + private static final IgniteProductVersion CUSTOM_MSG_ALLOW_JOINING_FOR_VERIFIED_SINCE = + IgniteProductVersion.fromString("1.5.0"); + + /** */ private final ThreadPoolExecutor utilityPool = new ThreadPoolExecutor(0, 1, 2000, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>()); @@ -4466,7 +4470,16 @@ class ServerImpl extends TcpDiscoveryImpl { */ private void processCustomMessage(TcpDiscoveryCustomEventMessage msg) { if (isLocalNodeCoordinator()) { - if (msg.topologyVersion() == 0L && !joiningNodes.isEmpty()) { + boolean delayMsg; + + assert ring.minimumNodeVersion() != null : ring; + + if (ring.minimumNodeVersion().compareTo(CUSTOM_MSG_ALLOW_JOINING_FOR_VERIFIED_SINCE) >= 0) + delayMsg = msg.topologyVersion() == 0L && !joiningNodes.isEmpty(); + else + delayMsg = !joiningNodes.isEmpty(); + + if (delayMsg) { if (log.isDebugEnabled()) { log.debug("Delay custom message processing, there are joining nodes [msg=" + msg + ", joiningNodes=" + joiningNodes + ']'); http://git-wip-us.apache.org/repos/asf/ignite/blob/1a1cb8ca/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java index f184fbd..8e6274f 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java @@ -26,6 +26,7 @@ import org.apache.ignite.internal.util.typedef.PN; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.internal.util.typedef.internal.U; import org.apache.ignite.lang.IgnitePredicate; +import org.apache.ignite.lang.IgniteProductVersion; import org.jetbrains.annotations.Nullable; import java.util.Collection; @@ -88,6 +89,16 @@ public class TcpDiscoveryNodesRing { @GridToStringExclude private final ReadWriteLock rwLock = new ReentrantReadWriteLock(); + /** */ + private IgniteProductVersion minNodeVer; + + /** + * @return Minimum node version. + */ + public IgniteProductVersion minimumNodeVersion() { + return minNodeVer; + } + /** * Sets local node. * @@ -104,6 +115,8 @@ public class TcpDiscoveryNodesRing { clear(); maxInternalOrder = locNode.internalOrder(); + + minNodeVer = locNode.version(); } finally { rwLock.writeLock().unlock(); @@ -225,6 +238,9 @@ public class TcpDiscoveryNodesRing { nodeOrder = node.internalOrder(); maxInternalOrder = node.internalOrder(); + + if (node.version().compareTo(minNodeVer) < 0) + minNodeVer = node.version(); } finally { rwLock.writeLock().unlock();
