You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/07/17 08:29:01 UTC
[49/50] [abbrv] incubator-ignite git commit: ignite-752: merge from
master
ignite-752: merge from master
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/37198cae
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/37198cae
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/37198cae
Branch: refs/heads/ignite-752
Commit: 37198cae51a7ba54e3bd8c476bad7ecdf65c6e16
Parents: 5fc818c 44043fa
Author: Denis Magda <dm...@gridgain.com>
Authored: Fri Jul 17 09:17:01 2015 +0300
Committer: Denis Magda <dm...@gridgain.com>
Committed: Fri Jul 17 09:17:01 2015 +0300
----------------------------------------------------------------------
DEVNOTES.txt | 8 +
RELEASE_NOTES.txt | 2 +
assembly/LICENSE_FABRIC | 306 +++++
assembly/LICENSE_HADOOP | 259 ++++
assembly/NOTICE_FABRIC | 13 +
assembly/NOTICE_HADOOP | 12 +
assembly/dependencies-fabric.xml | 13 +
assembly/dependencies-hadoop.xml | 12 +
assembly/dependencies-visor-console.xml | 20 +-
assembly/release-base.xml | 10 -
assembly/release-fabric.xml | 12 +
assembly/release-hadoop.xml | 12 +
modules/aop/licenses/aspectj-epl-license.txt | 69 -
modules/apache-license-gen/pom.xml | 48 +
.../src/main/resources/META-INF/licenses.txt.vm | 42 +
modules/core/licenses/jsr166-license.txt | 3 +
modules/core/licenses/snaptree-bsd-license.txt | 2 +-
.../IgniteClientDisconnectedException.java | 61 +
.../java/org/apache/ignite/IgniteCluster.java | 8 +
.../apache/ignite/internal/GridComponent.java | 18 +
.../ignite/internal/GridJobSiblingImpl.java | 2 +-
.../ignite/internal/GridKernalContext.java | 5 +
.../ignite/internal/GridKernalContextImpl.java | 31 +-
.../ignite/internal/GridKernalGateway.java | 46 +-
.../ignite/internal/GridKernalGatewayImpl.java | 85 +-
.../apache/ignite/internal/GridKernalState.java | 3 +
.../ignite/internal/GridPluginComponent.java | 11 +
...gniteClientDisconnectedCheckedException.java | 49 +
.../apache/ignite/internal/IgniteKernal.java | 222 +++-
.../cluster/IgniteClusterAsyncImpl.java | 5 +
.../internal/cluster/IgniteClusterImpl.java | 18 +
.../internal/managers/GridManagerAdapter.java | 19 +-
.../deployment/GridDeploymentCommunication.java | 2 +-
.../deployment/GridDeploymentManager.java | 95 +-
.../discovery/GridDiscoveryManager.java | 188 ++-
.../processors/GridProcessorAdapter.java | 11 +
.../affinity/GridAffinityAssignmentCache.java | 26 +-
.../cache/CacheOsConflictResolutionManager.java | 6 +
.../cache/DynamicCacheChangeBatch.java | 17 +
.../processors/cache/GridCacheAdapter.java | 25 +-
.../cache/GridCacheAffinityManager.java | 21 +-
.../cache/GridCacheConcurrentMap.java | 15 +-
.../processors/cache/GridCacheGateway.java | 116 +-
.../processors/cache/GridCacheIoManager.java | 8 +
.../processors/cache/GridCacheManager.java | 6 +
.../cache/GridCacheManagerAdapter.java | 6 +
.../processors/cache/GridCacheMvccManager.java | 41 +-
.../GridCachePartitionExchangeManager.java | 81 +-
.../processors/cache/GridCachePreloader.java | 5 +
.../cache/GridCachePreloaderAdapter.java | 5 +
.../processors/cache/GridCacheProcessor.java | 311 ++++-
.../cache/GridCacheSharedContext.java | 113 +-
.../cache/GridCacheSharedManager.java | 11 +-
.../cache/GridCacheSharedManagerAdapter.java | 20 +-
.../processors/cache/GridCacheUtils.java | 11 +
.../processors/cache/IgniteCacheFutureImpl.java | 11 +
.../processors/cache/IgniteCacheProxy.java | 57 +-
.../CacheDataStructuresManager.java | 35 +
.../distributed/GridCacheTxFinishSync.java | 46 +
.../distributed/dht/GridDhtCacheAdapter.java | 14 +-
.../dht/GridDhtPartitionTopologyImpl.java | 24 +
.../dht/GridPartitionedGetFuture.java | 13 +-
.../dht/preloader/GridDhtPreloader.java | 16 +-
.../distributed/near/GridNearCacheAdapter.java | 8 +
.../distributed/near/GridNearGetFuture.java | 13 +-
.../cache/dr/GridOsCacheDrManager.java | 7 +-
.../query/GridCacheDistributedQueryManager.java | 22 +
.../cache/query/GridCacheQueryAdapter.java | 11 +-
.../query/GridCacheQueryFutureAdapter.java | 2 +-
.../continuous/CacheContinuousQueryHandler.java | 5 +
.../transactions/IgniteTransactionsImpl.java | 59 +-
.../cache/transactions/IgniteTxManager.java | 19 +-
.../transactions/TransactionProxyImpl.java | 2 +-
.../cache/version/GridCacheVersionManager.java | 9 +-
.../clock/GridClockSyncProcessor.java | 6 +-
.../processors/cluster/ClusterProcessor.java | 11 +
.../continuous/GridContinuousHandler.java | 9 +-
.../continuous/GridContinuousProcessor.java | 127 +-
.../datastreamer/DataStreamProcessor.java | 24 +-
.../datastreamer/DataStreamerImpl.java | 90 +-
.../datastructures/DataStructuresProcessor.java | 33 +-
.../datastructures/GridCacheAtomicLongImpl.java | 33 +-
.../GridCacheAtomicReferenceImpl.java | 34 +-
.../GridCacheAtomicSequenceImpl.java | 33 +-
.../GridCacheAtomicStampedImpl.java | 33 +-
.../GridCacheCountDownLatchImpl.java | 51 +-
.../datastructures/GridCacheRemovable.java | 6 +-
.../datastructures/GridCacheSetImpl.java | 15 +-
.../datastructures/GridCacheSetProxy.java | 47 +-
.../processors/job/GridJobProcessor.java | 2 +-
.../internal/processors/job/GridJobWorker.java | 2 +-
.../processors/query/GridQueryIndexing.java | 7 +
.../processors/query/GridQueryProcessor.java | 10 +-
.../service/GridServiceProcessor.java | 45 +-
.../processors/service/GridServiceProxy.java | 13 +-
.../processors/task/GridTaskProcessor.java | 55 +-
.../processors/task/GridTaskWorker.java | 59 +-
.../ignite/internal/util/IgniteUtils.java | 28 +
.../util/future/GridFutureChainListener.java | 4 -
.../internal/util/future/IgniteFutureImpl.java | 12 +-
.../shmem/IpcSharedMemoryClientEndpoint.java | 5 +-
.../ignite/internal/util/lang/GridFunc.java | 2 +
.../plugin/security/SecuritySubjectType.java | 3 +-
.../java/org/apache/ignite/spi/IgniteSpi.java | 15 +
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 52 +-
.../communication/tcp/TcpCommunicationSpi.java | 357 ++++--
.../spi/discovery/DiscoverySpiDataExchange.java | 3 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 429 +++++--
.../ignite/spi/discovery/tcp/ServerImpl.java | 172 ++-
.../spi/discovery/tcp/TcpDiscoveryImpl.java | 9 +-
.../spi/discovery/tcp/TcpDiscoverySpi.java | 33 +-
.../tcp/internal/TcpDiscoveryNode.java | 19 +
.../messages/TcpDiscoveryAbstractMessage.java | 3 +
.../messages/TcpDiscoveryClientAckResponse.java | 64 +
.../messages/TcpDiscoveryHandshakeResponse.java | 14 +
.../spi/swapspace/file/FileSwapSpaceSpi.java | 2 +-
.../internal/GridUpdateNotifierSelfTest.java | 15 +-
.../IgniteClientReconnectAbstractTest.java | 363 ++++++
.../IgniteClientReconnectApiExceptionTest.java | 846 ++++++++++++
.../IgniteClientReconnectAtomicsTest.java | 672 ++++++++++
.../IgniteClientReconnectCacheTest.java | 1202 ++++++++++++++++++
.../IgniteClientReconnectCollectionsTest.java | 443 +++++++
.../IgniteClientReconnectComputeTest.java | 192 +++
...eClientReconnectContinuousProcessorTest.java | 372 ++++++
...IgniteClientReconnectDiscoveryStateTest.java | 123 ++
...niteClientReconnectFailoverAbstractTest.java | 231 ++++
.../IgniteClientReconnectFailoverTest.java | 212 +++
.../IgniteClientReconnectServicesTest.java | 260 ++++
.../internal/IgniteClientReconnectStopTest.java | 106 ++
.../IgniteClientReconnectStreamerTest.java | 233 ++++
.../IgniteSlowClientDetectionSelfTest.java | 1 +
.../GridDeploymentManagerStopSelfTest.java | 7 +
.../cache/GridCacheAbstractFullApiSelfTest.java | 32 +
.../IgniteCacheAbstractStopBusySelfTest.java | 2 +-
.../cache/IgniteCacheDynamicStopSelfTest.java | 6 +-
.../IgniteTxExceptionAbstractSelfTest.java | 1 +
.../GridCacheAbstractNodeRestartSelfTest.java | 12 +-
.../IgniteCacheSystemTransactionsSelfTest.java | 2 +-
.../GridCacheReplicatedInvalidateSelfTest.java | 3 +-
...acheAtomicReplicatedNodeRestartSelfTest.java | 5 +
.../loadtests/hashmap/GridCacheTestContext.java | 4 +-
.../tcp/TcpClientDiscoverySpiSelfTest.java | 562 +++++++-
.../tcp/TcpDiscoveryMultiThreadedTest.java | 140 +-
.../multijvm/IgniteClusterProcessProxy.java | 5 +
.../IgniteClientReconnectTestSuite.java | 48 +
.../geospatial/licenses/jts-lgpl-license.txt | 165 ---
.../licenses/hibernate-lgpl-2.1-license.txt | 174 ---
modules/indexing/licenses/h2-epl-license.txt | 69 -
.../processors/query/h2/IgniteH2Indexing.java | 5 +
.../query/h2/twostep/GridMergeIndex.java | 45 +-
.../h2/twostep/GridReduceQueryExecutor.java | 70 +-
...ClientReconnectCacheQueriesFailoverTest.java | 225 ++++
.../cache/IgniteClientReconnectQueriesTest.java | 427 +++++++
...dCacheAbstractReduceFieldsQuerySelfTest.java | 4 +
.../IgniteCacheWithIndexingTestSuite.java | 1 +
modules/jta/licenses/jta-license.txt | 2 +
modules/mesos/licenses/jetty-epl-license.txt | 69 -
modules/mesos/pom.xml | 1 +
.../apache/ignite/mesos/ClusterProperties.java | 14 +
.../apache/ignite/mesos/IgniteScheduler.java | 26 +-
modules/rest-http/pom.xml | 6 +-
.../scalar-2.10/licenses/scala-bsd-license.txt | 18 -
modules/scalar/licenses/scala-bsd-license.txt | 18 -
.../licenses/cron4j-lgpl-2.1-license.txt | 174 ---
modules/slf4j/licenses/sl4j-mit-license.txt | 21 -
.../spark-2.10/licenses/scala-bsd-license.txt | 18 -
modules/spark/licenses/scala-bsd-license.txt | 18 -
.../apache/ignite/cache/spring/SpringCache.java | 2 +-
.../ignite/cache/spring/SpringCacheManager.java | 22 +-
modules/ssh/licenses/jcraft-revised-bsd.txt | 28 -
modules/tools/licenses/jodd-revised-bsd.txt | 21 -
.../urideploy/licenses/jtidy-mit-license.txt | 50 -
modules/urideploy/pom.xml | 6 +-
.../licenses/jline-bsd-license.txt | 18 -
.../licenses/scala-bsd-license.txt | 18 -
.../licenses/slf4j-mit-license.txt | 21 -
modules/web/pom.xml | 6 +-
modules/yarn/README.txt | 28 +
modules/yarn/licenses/apache-2.0.txt | 202 +++
modules/yarn/pom.xml | 101 ++
.../apache/ignite/yarn/ApplicationMaster.java | 400 ++++++
.../apache/ignite/yarn/ClusterProperties.java | 432 +++++++
.../org/apache/ignite/yarn/IgniteContainer.java | 84 ++
.../org/apache/ignite/yarn/IgniteProvider.java | 339 +++++
.../apache/ignite/yarn/IgniteYarnClient.java | 178 +++
.../org/apache/ignite/yarn/package-info.java | 22 +
.../ignite/yarn/utils/IgniteYarnUtils.java | 81 ++
.../main/resources/ignite-default-config.xml | 35 +
.../org/apache/ignite/IgniteYarnTestSuite.java | 38 +
.../yarn/IgniteApplicationMasterSelfTest.java | 532 ++++++++
parent/pom.xml | 55 +
pom.xml | 16 +-
192 files changed, 13357 insertions(+), 1958 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37198cae/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
index 500c461,2f3def9..5f8a7d0
--- a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
@@@ -73,15 -71,9 +72,18 @@@ public abstract class IgniteSpiAdapter
/** Discovery listener. */
private GridLocalEventListener paramsLsnr;
+ /** Local node. */
+ private ClusterNode locNode;
+
+ /** Failure detection threshold will not be used usage switch. */
+ private boolean failureDetectionThresholdEnabled = true;
+
+ /**
+ * Failure detection threshold. Initialized with the value of
+ * {@link IgniteConfiguration#getFailureDetectionThreshold()}.
+ */
+ private long failureDetectionThreshold;
+
/**
* Creates new adapter and initializes it from the current (this) class.
* SPI name will be initialized to the simple name of the class
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37198cae/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index b2bc9eb,e9fd696..06804ed
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@@ -1243,10 -1264,6 +1270,8 @@@ public class TcpCommunicationSpi extend
/** {@inheritDoc} */
@Override public Map<String, Object> getNodeAttributes() throws IgniteSpiException {
- nodeIdMsg = new NodeIdMessage(getLocalNodeId());
-
+ initFailureDetectionThreshold();
+
assertParameter(locPort > 1023, "locPort > 1023");
assertParameter(locPort <= 0xffff, "locPort < 0xffff");
assertParameter(locPortRange >= 0, "locPortRange >= 0");
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37198cae/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
index 62b2e35,572ba2c..4fff162
--- 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
@@@ -265,17 -275,28 +276,29 @@@ class ClientImpl extends TcpDiscoveryIm
return false;
}
+ else if (state == DISCONNECTED) {
+ if (pingFuts.remove(nodeId, fut))
+ fut.onDone(new IgniteClientDisconnectedCheckedException(null,
+ "Failed to ping node, client node disconnected."));
+ }
+ else {
+ final GridFutureAdapter<Boolean> finalFut = fut;
- final GridFutureAdapter<Boolean> finalFut = fut;
-
- timer.schedule(new TimerTask() {
- @Override public void run() {
- if (pingFuts.remove(nodeId, finalFut))
- finalFut.onDone(false);
- }
- }, spi.failureDetectionThresholdEnabled() ? spi.failureDetectionThreshold() : spi.getNetworkTimeout());
+ timer.schedule(new TimerTask() {
+ @Override public void run() {
+ if (pingFuts.remove(nodeId, finalFut)) {
+ if (ClientImpl.this.state == DISCONNECTED)
+ finalFut.onDone(new IgniteClientDisconnectedCheckedException(null,
+ "Failed to ping node, client node disconnected."));
+ else
+ finalFut.onDone(false);
+ }
+ }
- }, spi.netTimeout);
++ }, spi.failureDetectionThresholdEnabled() ? spi.failureDetectionThreshold() :
++ spi.getNetworkTimeout());
- sockWriter.sendMessage(new TcpDiscoveryClientPingRequest(getLocalNodeId(), nodeId));
+ sockWriter.sendMessage(new TcpDiscoveryClientPingRequest(getLocalNodeId(), nodeId));
+ }
}
}
@@@ -512,7 -540,7 +546,7 @@@
log.debug("Message has been sent to address [msg=" + msg + ", addr=" + addr +
", rmtNodeId=" + rmtNodeId + ']');
- return new T2<>(sock, spi.readReceipt(sock, ackTimeout0));
- return new T3<>(sock, spi.readReceipt(sock, ackTimeout0), res.clientAck());
++ return new T3<>(sock, spi.readReceipt(sock, timeoutCtrl.nextTimeoutChunk(ackTimeout0)), res.clientAck());
}
catch (IOException | IgniteCheckedException e) {
U.closeQuiet(sock);
@@@ -838,8 -868,8 +881,11 @@@
private final Queue<TcpDiscoveryAbstractMessage> queue = new ArrayDeque<>();
/** */
+ private final long socketTimeout;
+
++ /** */
+ private TcpDiscoveryAbstractMessage unackedMsg;
+
/**
*
*/
@@@ -910,10 -957,43 +976,47 @@@
for (IgniteInClosure<TcpDiscoveryAbstractMessage> msgLsnr : spi.sendMsgLsnrs)
msgLsnr.apply(msg);
+ boolean ack = clientAck && !(msg instanceof TcpDiscoveryPingResponse);
+
try {
+ if (ack) {
+ synchronized (mux) {
+ assert unackedMsg == null : unackedMsg;
+
+ unackedMsg = msg;
+ }
+ }
+
- spi.writeToSocket(sock, msg);
+ spi.writeToSocket(sock, msg, socketTimeout);
msg = null;
+
+ if (ack) {
- long waitEnd = U.currentTimeMillis() + spi.ackTimeout;
++ long waitEnd = U.currentTimeMillis() + (spi.failureDetectionThresholdEnabled() ?
++ spi.failureDetectionThreshold() : spi.getAckTimeout());
+
+ TcpDiscoveryAbstractMessage unacked;
+
+ synchronized (mux) {
+ while (unackedMsg != null && U.currentTimeMillis() < waitEnd)
+ mux.wait(waitEnd);
+
+ unacked = unackedMsg;
+
+ unackedMsg = null;
+ }
+
+ if (unacked != null) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to get acknowledge for message, will try to reconnect " +
- "[msg=" + unacked + ", timeout=" + spi.ackTimeout + ']');
++ "[msg=" + unacked +
++ (spi.failureDetectionThresholdEnabled() ?
++ ", failureDetectionThershold=" + spi.failureDetectionThreshold() :
++ ", timeout=" + spi.getAckTimeout()) + ']');
+
+ throw new IOException("Failed to get acknowledge for message: " + unacked);
+ }
+ }
}
catch (IOException e) {
if (log.isDebugEnabled())
@@@ -985,18 -1068,12 +1092,18 @@@
if (join) {
joinError(new IgniteSpiException("Join process timed out, connection failed and " +
"failed to reconnect (consider increasing 'joinTimeout' configuration property) " +
- "[networkTimeout=" + spi.joinTimeout + ", sock=" + sock + ']'));
+ "[networkTimeout=" + spi.getJoinTimeout() + ", sock=" + sock + ']'));
}
else
- U.error(log, "Failed to reconnect to cluster (consider increasing 'networkTimeout' " +
- "configuration property) [networkTimeout=" + spi.netTimeout + ", sock=" + sock + ']');
-
+ U.error(log, "Failed to reconnect to cluster " +
+ (spi.failureDetectionThresholdEnabled() ?
- "(consider increasing 'failureDetectionThreshold' configuration property) " +
- "[failureDetectionThreshold=" + spi.failureDetectionThreshold()
- :
- "(consider increasing 'networkTimeout' configuration property) [networkTimeout="
- + spi.getNetworkTimeout())
++ "(consider increasing 'failureDetectionThreshold' configuration property) " +
++ "[failureDetectionThreshold=" + spi.failureDetectionThreshold()
++ :
++ "(consider increasing 'networkTimeout' configuration property) [networkTimeout="
++ + spi.getNetworkTimeout())
+ + ", sock=" + sock + ']');
-
++
return;
}
@@@ -1160,15 -1216,26 +1260,26 @@@
Object msg = queue.take();
if (msg == JOIN_TIMEOUT) {
- if (joinLatch.getCount() > 0) {
+ if (state == STARTING) {
joinError(new IgniteSpiException("Join process timed out, did not receive response for " +
"join request (consider increasing 'joinTimeout' configuration property) " +
- "[joinTimeout=" + spi.getJoinTimeout() + ", sock=" + sock + ']'));
- "[joinTimeout=" + spi.joinTimeout + ", sock=" + currSock + ']'));
++ "[joinTimeout=" + spi.getJoinTimeout() + ", sock=" + currSock + ']'));
break;
}
+ else if (state == DISCONNECTED) {
+ if (log.isDebugEnabled())
+ log.debug("Failed to reconnect, local node segmented " +
- "[joinTimeout=" + spi.joinTimeout + ']');
++ "[joinTimeout=" + spi.getJoinTimeout() + ']');
+
+ state = SEGMENTED;
+
+ notifyDiscovery(EVT_NODE_SEGMENTED, topVer, locNode, allVisibleNodes());
+ }
}
else if (msg == SPI_STOP) {
+ state = STOPPED;
+
assert spi.getSpiContext().isStopping();
if (currSock != null) {
@@@ -1206,13 -1276,58 +1320,64 @@@
}
}
else if (msg == SPI_RECONNECT_FAILED) {
- if (!segmented) {
- segmented = true;
+ reconnector.cancel();
+ reconnector.join();
+
+ reconnector = null;
+
+ if (spi.isClientReconnectDisabled()) {
+ if (state != SEGMENTED && state != STOPPED) {
+ if (log.isDebugEnabled()) {
+ log.debug("Failed to restore closed connection, reconnect disabled, " +
- "local node segmented [networkTimeout=" + spi.netTimeout + ']');
++ (spi.failureDetectionThresholdEnabled() ?
++ "local node segmented [failureDetectionThreshold=" +
++ spi.failureDetectionThreshold() + ']' :
++ "local node segmented [networkTimeout=" + spi.getNetworkTimeout() + ']'));
+ }
- reconnector.cancel();
- reconnector.join();
+ state = SEGMENTED;
- notifyDiscovery(EVT_NODE_SEGMENTED, topVer, locNode, allVisibleNodes());
+ notifyDiscovery(EVT_NODE_SEGMENTED, topVer, locNode, allVisibleNodes());
+ }
+ }
+ else {
+ if (state == STARTING || state == CONNECTED) {
+ if (log.isDebugEnabled()) {
+ log.debug("Failed to restore closed connection, will try to reconnect " +
- "[networkTimeout=" + spi.netTimeout + ", joinTimeout=" + spi.joinTimeout + ']');
++ (spi.failureDetectionThresholdEnabled() ?
++ "[failureDetectionThreshold=" + spi.failureDetectionThreshold() :
++ "[networkTimeout=" + spi.getNetworkTimeout()) +
++ ", joinTimeout=" + spi.getJoinTimeout() + ']');
+ }
+
+ state = DISCONNECTED;
+
+ nodeAdded = false;
+
+ IgniteClientDisconnectedCheckedException err =
+ new IgniteClientDisconnectedCheckedException(null, "Failed to ping node, " +
+ "client node disconnected.");
+
+ for (Map.Entry<UUID, GridFutureAdapter<Boolean>> e : pingFuts.entrySet()) {
+ GridFutureAdapter<Boolean> fut = e.getValue();
+
+ if (pingFuts.remove(e.getKey(), fut))
+ fut.onDone(err);
+ }
+
+ notifyDiscovery(EVT_CLIENT_NODE_DISCONNECTED, topVer, locNode, allVisibleNodes());
+ }
+
+ UUID newId = UUID.randomUUID();
+
+ if (log.isInfoEnabled()) {
+ log.info("Client node disconnected from cluster, will try to reconnect with new id " +
+ "[newId=" + newId + ", prevId=" + locNode.id() + ", locNode=" + locNode + ']');
+ }
+
+ locNode.onClientDisconnected(newId);
+
+ tryJoin();
}
}
else {
@@@ -1254,6 -1377,48 +1427,48 @@@
}
/**
+ * @throws InterruptedException If interrupted.
+ */
+ private void tryJoin() throws InterruptedException {
+ assert state == DISCONNECTED || state == STARTING : state;
+
+ boolean join = state == STARTING;
+
+ joinCnt++;
+
- T2<Socket, Boolean> joinRes = joinTopology(false, spi.joinTimeout);
++ T2<Socket, Boolean> joinRes = joinTopology(false, spi.getJoinTimeout());
+
+ if (joinRes == null) {
+ if (join)
+ joinError(new IgniteSpiException("Join process timed out."));
+ else {
+ state = SEGMENTED;
+
+ notifyDiscovery(EVT_NODE_SEGMENTED, topVer, locNode, allVisibleNodes());
+ }
+
+ return;
+ }
+
+ currSock = joinRes.get1();
+
+ sockWriter.setSocket(joinRes.get1(), joinRes.get2());
+
- if (spi.joinTimeout > 0) {
++ if (spi.getJoinTimeout() > 0) {
+ final int joinCnt0 = joinCnt;
+
+ timer.schedule(new TimerTask() {
+ @Override public void run() {
+ if (joinCnt == joinCnt0 && joining())
+ queue.add(JOIN_TIMEOUT);
+ }
- }, spi.joinTimeout);
++ }, spi.getJoinTimeout());
+ }
+
+ sockReader.setSocket(joinRes.get1(), locNode.clientRouterNodeId());
+ }
+
+ /**
* @param msg Message.
*/
protected void processDiscoveryMessage(TcpDiscoveryAbstractMessage msg) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37198cae/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index fca3104,591f404..3a50c31
--- 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
@@@ -4238,7 -4203,10 +4271,10 @@@ class ServerImpl extends TcpDiscoveryIm
TcpDiscoveryHandshakeResponse res =
new TcpDiscoveryHandshakeResponse(locNodeId, locNode.internalOrder());
+ if (req.client())
+ res.clientAck(true);
+
- spi.writeToSocket(sock, res);
+ spi.writeToSocket(sock, res, timeoutCtrl.nextTimeoutChunk(spi.getSocketTimeout()));
// It can happen if a remote node is stopped and it has a loopback address in the list of addresses,
// the local node sends a handshake request message on the loopback address, so we get here.
@@@ -4387,9 -4347,9 +4423,9 @@@
TcpDiscoverySpiState state = spiStateCopy();
if (state == CONNECTED) {
- spi.writeToSocket(msg, sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK, socketTimeout);
- if (clientMsgWrk != null && clientMsgWrk.getState() == State.NEW)
+ if (clientMsgWrk.getState() == State.NEW)
clientMsgWrk.start();
msgWorker.addMessage(msg);
@@@ -4533,8 -4493,15 +4569,15 @@@
msgWorker.addMessage(msg);
// Send receipt back.
- if (clientMsgWrk == null)
+ if (clientMsgWrk != null) {
+ TcpDiscoveryClientAckResponse ack = new TcpDiscoveryClientAckResponse(locNodeId, msg.id());
+
+ ack.verify(locNodeId);
+
+ clientMsgWrk.addMessage(ack);
+ }
+ else
- spi.writeToSocket(msg, sock, RES_OK);
+ spi.writeToSocket(msg, sock, RES_OK, socketTimeout);
}
catch (IgniteCheckedException e) {
if (log.isDebugEnabled())
@@@ -4788,18 -4765,40 +4844,42 @@@
try {
assert msg.verified() : msg;
- if (log.isDebugEnabled())
- log.debug("Redirecting message to client [sock=" + sock + ", locNodeId="
- + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
+ if (msg instanceof TcpDiscoveryClientAckResponse) {
+ if (clientVer == null) {
+ ClusterNode node = spi.getNode(clientNodeId);
- try {
- prepareNodeAddedMessage(msg, clientNodeId, null, null);
+ if (node != null)
+ clientVer = IgniteUtils.productVersion(node);
+ else if (log.isDebugEnabled())
+ log.debug("Skip sending message ack to client, fail to get client node " +
+ "[sock=" + sock + ", locNodeId=" + getLocalNodeId() +
+ ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
+ }
- writeToSocket(sock, msg, spi.failureDetectionThresholdEnabled() ? spi.failureDetectionThreshold() :
- spi.getSocketTimeout());
+ if (clientVer != null &&
+ clientVer.compareTo(TcpDiscoveryClientAckResponse.CLIENT_ACK_SINCE_VERSION) >= 0) {
+ if (log.isDebugEnabled())
+ log.debug("Sending message ack to client [sock=" + sock + ", locNodeId="
+ + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
+
- writeToSocket(sock, msg);
++ writeToSocket(sock, msg, spi.failureDetectionThresholdEnabled() ? spi.failureDetectionThreshold() :
++ spi.getSocketTimeout());
+ }
}
- finally {
- clearNodeAddedMessage(msg);
+ else {
+ try {
+ if (log.isDebugEnabled())
+ log.debug("Redirecting message to client [sock=" + sock + ", locNodeId="
+ + getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ", msg=" + msg + ']');
+
+ prepareNodeAddedMessage(msg, clientNodeId, null, null);
+
- writeToSocket(sock, msg);
++ writeToSocket(sock, msg, spi.failureDetectionThresholdEnabled() ? spi.failureDetectionThreshold() :
++ spi.getSocketTimeout());
+ }
+ finally {
+ clearNodeAddedMessage(msg);
+ }
}
}
catch (IgniteCheckedException | IOException e) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37198cae/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37198cae/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37198cae/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
index 0000000,fbaea11..3af6fd8
mode 000000,100644..100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientReconnectAbstractTest.java
@@@ -1,0 -1,363 +1,363 @@@
+ /*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+ package org.apache.ignite.internal;
+
+ import org.apache.ignite.*;
+ import org.apache.ignite.cluster.*;
+ import org.apache.ignite.configuration.*;
+ import org.apache.ignite.events.*;
+ import org.apache.ignite.internal.managers.communication.*;
+ import org.apache.ignite.internal.util.typedef.*;
+ import org.apache.ignite.internal.util.typedef.internal.*;
+ import org.apache.ignite.lang.*;
+ import org.apache.ignite.plugin.extensions.communication.*;
+ import org.apache.ignite.resources.*;
+ import org.apache.ignite.spi.*;
+ import org.apache.ignite.spi.communication.tcp.*;
+ import org.apache.ignite.spi.discovery.tcp.*;
+ import org.apache.ignite.spi.discovery.tcp.internal.*;
+ import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
+ import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
+ import org.apache.ignite.spi.discovery.tcp.messages.*;
+ import org.apache.ignite.testframework.junits.common.*;
+ import org.jetbrains.annotations.*;
+
+ import javax.cache.*;
+ import java.io.*;
+ import java.net.*;
+ import java.util.concurrent.*;
+ import java.util.concurrent.atomic.*;
+
+ import static java.util.concurrent.TimeUnit.*;
+ import static org.apache.ignite.events.EventType.*;
+
+ /**
+ *
+ */
+ public abstract class IgniteClientReconnectAbstractTest extends GridCommonAbstractTest {
+ /** */
+ private static TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** */
+ private static final long RECONNECT_TIMEOUT = 10_000;
+
+ /** */
+ protected boolean clientMode;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TestTcpDiscoverySpi disco = new TestTcpDiscoverySpi();
+
+ disco.setIpFinder(ipFinder);
+ disco.setJoinTimeout(2 * 60_000);
+ disco.setSocketTimeout(1000);
+ disco.setNetworkTimeout(2000);
+
+ cfg.setDiscoverySpi(disco);
+
+ BlockTpcCommunicationSpi commSpi = new BlockTpcCommunicationSpi();
+
+ commSpi.setSharedMemoryPort(-1);
+
+ cfg.setCommunicationSpi(commSpi);
+
+ if (clientMode)
+ cfg.setClientMode(true);
+
+ return cfg;
+ }
+
+ /**
+ * @param latch Latch.
+ * @throws Exception If failed.
+ */
+ protected void waitReconnectEvent(CountDownLatch latch) throws Exception {
+ if (!latch.await(RECONNECT_TIMEOUT, MILLISECONDS)) {
+ log.error("Failed to wait for reconnect event, will dump threads, latch count: " + latch.getCount());
+
+ U.dumpThreads(log);
+
+ fail("Failed to wait for disconnect/reconnect event.");
+ }
+ }
+
+ /**
+ * @return Number of server nodes started before tests.
+ */
+ protected abstract int serverCount();
+
+ /**
+ * @return Number of client nodes started before tests.
+ */
+ protected int clientCount() {
+ return 0;
+ }
+
+ /**
+ * @param ignite Node.
+ * @return Discovery SPI.
+ */
+ protected TestTcpDiscoverySpi spi(Ignite ignite) {
+ return ((TestTcpDiscoverySpi)ignite.configuration().getDiscoverySpi());
+ }
+
+ /**
+ * @param ignite Node.
+ * @return Communication SPI.
+ */
+ protected BlockTpcCommunicationSpi commSpi(Ignite ignite) {
+ return ((BlockTpcCommunicationSpi)ignite.configuration().getCommunicationSpi());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ int srvs = serverCount();
+
+ if (srvs > 0)
+ startGrids(srvs);
+
+ int clients = clientCount();
+
+ if (clients > 0) {
+ clientMode = true;
+
+ startGridsMultiThreaded(srvs, clients);
+
+ clientMode = false;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+ }
+
+ /**
+ * @param client Client.
+ * @return Server node client connected to.
+ */
+ protected Ignite clientRouter(Ignite client) {
+ TcpDiscoveryNode node = (TcpDiscoveryNode)client.cluster().localNode();
+
+ assertTrue(node.isClient());
+ assertNotNull(node.clientRouterNodeId());
+
+ Ignite srv = G.ignite(node.clientRouterNodeId());
+
+ assertNotNull(srv);
+
+ return srv;
+ }
+
+ /**
+ * @param fut Future.
+ * @throws Exception If failed.
+ */
+ protected void assertNotDone(IgniteInternalFuture<?> fut) throws Exception {
+ assertNotNull(fut);
+
+ if (fut.isDone())
+ fail("Future completed with result: " + fut.get());
+ }
+
+ /**
+ * Reconnect client node.
+ *
+ * @param client Client.
+ * @param srv Server.
+ * @param disconnectedC Closure which will be run when client node disconnected.
+ * @throws Exception If failed.
+ */
+ protected void reconnectClientNode(Ignite client, Ignite srv, @Nullable Runnable disconnectedC)
+ throws Exception {
+ final TestTcpDiscoverySpi clientSpi = spi(client);
+ final TestTcpDiscoverySpi srvSpi = spi(srv);
+
+ final CountDownLatch disconnectLatch = new CountDownLatch(1);
+ final CountDownLatch reconnectLatch = new CountDownLatch(1);
+
+ log.info("Block reconnect.");
+
+ clientSpi.writeLatch = new CountDownLatch(1);
+
+ IgnitePredicate<Event> p = new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event evt) {
+ if (evt.type() == EVT_CLIENT_NODE_DISCONNECTED) {
+ info("Disconnected: " + evt);
+
+ disconnectLatch.countDown();
+ }
+ else if (evt.type() == EVT_CLIENT_NODE_RECONNECTED) {
+ info("Reconnected: " + evt);
+
+ reconnectLatch.countDown();
+ }
+
+ return true;
+ }
+ };
+
+ client.events().localListen(p, EVT_CLIENT_NODE_DISCONNECTED, EVT_CLIENT_NODE_RECONNECTED);
+
+ srvSpi.failNode(client.cluster().localNode().id(), null);
+
+ waitReconnectEvent(disconnectLatch);
+
+ if (disconnectedC != null)
+ disconnectedC.run();
+
+ log.info("Allow reconnect.");
+
+ clientSpi.writeLatch.countDown();
+
+ waitReconnectEvent(reconnectLatch);
+
+ client.events().stopLocalListen(p);
+ }
+
+ /**
+ * @param e Client disconnected exception.
+ * @return Reconnect future.
+ */
+ protected IgniteFuture<?> check(CacheException e) {
+ log.info("Expected exception: " + e);
+
+ if (!(e.getCause() instanceof IgniteClientDisconnectedException))
+ log.error("Unexpected cause: " + e.getCause(), e);
+
+ assertTrue("Unexpected cause: " + e.getCause(), e.getCause() instanceof IgniteClientDisconnectedException);
+
+ IgniteClientDisconnectedException e0 = (IgniteClientDisconnectedException)e.getCause();
+
+ assertNotNull(e0.reconnectFuture());
+
+ return e0.reconnectFuture();
+ }
+
+ /**
+ * @param e Client disconnected exception.
+ */
+ protected void checkAndWait(CacheException e) {
+ check(e).get();
+ }
+
+ /**
+ * @param e Client disconnected exception.
+ */
+ protected void checkAndWait(IgniteClientDisconnectedException e) {
+ log.info("Expected exception: " + e);
+
+ assertNotNull(e.reconnectFuture());
+
+ e.reconnectFuture().get();
+ }
+
+ /**
+ *
+ */
+ protected static class TestTcpDiscoverySpi extends TcpDiscoverySpi {
+ /** */
+ volatile CountDownLatch writeLatch;
+
+ /** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg)
++ @Override protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout)
+ throws IOException, IgniteCheckedException {
+ if (msg instanceof TcpDiscoveryJoinRequestMessage) {
+ CountDownLatch writeLatch0 = writeLatch;
+
+ if (writeLatch0 != null) {
+ log.info("Block join request send: " + msg);
+
+ U.await(writeLatch0);
+ }
+ }
+
- super.writeToSocket(sock, msg);
++ super.writeToSocket(sock, msg, getSocketTimeout());
+ }
+ }
+
+ /**
+ *
+ */
+ protected static class BlockTpcCommunicationSpi extends TcpCommunicationSpi {
+ /** */
+ volatile Class msgCls;
+
+ /** */
+ AtomicBoolean collectStart = new AtomicBoolean(false);
+
+ /** */
+ ConcurrentHashMap<String, ClusterNode> classes = new ConcurrentHashMap<>();
+
+ /** */
+ @LoggerResource
+ private IgniteLogger log;
+
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg) throws IgniteSpiException {
+ Class msgCls0 = msgCls;
+
+ if (collectStart.get() && msg instanceof GridIoMessage)
+ classes.put(((GridIoMessage)msg).message().getClass().getName(), node);
+
+ if (msgCls0 != null && msg instanceof GridIoMessage
+ && ((GridIoMessage)msg).message().getClass().equals(msgCls)) {
+ log.info("Block message: " + msg);
+
+ return;
+ }
+
+ super.sendMessage(node, msg);
+ }
+
+ /**
+ * @param clazz Class of messages which will be block.
+ */
+ public void blockMessage(Class clazz) {
+ msgCls = clazz;
+ }
+
+ /**
+ * Unlock all message.
+ */
+ public void unblockMessage() {
+ msgCls = null;
+ }
+
+ /**
+ * Start collect messages.
+ */
+ public void start() {
+ collectStart.set(true);
+ }
+
+ /**
+ * Print collected messages.
+ */
+ public void print() {
+ for (String s : classes.keySet())
+ log.error(s);
+ }
+ }
+ }
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/37198cae/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------