You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/06/15 08:33:08 UTC
ignite git commit: IGNITE-8657 Simultaneous start of multiple clients
may lead to client start hang when exchange history size is too short - Fixes
#4102.
Repository: ignite
Updated Branches:
refs/heads/master 173e54a85 -> 2b928c702
IGNITE-8657 Simultaneous start of multiple clients may lead to client start hang when exchange history size is too short - Fixes #4102.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/2b928c70
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/2b928c70
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/2b928c70
Branch: refs/heads/master
Commit: 2b928c702b2d6a7c1de8d1b3b0a1e0e65e653f21
Parents: 173e54a
Author: Sergey Chugunov <se...@gmail.com>
Authored: Fri Jun 15 11:31:33 2018 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Fri Jun 15 11:31:33 2018 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/IgniteKernal.java | 8 +-
.../internal/IgniteNeedReconnectException.java | 8 +-
.../GridCachePartitionExchangeManager.java | 30 +++-
.../GridDhtPartitionsExchangeFuture.java | 40 ++++-
.../IgniteCacheClientReconnectTest.java | 177 ++++++++++++++++++-
5 files changed, 244 insertions(+), 19 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/2b928c70/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 0171c31..4be5c4d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -1105,7 +1105,13 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
comp.onKernalStart(active);
}
catch (IgniteNeedReconnectException e) {
- assert ctx.discovery().reconnectSupported();
+ ClusterNode locNode = ctx.discovery().localNode();
+
+ assert CU.clientNode(locNode);
+
+ if (!locNode.isClient())
+ throw new IgniteCheckedException("Client node in forceServerMode " +
+ "is not allowed to reconnect to the cluster and will be stopped.");
if (log.isDebugEnabled())
log.debug("Failed to start node components on node start, will wait for reconnect: " + e);
http://git-wip-us.apache.org/repos/asf/ignite/blob/2b928c70/modules/core/src/main/java/org/apache/ignite/internal/IgniteNeedReconnectException.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNeedReconnectException.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNeedReconnectException.java
index f384950..c26f4da 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNeedReconnectException.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNeedReconnectException.java
@@ -29,12 +29,10 @@ public class IgniteNeedReconnectException extends IgniteCheckedException {
private static final long serialVersionUID = 0L;
/**
- * @param locNode Local node.
+ * @param node Node that should reconnect.
* @param cause Cause.
*/
- public IgniteNeedReconnectException(ClusterNode locNode, @Nullable Throwable cause) {
- super("Local node need try to reconnect [locNodeId=" + locNode.id() + ']', cause);
-
- assert locNode.isClient() : locNode;
+ public IgniteNeedReconnectException(ClusterNode node, @Nullable Throwable cause) {
+ super("Node need try to reconnect [nodeId=" + node.id() + ']', cause);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2b928c70/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 3a99089..38ddaf6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -1558,6 +1558,20 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
if (log.isDebugEnabled())
log.debug("Notifying exchange future about single message: " + exchFut);
+ if (msg.client() && !exchFut.isDone()) {
+ if (exchFut.initialVersion().compareTo(readyAffinityVersion()) <= 0) {
+ U.warn(log, "Client node tries to connect but its exchange " +
+ "info is cleaned up from exchange history." +
+ " Consider increasing 'IGNITE_EXCHANGE_HISTORY_SIZE' property " +
+ "or start clients in smaller batches."
+ );
+
+ exchFut.forceClientReconnect(node, msg);
+
+ return;
+ }
+ }
+
exchFut.onReceiveSingleMessage(node, msg);
}
}
@@ -2648,14 +2662,18 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
throw e;
}
catch (IgniteClientDisconnectedCheckedException | IgniteNeedReconnectException e) {
- assert cctx.discovery().reconnectSupported();
-
- U.warn(log,"Local node failed to complete partition map exchange due to " +
- "network issues, will try to reconnect to cluster", e);
+ if (cctx.discovery().reconnectSupported()) {
+ U.warn(log, "Local node failed to complete partition map exchange due to " +
+ "exception, will try to reconnect to cluster: " + e.getMessage(), e);
- cctx.discovery().reconnect();
+ cctx.discovery().reconnect();
- reconnectNeeded = true;
+ reconnectNeeded = true;
+ }
+ else
+ U.warn(log, "Local node received IgniteClientDisconnectedCheckedException or " +
+ " IgniteNeedReconnectException exception but doesn't support reconnect, stopping node: " +
+ e.getMessage(), e);
return;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/2b928c70/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index 2973194..6e0328d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -2056,6 +2056,31 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
}
/**
+ * Method is called on coordinator in situation when initial ExchangeFuture created on client join event was preempted
+ * from exchange history because of IGNITE_EXCHANGE_HISTORY_SIZE property.
+ *
+ * @param node Client node that should try to reconnect to the cluster.
+ * @param msg Single message received from the client which didn't find original ExchangeFuture.
+ */
+ public void forceClientReconnect(ClusterNode node, GridDhtPartitionsSingleMessage msg) {
+ Exception e = new IgniteNeedReconnectException(node, null);
+
+ changeGlobalStateExceptions.put(node.id(), e);
+
+ onDone(null, e);
+
+ GridDhtPartitionsFullMessage fullMsg = createPartitionsMessage(true, false);
+
+ fullMsg.setErrorsMap(changeGlobalStateExceptions);
+
+ FinishState finishState0 = new FinishState(cctx.localNodeId(),
+ initialVersion(),
+ fullMsg);
+
+ sendAllPartitionsToNode(finishState0, msg, node.id());
+ }
+
+ /**
* Processing of received single message. Actual processing in future may be delayed if init method was not
* completed, see {@link #initDone()}
*
@@ -3153,6 +3178,16 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
return;
}
else {
+ if (!F.isEmpty(msg.getErrorsMap())) {
+ Exception e = msg.getErrorsMap().get(cctx.localNodeId());
+
+ assert e != null : msg.getErrorsMap();
+
+ onDone(e);
+
+ return;
+ }
+
AffinityTopologyVersion resVer = msg.resultTopologyVersion() != null ? msg.resultTopologyVersion() : initialVersion();
if (log.isInfoEnabled()) {
@@ -3775,8 +3810,9 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
* @return {@code True} if local node should try reconnect in case of error.
*/
public boolean reconnectOnError(Throwable e) {
- return X.hasCause(e, IOException.class, IgniteClientDisconnectedCheckedException.class) &&
- cctx.discovery().reconnectSupported();
+ return (e instanceof IgniteNeedReconnectException
+ || X.hasCause(e, IOException.class, IgniteClientDisconnectedCheckedException.class))
+ && cctx.discovery().reconnectSupported();
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/2b928c70/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientReconnectTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientReconnectTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientReconnectTest.java
index 29072e8..4beb31a 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientReconnectTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheClientReconnectTest.java
@@ -24,9 +24,17 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -49,14 +57,23 @@ public class IgniteCacheClientReconnectTest extends GridCommonAbstractTest {
private static final int SRV_CNT = 3;
/** */
+ private static final int CLIENTS_CNT = 3;
+
+ /** */
private static final int CACHES = 10;
/** */
+ private static final int PARTITIONS_CNT = 32;
+
+ /** */
private static final long TEST_TIME = 60_000;
/** */
private boolean client;
+ /** */
+ private boolean forceServerMode;
+
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
@@ -69,19 +86,22 @@ public class IgniteCacheClientReconnectTest extends GridCommonAbstractTest {
CacheConfiguration[] ccfgs = new CacheConfiguration[CACHES];
for (int i = 0; i < CACHES; i++) {
- CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME);
+ CacheConfiguration ccfg = new CacheConfiguration();
ccfg.setCacheMode(PARTITIONED);
ccfg.setAtomicityMode(TRANSACTIONAL);
ccfg.setBackups(1);
ccfg.setName("cache-" + i);
ccfg.setWriteSynchronizationMode(FULL_SYNC);
+ ccfg.setAffinity(new RendezvousAffinityFunction(PARTITIONS_CNT, null));
ccfgs[i] = ccfg;
}
cfg.setCacheConfiguration(ccfgs);
}
+ else
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(forceServerMode);
cfg.setClientMode(client);
@@ -89,19 +109,166 @@ public class IgniteCacheClientReconnectTest extends GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void beforeTestsStarted() throws Exception {
- startGrids(SRV_CNT);
+ @Override protected long getTestTimeout() {
+ return TEST_TIME + 60_000;
}
/** {@inheritDoc} */
- @Override protected long getTestTimeout() {
- return TEST_TIME + 60_000;
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * If setting IGNITE_EXCHANGE_HISTORY_SIZE is set to small value
+ * it is possible that bunch of clients simultaneous start (amount > IGNITE_EXCHANGE_HISTORY_SIZE)
+ * may result in ExchangeFuture for some client being flushed from exchange history.
+ *
+ * In that case client attempts to reconnect until success.
+ *
+ * After that it should get correct information about topology version and affinity distribution.
+ *
+ * @throws Exception If failed
+ */
+ public void testClientReconnectOnExchangeHistoryExhaustion() throws Exception {
+ System.setProperty(IgniteSystemProperties.IGNITE_EXCHANGE_HISTORY_SIZE, "1");
+
+ try {
+ startGrids(SRV_CNT);
+
+ client = true;
+
+ startGridsMultiThreaded(SRV_CNT, CLIENTS_CNT);
+
+ waitForTopology(SRV_CNT + CLIENTS_CNT);
+
+ verifyPartitionToNodeMappings();
+
+ verifyAffinityTopologyVersions();
+
+ verifyCacheOperationsOnClients();
+ }
+ finally {
+ System.clearProperty(IgniteSystemProperties.IGNITE_EXCHANGE_HISTORY_SIZE);
+ }
+ }
+
+ /**
+ * Verifies that in case of exchange history exhaustion
+ * (refer to javadoc at {@link #testClientReconnectOnExchangeHistoryExhaustion()} for more info about it)
+ * clients with forceServerMode=true flag don't try to reconnect to the cluster and stop.
+ *
+ * @throws Exception If failed
+ */
+ public void testClientInForceServerModeStopsOnExchangeHistoryExhaustion() throws Exception {
+ System.setProperty(IgniteSystemProperties.IGNITE_EXCHANGE_HISTORY_SIZE, "1");
+
+ try {
+ startGrids(SRV_CNT);
+
+ client = true;
+
+ forceServerMode = true;
+
+ int clientNodes = 10;
+
+ try {
+ startGridsMultiThreaded(SRV_CNT, clientNodes);
+ }
+ catch (IgniteCheckedException e) {
+ //Ignored: it is expected to get exception here
+ }
+
+ awaitPartitionMapExchange();
+
+ int topSize = G.allGrids().size();
+
+ assertTrue("Actual size: " + topSize, topSize < SRV_CNT + clientNodes);
+
+ }
+ finally {
+ System.clearProperty(IgniteSystemProperties.IGNITE_EXCHANGE_HISTORY_SIZE);
+ }
+ }
+
+ /**
+ * Verifies basic cache operations from all clients.
+ */
+ private void verifyCacheOperationsOnClients() {
+ for (int i = SRV_CNT; i < SRV_CNT + CLIENTS_CNT; i++) {
+ IgniteEx cl = grid(i);
+
+ if (!forceServerMode)
+ assertTrue(cl.localNode().isClient());
+
+ for (int j = 0; j < CACHES; j++) {
+ IgniteCache<Object, Object> cache = cl.cache("cache-" + j);
+
+ String keyPrefix = "cl-" + i + "_key-";
+
+ for (int k = 0; k < 1_000; k++)
+ cache.put(keyPrefix + k, "val-" + k);
+
+ for (int k = 999; k >= 0; k--)
+ assertEquals("val-" + k, cache.get(keyPrefix + k));
+ }
+ }
+ }
+
+ /**
+ * Verifies that affinity mappings are the same on clients and servers.
+ */
+ private void verifyPartitionToNodeMappings() {
+ IgniteEx refSrv = grid(0);
+ String cacheName;
+
+ for (int i = 0; i < CACHES; i++) {
+ cacheName = "cache-" + i;
+
+ Affinity<Object> refAffinity = refSrv.affinity(cacheName);
+
+ for (int j = 0; j < PARTITIONS_CNT; j++) {
+ ClusterNode refAffNode = refAffinity.mapPartitionToNode(j);
+
+ assertNotNull("Affinity node for " + j + " partition is null", refAffNode);
+
+ for (int k = SRV_CNT; k < SRV_CNT + CLIENTS_CNT; k++) {
+ ClusterNode clAffNode = grid(k).affinity(cacheName).mapPartitionToNode(j);
+
+ assertNotNull("Affinity node for " + k + " client and " + j + " partition is null", clAffNode);
+
+ assertEquals("Affinity node for "
+ + k
+ + " client and "
+ + j
+ + " partition is different on client",
+ refAffNode.id(),
+ clAffNode.id());
+ }
+ }
+ }
+ }
+
+ /**
+ * Verifies {@link AffinityTopologyVersion}s: one obtained from coordinator and all from each client node.
+ */
+ private void verifyAffinityTopologyVersions() {
+ IgniteEx srv = grid(0);
+
+ AffinityTopologyVersion srvTopVer = srv.context().discovery().topologyVersionEx();
+
+ for (int i = SRV_CNT; i < SRV_CNT + CLIENTS_CNT; i++) {
+ AffinityTopologyVersion clntTopVer = grid(i).context().discovery().topologyVersionEx();
+
+ assertTrue(clntTopVer.equals(srvTopVer));
+ }
}
/**
* @throws Exception If failed.
*/
public void testClientReconnect() throws Exception {
+ startGrids(SRV_CNT);
+
client = true;
final AtomicBoolean stop = new AtomicBoolean(false);