ignite-1.5 Fixed client discovery impl to skip discovery message processing while disconnected.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/9ed39096 Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/9ed39096 Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/9ed39096 Branch: refs/heads/ignite-1537 Commit: 9ed39096100cdd9a937e33ffcbe8b46784616ff9 Parents: a8beeae Author: sboikov <[email protected]> Authored: Fri Dec 18 14:11:02 2015 +0300 Committer: sboikov <[email protected]> Committed: Fri Dec 18 14:11:02 2015 +0300 ---------------------------------------------------------------------- .../discovery/GridDiscoveryManager.java | 4 +- .../ignite/spi/discovery/tcp/ClientImpl.java | 48 +++++++++++--------- .../IgniteClientReconnectCacheTest.java | 26 +++++++++-- 3 files changed, 52 insertions(+), 26 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/ignite/blob/9ed39096/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java index 92d66d7..72a2bef 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java @@ -1641,7 +1641,9 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> { if (cache == null) { throw new IgniteException("Failed to resolve nodes topology [cacheName=" + cacheName + - ", topVer=" + topVer + ", history=" + discoCacheHist.keySet() + + ", topVer=" + topVer + + ", history=" + discoCacheHist.keySet() + + ", snap=" + snap + ", locNode=" + ctx.discovery().localNode() + ']'); } http://git-wip-us.apache.org/repos/asf/ignite/blob/9ed39096/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java index 8f6c8a9..850cc24 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java @@ -1828,36 +1828,42 @@ class ClientImpl extends TcpDiscoveryImpl { return; } - if (!getLocalNodeId().equals(msg.creatorNodeId())) { - TcpDiscoveryNode node = rmtNodes.remove(msg.failedNodeId()); + if (nodeAdded()) { + if (!getLocalNodeId().equals(msg.creatorNodeId())) { + TcpDiscoveryNode node = rmtNodes.remove(msg.failedNodeId()); - if (node == null) { - if (log.isDebugEnabled()) - log.debug("Discarding node failed message since node is not found [msg=" + msg + ']'); + if (node == null) { + if (log.isDebugEnabled()) + log.debug("Discarding node failed message since node is not found [msg=" + msg + ']'); - return; - } + return; + } - Collection<ClusterNode> top = updateTopologyHistory(msg.topologyVersion(), msg); + Collection<ClusterNode> top = updateTopologyHistory(msg.topologyVersion(), msg); - if (state != CONNECTED) { - if (log.isDebugEnabled()) - log.debug("Discarding node failed message (join process is not finished): " + msg); + if (state != CONNECTED) { + if (log.isDebugEnabled()) + log.debug("Discarding node failed message (join process is not finished): " + msg); - return; - } + return; + } - if (msg.warning() != null) { - ClusterNode creatorNode = rmtNodes.get(msg.creatorNodeId()); + if (msg.warning() != null) { + ClusterNode creatorNode = rmtNodes.get(msg.creatorNodeId()); - U.warn(log, "Received EVT_NODE_FAILED event with warning [" + - "nodeInitiatedEvt=" + (creatorNode != null ? creatorNode : msg.creatorNodeId()) + - ", msg=" + msg.warning() + ']'); - } + U.warn(log, "Received EVT_NODE_FAILED event with warning [" + + "nodeInitiatedEvt=" + (creatorNode != null ? creatorNode : msg.creatorNodeId()) + + ", msg=" + msg.warning() + ']'); + } - notifyDiscovery(EVT_NODE_FAILED, msg.topologyVersion(), node, top); + notifyDiscovery(EVT_NODE_FAILED, msg.topologyVersion(), node, top); - spi.stats.onNodeFailed(); + spi.stats.onNodeFailed(); + } + } + else { + if (log.isDebugEnabled()) + log.debug("Ignore topology message, local node not added to topology: " + msg); } } http://git-wip-us.apache.org/repos/asf/ignite/blob/9ed39096/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java index 170d8ae..ed38d2b 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectCacheTest.java @@ -1087,7 +1087,7 @@ public class IgniteClientReconnectCacheTest extends IgniteClientReconnectAbstrac clientMode = true; - final int CLIENTS = 2; + final int CLIENTS = 5; List<Ignite> clients = new ArrayList<>(); @@ -1102,12 +1102,14 @@ public class IgniteClientReconnectCacheTest extends IgniteClientReconnectAbstrac int nodes = SRV_CNT + CLIENTS; int srvNodes = SRV_CNT; - for (int iter = 0; iter < 3; iter++) { + for (int iter = 0; iter < 5; iter++) { log.info("Iteration: " + iter); reconnectClientNodes(log, clients, grid(0), null); - for (Ignite client : clients) { + final int expNodes = CLIENTS + srvNodes; + + for (final Ignite client : clients) { IgniteCache<Object, Object> cache = client.cache(null); assertNotNull(cache); @@ -1116,6 +1118,14 @@ public class IgniteClientReconnectCacheTest extends IgniteClientReconnectAbstrac assertEquals(1, cache.get(client.name())); + GridTestUtils.waitForCondition(new GridAbsPredicate() { + @Override public boolean apply() { + ClusterGroup grp = client.cluster().forCacheNodes(null); + + return grp.nodes().size() == expNodes; + } + }, 5000); + ClusterGroup grp = client.cluster().forCacheNodes(null); assertEquals(CLIENTS + srvNodes, grp.nodes().size()); @@ -1126,7 +1136,15 @@ public class IgniteClientReconnectCacheTest extends IgniteClientReconnectAbstrac } for (int i = 0; i < nodes; i++) { - Ignite ignite = grid(i); + final Ignite ignite = grid(i); + + GridTestUtils.waitForCondition(new GridAbsPredicate() { + @Override public boolean apply() { + ClusterGroup grp = ignite.cluster().forCacheNodes(null); + + return grp.nodes().size() == expNodes; + } + }, 5000); ClusterGroup grp = ignite.cluster().forCacheNodes(null);
