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/06/16 08:44:22 UTC
[1/9] incubator-ignite git commit: # ignite-883
Repository: incubator-ignite
Updated Branches:
refs/heads/ignite-sprint-6 090733868 -> 5d8a5e619
# ignite-883
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/de0d61f6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/de0d61f6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/de0d61f6
Branch: refs/heads/ignite-sprint-6
Commit: de0d61f6a3467379c168a15bbd9c4580aeebb092
Parents: 89a4f7c
Author: sboikov <sb...@gridgain.com>
Authored: Thu Jun 11 10:40:27 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Jun 11 17:40:01 2015 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/IgniteKernal.java | 1 -
.../discovery/GridDiscoveryManager.java | 9 +-
.../dht/preloader/GridDhtPreloader.java | 2 +-
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 28 ++-
.../communication/tcp/TcpCommunicationSpi.java | 2 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 214 ++++++++++++-------
.../ignite/spi/discovery/tcp/ServerImpl.java | 64 ------
.../spi/discovery/tcp/TcpDiscoveryImpl.java | 66 ++++++
.../distributed/IgniteCacheManyClientsTest.java | 87 +++++++-
9 files changed, 323 insertions(+), 150 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/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 4f5e365..f38fee1 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
@@ -1001,7 +1001,6 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
A.notNull(cfg.getMBeanServer(), "cfg.getMBeanServer()");
A.notNull(cfg.getGridLogger(), "cfg.getGridLogger()");
A.notNull(cfg.getMarshaller(), "cfg.getMarshaller()");
- A.notNull(cfg.getPublicThreadPoolSize(), "cfg.getPublicThreadPoolSize()");
A.notNull(cfg.getUserAttributes(), "cfg.getUserAttributes()");
// All SPIs should be non-null.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/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 71fbc61..464110c 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
@@ -2057,17 +2057,22 @@ public class GridDiscoveryManager extends GridManagerAdapter<DiscoverySpi> {
private final AffinityTopologyVersion topVer;
/** */
+ @GridToStringExclude
private final DiscoCache discoCache;
/**
* @param topVer Topology version.
* @param discoCache Disco cache.
*/
- private Snapshot(AffinityTopologyVersion topVer,
- DiscoCache discoCache) {
+ private Snapshot(AffinityTopologyVersion topVer, DiscoCache discoCache) {
this.topVer = topVer;
this.discoCache = discoCache;
}
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(Snapshot.class, this);
+ }
}
/** Cache for discovery collections. */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
index 51010ce..0355bb3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
@@ -102,7 +102,7 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
boolean set = topVer.setIfGreater(e.topologyVersion());
assert set : "Have you configured TcpDiscoverySpi for your in-memory data grid? [newVer=" +
- e.topologyVersion() + ", curVer=" + topVer.get() + ']';
+ e.topologyVersion() + ", curVer=" + topVer.get() + ", evt=" + e + ']';
if (e.type() == EVT_NODE_LEFT || e.type() == EVT_NODE_FAILED) {
for (GridDhtAssignmentFetchFuture fut : pendingAssignmentFetchFuts.values())
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
index 6e7a706..476f8a8 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/IgniteSpiAdapter.java
@@ -585,8 +585,32 @@ public abstract class IgniteSpiAdapter implements IgniteSpi, IgniteSpiManagement
GridDummySpiContext(ClusterNode locNode, boolean stopping, @Nullable IgniteSpiContext spiCtx) {
this.locNode = locNode;
this.stopping = stopping;
- this.msgFactory = spiCtx != null ? spiCtx.messageFactory() : null;
- this.msgFormatter = spiCtx != null ? spiCtx.messageFormatter() : null;
+
+ MessageFactory msgFactory0 = spiCtx != null ? spiCtx.messageFactory() : null;
+ MessageFormatter msgFormatter0 = spiCtx != null ? spiCtx.messageFormatter() : null;
+
+ if (msgFactory0 == null) {
+ msgFactory0 = new MessageFactory() {
+ @Nullable @Override public Message create(byte type) {
+ throw new IgniteException("Failed to read message, node is not started.");
+ }
+ };
+ }
+
+ if (msgFormatter0 == null) {
+ msgFormatter0 = new MessageFormatter() {
+ @Override public MessageWriter writer() {
+ throw new IgniteException("Failed to write message, node is not started.");
+ }
+
+ @Override public MessageReader reader(MessageFactory factory) {
+ throw new IgniteException("Failed to read message, node is not started.");
+ }
+ };
+ }
+
+ this.msgFactory = msgFactory0;
+ this.msgFormatter = msgFormatter0;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git 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
index a661965..f19e25b 100644
--- 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
@@ -2028,7 +2028,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
if (X.hasCause(e, SocketTimeoutException.class))
LT.warn(log, null, "Connect timed out (consider increasing 'connTimeout' " +
- "configuration property) [addr=" + addr + ']');
+ "configuration property) [addr=" + addr + ", connTimeout" + connTimeout + ']');
if (errs == null)
errs = new IgniteCheckedException("Failed to connect to node (is node still alive?). " +
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/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 d064c8d..23297ed 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
@@ -195,7 +195,7 @@ class ClientImpl extends TcpDiscoveryImpl {
U.warn(log, "Failed to left node: timeout [nodeId=" + locNode + ']');
}
catch (InterruptedException ignored) {
-
+ // No-op.
}
}
@@ -282,7 +282,7 @@ class ClientImpl extends TcpDiscoveryImpl {
return false;
}
catch (IgniteCheckedException e) {
- throw new IgniteSpiException(e); // Should newer occur
+ throw new IgniteSpiException(e); // Should newer occur.
}
}
@@ -347,7 +347,10 @@ class ClientImpl extends TcpDiscoveryImpl {
}
/**
+ * @param recon {@code True} if reconnects.
* @return Opened socket or {@code null} if timeout.
+ * @throws InterruptedException If interrupted.
+ * @throws IgniteSpiException If failed.
* @see TcpDiscoverySpi#joinTimeout
*/
@SuppressWarnings("BusyWait")
@@ -387,71 +390,152 @@ class ClientImpl extends TcpDiscoveryImpl {
InetSocketAddress addr = it.next();
- Socket sock = null;
+ T2<Socket, Integer> sockAndRes = sendJoinRequest(recon, addr);
- try {
- long ts = U.currentTimeMillis();
+ if (sockAndRes == null) {
+ it.remove();
- IgniteBiTuple<Socket, UUID> t = initConnection(addr);
+ continue;
+ }
- sock = t.get1();
+ assert sockAndRes.get1() != null;
+ assert sockAndRes.get2() != null;
- UUID rmtNodeId = t.get2();
+ Socket sock = sockAndRes.get1();
- spi.stats.onClientSocketInitialized(U.currentTimeMillis() - ts);
+ switch (sockAndRes.get2()) {
+ case RES_OK:
+ return sock;
- locNode.clientRouterNodeId(rmtNodeId);
+ case RES_CONTINUE_JOIN:
+ case RES_WAIT:
+ U.closeQuiet(sock);
- TcpDiscoveryAbstractMessage msg = recon ?
- new TcpDiscoveryClientReconnectMessage(getLocalNodeId(), rmtNodeId,
- lastMsgId) :
- new TcpDiscoveryJoinRequestMessage(locNode, spi.collectExchangeData(getLocalNodeId()));
+ break;
- msg.client(true);
+ default:
+ if (log.isDebugEnabled())
+ log.debug("Received unexpected response to join request: " + sockAndRes.get2());
- spi.writeToSocket(sock, msg);
+ U.closeQuiet(sock);
+ }
+ }
- int res = spi.readReceipt(sock, spi.ackTimeout);
+ if (addrs.isEmpty()) {
+ if (spi.joinTimeout > 0 && (U.currentTimeMillis() - startTime) > spi.joinTimeout)
+ return null;
- switch (res) {
- case RES_OK:
- return sock;
+ Thread.sleep(2000);
- case RES_CONTINUE_JOIN:
- case RES_WAIT:
- U.closeQuiet(sock);
+ U.warn(log, "Failed to connect to any address from IP finder (will retry to join topology " +
+ "in 2000ms): " + addrs0);
+ }
+ }
+ }
- break;
+ /**
+ * @param recon {@code True} if reconnects.
+ * @param addr Address.
+ * @return Socket and connect response.
+ */
+ @Nullable private T2<Socket, Integer> sendJoinRequest(boolean recon, InetSocketAddress addr) {
+ assert addr != null;
- default:
- if (log.isDebugEnabled())
- log.debug("Received unexpected response to join request: " + res);
+ Collection<Throwable> errs = null;
- U.closeQuiet(sock);
- }
- }
- catch (IOException | IgniteCheckedException e) {
- if (log.isDebugEnabled())
- U.error(log, "Failed to establish connection with address: " + addr, e);
+ long ackTimeout0 = spi.ackTimeout;
- U.closeQuiet(sock);
+ int connectAttempts = 1;
- it.remove();
- }
+ UUID locNodeId = getLocalNodeId();
+
+ for (int i = 0; i < spi.reconCnt; i++) {
+ boolean openSock = false;
+
+ Socket sock = null;
+
+ try {
+ long tstamp = U.currentTimeMillis();
+
+ sock = spi.openSocket(addr);
+
+ openSock = true;
+
+ TcpDiscoveryHandshakeRequest req = new TcpDiscoveryHandshakeRequest(locNodeId);
+
+ req.client(true);
+
+ spi.writeToSocket(sock, req);
+
+ TcpDiscoveryHandshakeResponse res = spi.readMessage(sock, null, ackTimeout0);
+
+ UUID rmtNodeId = res.creatorNodeId();
+
+ assert rmtNodeId != null;
+ assert !getLocalNodeId().equals(rmtNodeId);
+
+ spi.stats.onClientSocketInitialized(U.currentTimeMillis() - tstamp);
+
+ locNode.clientRouterNodeId(rmtNodeId);
+
+ tstamp = U.currentTimeMillis();
+
+ TcpDiscoveryAbstractMessage msg = recon ?
+ new TcpDiscoveryClientReconnectMessage(getLocalNodeId(), rmtNodeId, lastMsgId) :
+ new TcpDiscoveryJoinRequestMessage(locNode, spi.collectExchangeData(getLocalNodeId()));
+
+ msg.client(true);
+
+ spi.writeToSocket(sock, msg);
+
+ spi.stats.onMessageSent(msg, U.currentTimeMillis() - tstamp);
+
+ if (log.isDebugEnabled())
+ log.debug("Message has been sent to address [msg=" + msg + ", addr=" + addr +
+ ", rmtNodeId=" + rmtNodeId + ']');
+
+ return new T2<>(sock, spi.readReceipt(sock, ackTimeout0));
}
+ catch (IOException | IgniteCheckedException e) {
+ U.closeQuiet(sock);
- if (addrs.isEmpty()) {
- U.warn(log, "Failed to connect to any address from IP finder (will retry to join topology " +
- "in 2000ms): " + addrs0);
+ if (log.isDebugEnabled())
+ log.error("Exception on joining: " + e.getMessage(), e);
- if (spi.joinTimeout > 0 && (U.currentTimeMillis() - startTime) > spi.joinTimeout)
- return null;
+ onException("Exception on joining: " + e.getMessage(), e);
- Thread.sleep(2000);
+ if (errs == null)
+ errs = new ArrayList<>();
+
+ errs.add(e);
+
+ if (!openSock) {
+ // Reconnect for the second time, if connection is not established.
+ if (connectAttempts < 2) {
+ connectAttempts++;
+
+ continue;
+ }
+
+ break; // Don't retry if we can not establish connection.
+ }
+
+ if (e instanceof SocketTimeoutException || X.hasCause(e, SocketTimeoutException.class)) {
+ ackTimeout0 *= 2;
+
+ if (!checkAckTimeout(ackTimeout0))
+ break;
+ }
}
}
+
+ if (log.isDebugEnabled())
+ log.debug("Failed to join to address [addr=" + addr + ", recon=" + recon + ", errs=" + errs + ']');
+
+ return null;
}
+
/**
* @param topVer New topology version.
* @return Latest topology snapshot.
@@ -493,33 +577,6 @@ class ClientImpl extends TcpDiscoveryImpl {
return allNodes;
}
- /**
- * @param addr Address.
- * @return Remote node ID.
- * @throws IOException In case of I/O error.
- * @throws IgniteCheckedException In case of other error.
- */
- private IgniteBiTuple<Socket, UUID> initConnection(InetSocketAddress addr) throws IOException, IgniteCheckedException {
- assert addr != null;
-
- Socket sock = spi.openSocket(addr);
-
- TcpDiscoveryHandshakeRequest req = new TcpDiscoveryHandshakeRequest(getLocalNodeId());
-
- req.client(true);
-
- spi.writeToSocket(sock, req);
-
- TcpDiscoveryHandshakeResponse res = spi.readMessage(sock, null, spi.ackTimeout);
-
- UUID nodeId = res.creatorNodeId();
-
- assert nodeId != null;
- assert !getLocalNodeId().equals(nodeId);
-
- return F.t(sock, nodeId);
- }
-
/** {@inheritDoc} */
@Override void simulateNodeFailure() {
U.warn(log, "Simulating client node failure: " + getLocalNodeId());
@@ -736,7 +793,7 @@ class ClientImpl extends TcpDiscoveryImpl {
}
/**
- *
+ * @return {@code True} if connection is alive.
*/
public boolean isOnline() {
synchronized (mux) {
@@ -780,7 +837,8 @@ class ClientImpl extends TcpDiscoveryImpl {
}
catch (IOException e) {
if (log.isDebugEnabled())
- U.error(log, "Failed to send node left message (will stop anyway) [sock=" + sock + ']', e);
+ U.error(log, "Failed to send node left message (will stop anyway) " +
+ "[sock=" + sock + ", msg=" + msg + ']', e);
U.closeQuiet(sock);
@@ -909,7 +967,7 @@ class ClientImpl extends TcpDiscoveryImpl {
final Socket sock = joinTopology(false);
if (sock == null) {
- joinErr = new IgniteSpiException("Join process timed out");
+ joinErr = new IgniteSpiException("Join process timed out.");
joinLatch.countDown();
@@ -934,8 +992,9 @@ class ClientImpl extends TcpDiscoveryImpl {
if (msg == JOIN_TIMEOUT) {
if (joinLatch.getCount() > 0) {
- joinErr = new IgniteSpiException("Join process timed out [sock=" + sock +
- ", timeout=" + spi.netTimeout + ']');
+ joinErr = new IgniteSpiException("Join process timed out, did not receive response for " +
+ "join request (consider increasing 'networkTimeout' configuration property) " +
+ "[networkTimeout=" + spi.netTimeout + ", sock=" + sock +']');
joinLatch.countDown();
@@ -1027,7 +1086,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (joinLatch.getCount() > 0) {
// This should not occurs.
- joinErr = new IgniteSpiException("Some error occurs in joinig process");
+ joinErr = new IgniteSpiException("Some error occur in join process.");
joinLatch.countDown();
}
@@ -1236,8 +1295,9 @@ class ClientImpl extends TcpDiscoveryImpl {
if (spi.getSpiContext().isStopping()) {
if (!getLocalNodeId().equals(msg.creatorNodeId()) && getLocalNodeId().equals(msg.failedNodeId())) {
if (leaveLatch.getCount() > 0) {
- log.debug("Remote node fail this node while node is stopping [locNode=" + getLocalNodeId()
- + ", rmtNode=" + msg.creatorNodeId() + ']');
+ if (log.isDebugEnabled())
+ log.debug("Remote node fail this node while node is stopping [locNode=" + getLocalNodeId()
+ + ", rmtNode=" + msg.creatorNodeId() + ']');
leaveLatch.countDown();
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/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 5aceaae..311c783 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
@@ -125,16 +125,6 @@ class ServerImpl extends TcpDiscoveryImpl {
private final ConcurrentMap<InetSocketAddress, IgniteInternalFuture<IgniteBiTuple<UUID, Boolean>>> pingMap =
new ConcurrentHashMap8<>();
- /** Debug mode. */
- private boolean debugMode;
-
- /** Debug messages history. */
- private int debugMsgHist = 512;
-
- /** Received messages. */
- @SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized")
- private ConcurrentLinkedDeque<String> debugLog;
-
/**
* @param adapter Adapter.
*/
@@ -142,24 +132,6 @@ class ServerImpl extends TcpDiscoveryImpl {
super(adapter);
}
- /**
- * This method is intended for troubleshooting purposes only.
- *
- * @param debugMode {code True} to start SPI in debug mode.
- */
- public void setDebugMode(boolean debugMode) {
- this.debugMode = debugMode;
- }
-
- /**
- * This method is intended for troubleshooting purposes only.
- *
- * @param debugMsgHist Message history log size.
- */
- public void setDebugMessageHistory(int debugMsgHist) {
- this.debugMsgHist = debugMsgHist;
- }
-
/** {@inheritDoc} */
@Override public String getSpiState() {
synchronized (mux) {
@@ -1060,23 +1032,6 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/**
- * @param ackTimeout Acknowledgement timeout.
- * @return {@code True} if acknowledgement timeout is less or equal to
- * maximum acknowledgement timeout, {@code false} otherwise.
- */
- private boolean checkAckTimeout(long ackTimeout) {
- if (ackTimeout > spi.maxAckTimeout) {
- LT.warn(log, null, "Acknowledgement timeout is greater than maximum acknowledgement timeout " +
- "(consider increasing 'maxAckTimeout' configuration property) " +
- "[ackTimeout=" + ackTimeout + ", maxAckTimeout=" + spi.maxAckTimeout + ']');
-
- return false;
- }
-
- return true;
- }
-
- /**
* Notify external listener on discovery event.
*
* @param type Discovery event type. See {@link org.apache.ignite.events.DiscoveryEvent} for more details.
@@ -1422,25 +1377,6 @@ class ServerImpl extends TcpDiscoveryImpl {
/**
* @param msg Message.
- */
- private void debugLog(String msg) {
- assert debugMode;
-
- String msg0 = new SimpleDateFormat("[HH:mm:ss,SSS]").format(new Date(System.currentTimeMillis())) +
- '[' + Thread.currentThread().getName() + "][" + getLocalNodeId() +
- "-" + locNode.internalOrder() + "] " +
- msg;
-
- debugLog.add(msg0);
-
- int delta = debugLog.size() - debugMsgHist;
-
- for (int i = 0; i < delta && debugLog.size() > debugMsgHist; i++)
- debugLog.poll();
- }
-
- /**
- * @param msg Message.
* @return {@code True} if recordable in debug mode.
*/
private boolean recordable(TcpDiscoveryAbstractMessage msg) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
index b7e9e53..94097c9 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryImpl.java
@@ -26,7 +26,9 @@ import org.apache.ignite.spi.discovery.*;
import org.apache.ignite.spi.discovery.tcp.internal.*;
import org.jetbrains.annotations.*;
+import java.text.*;
import java.util.*;
+import java.util.concurrent.*;
/**
*
@@ -50,6 +52,16 @@ abstract class TcpDiscoveryImpl {
/** */
protected TcpDiscoveryNode locNode;
+ /** Debug mode. */
+ protected boolean debugMode;
+
+ /** Debug messages history. */
+ private int debugMsgHist = 512;
+
+ /** Received messages. */
+ @SuppressWarnings("FieldAccessedSynchronizedAndUnsynchronized")
+ protected ConcurrentLinkedDeque<String> debugLog;
+
/**
* @param spi Adapter.
*/
@@ -60,6 +72,43 @@ abstract class TcpDiscoveryImpl {
}
/**
+ * This method is intended for troubleshooting purposes only.
+ *
+ * @param debugMode {code True} to start SPI in debug mode.
+ */
+ public void setDebugMode(boolean debugMode) {
+ this.debugMode = debugMode;
+ }
+
+ /**
+ * This method is intended for troubleshooting purposes only.
+ *
+ * @param debugMsgHist Message history log size.
+ */
+ public void setDebugMessageHistory(int debugMsgHist) {
+ this.debugMsgHist = debugMsgHist;
+ }
+
+ /**
+ * @param msg Message.
+ */
+ protected void debugLog(String msg) {
+ assert debugMode;
+
+ String msg0 = new SimpleDateFormat("[HH:mm:ss,SSS]").format(new Date(System.currentTimeMillis())) +
+ '[' + Thread.currentThread().getName() + "][" + getLocalNodeId() +
+ "-" + locNode.internalOrder() + "] " +
+ msg;
+
+ debugLog.add(msg0);
+
+ int delta = debugLog.size() - debugMsgHist;
+
+ for (int i = 0; i < delta && debugLog.size() > debugMsgHist; i++)
+ debugLog.poll();
+ }
+
+ /**
* @return Local node ID.
*/
public UUID getLocalNodeId() {
@@ -209,4 +258,21 @@ abstract class TcpDiscoveryImpl {
}
}
}
+
+ /**
+ * @param ackTimeout Acknowledgement timeout.
+ * @return {@code True} if acknowledgement timeout is less or equal to
+ * maximum acknowledgement timeout, {@code false} otherwise.
+ */
+ protected boolean checkAckTimeout(long ackTimeout) {
+ if (ackTimeout > spi.maxAckTimeout) {
+ LT.warn(log, null, "Acknowledgement timeout is greater than maximum acknowledgement timeout " +
+ "(consider increasing 'maxAckTimeout' configuration property) " +
+ "[ackTimeout=" + ackTimeout + ", maxAckTimeout=" + spi.maxAckTimeout + ']');
+
+ return false;
+ }
+
+ return true;
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/de0d61f6/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
index 24ebb7c..77ddd40 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
@@ -18,15 +18,17 @@
package org.apache.ignite.internal.processors.cache.distributed;
import org.apache.ignite.*;
-import org.apache.ignite.cache.*;
import org.apache.ignite.configuration.*;
import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.apache.ignite.spi.communication.tcp.*;
import org.apache.ignite.spi.discovery.tcp.*;
import org.apache.ignite.spi.discovery.tcp.ipfinder.*;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.*;
import org.apache.ignite.testframework.*;
import org.apache.ignite.testframework.junits.common.*;
+import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
@@ -54,6 +56,12 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
+ cfg.setConnectorConfiguration(null);
+ cfg.setPeerClassLoadingEnabled(false);
+ cfg.setTimeServerPortRange(200);
+
+ ((TcpCommunicationSpi)cfg.getCommunicationSpi()).setLocalPortRange(200);
+
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
if (!clientDiscovery)
@@ -61,6 +69,12 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
cfg.setClientMode(client);
+ if (client) {
+// cfg.setPublicThreadPoolSize(1);
+// cfg.setPeerClassLoadingThreadPoolSize(1);
+// cfg.setIgfsThreadPoolSize(1);
+ }
+
CacheConfiguration ccfg = new CacheConfiguration();
ccfg.setCacheMode(PARTITIONED);
@@ -85,6 +99,11 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
stopAllGrids();
}
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return 10 * 60_000;
+ }
+
/**
* @throws Exception If failed.
*/
@@ -104,6 +123,66 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
/**
* @throws Exception If failed.
*/
+ public void testManyClientsSequentiallyClientDiscovery() throws Exception {
+ clientDiscovery = true;
+
+ manyClientsSequentially();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ private void manyClientsSequentially() throws Exception {
+ client = true;
+
+ List<Ignite> clients = new ArrayList<>();
+
+ final int CLIENTS = 50;
+
+ int idx = SRVS;
+
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
+
+ for (int i = 0; i < CLIENTS; i++) {
+ Ignite ignite = startGrid(idx++);
+
+ log.info("Started node: " + ignite.name());
+
+ assertTrue(ignite.configuration().isClientMode());
+
+ clients.add(ignite);
+
+ IgniteCache<Object, Object> cache = ignite.cache(null);
+
+ Integer key = rnd.nextInt(0, 1000);
+
+ cache.put(key, i);
+
+ assertNotNull(cache.get(key));
+ }
+
+ log.info("All clients started.");
+
+ assertEquals(SRVS + CLIENTS, G.allGrids().size());
+
+ long topVer = -1L;
+
+ for (Ignite ignite : G.allGrids()) {
+ assertEquals(SRVS + CLIENTS, ignite.cluster().nodes().size());
+
+ if (topVer == -1L)
+ topVer = ignite.cluster().topologyVersion();
+ else
+ assertEquals(topVer, ignite.cluster().topologyVersion());
+ }
+
+ for (Ignite client : clients)
+ client.close();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
private void manyClientsPutGet() throws Exception {
client = true;
@@ -111,7 +190,7 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
final AtomicBoolean stop = new AtomicBoolean();
- final int THREADS = 30;
+ final int THREADS = 50;
final CountDownLatch latch = new CountDownLatch(THREADS);
@@ -143,6 +222,8 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
cache.put(key, iter++);
assertNotNull(cache.get(key));
+
+ Thread.sleep(1);
}
log.info("Stopping node: " + ignite.name());
@@ -154,6 +235,8 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
latch.await();
+ log.info("All clients started.");
+
Thread.sleep(10_000);
log.info("Stop clients.");
[5/9] incubator-ignite git commit: Merge remote-tracking branch
'remotes/origin/ignite-883' into ignite-883-1
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/ignite-883' into ignite-883-1
Conflicts:
modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java
modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheCrossCacheQuerySelfTest.java
modules/scalar-2.10/pom.xml
modules/spark-2.10/pom.xml
modules/spark/pom.xml
modules/visor-console-2.10/pom.xml
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/65db72eb
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/65db72eb
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/65db72eb
Branch: refs/heads/ignite-sprint-6
Commit: 65db72eb4a731e4b459ec0686f1741ef4ade13ab
Parents: 5160088 8870a17
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jun 12 17:55:08 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jun 12 17:55:08 2015 +0300
----------------------------------------------------------------------
.../apache/ignite/cache/query/ScanQuery.java | 23 +
.../apache/ignite/internal/IgniteKernal.java | 1 -
.../discovery/GridDiscoveryManager.java | 9 +-
.../dht/preloader/GridDhtPreloader.java | 2 +-
.../org/apache/ignite/spi/IgniteSpiAdapter.java | 28 +-
.../communication/tcp/TcpCommunicationSpi.java | 2 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 439 ++++++++++++-------
.../ignite/spi/discovery/tcp/ServerImpl.java | 223 ++++++----
.../spi/discovery/tcp/TcpDiscoveryImpl.java | 66 +++
.../ipfinder/TcpDiscoveryIpFinderAdapter.java | 34 +-
.../TcpDiscoveryMulticastIpFinder.java | 19 +-
.../messages/TcpDiscoveryAbstractMessage.java | 10 +-
.../distributed/IgniteCacheManyClientsTest.java | 138 +++++-
.../tcp/TcpClientDiscoverySpiSelfTest.java | 73 ++-
.../testsuites/IgniteCacheTestSuite4.java | 2 +
15 files changed, 774 insertions(+), 295 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65db72eb/modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
----------------------------------------------------------------------
diff --cc modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
index e6b69bc,11a8c84..90000e8
--- a/modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
@@@ -99,23 -99,26 +99,46 @@@ public final class ScanQuery<K, V> exte
/**
* Gets partition number over which this query should iterate. Will return {@code null} if partition was not
* set. In this case query will iterate over all partitions in the cache.
+ *
+ * @return Partition number or {@code null}.
+ */
+ @Nullable public Integer getPartition() {
+ return part;
+ }
+
+ /**
+ * Sets partition number over which this query should iterate. If {@code null}, query will iterate over
+ * all partitions in the cache. Must be in the range [0, N) where N is partition number in the cache.
+ *
+ * @param part Partition number over which this query should iterate.
+ * @return {@code this} for chaining.
+ */
+ public ScanQuery<K, V> setPartition(@Nullable Integer part) {
+ this.part = part;
+
+ return this;
+ }
+
++ /**
++ * Gets partition number over which this query should iterate. Will return {@code null} if partition was not
++ * set. In this case query will iterate over all partitions in the cache.
+ *
+ * @return Partition number or {@code null}.
+ */
+ @Nullable public Integer getPartition() {
+ return part;
+ }
+
+ /**
+ * Sets partition number over which this query should iterate. If {@code null}, query will iterate over
+ * all partitions in the cache. Must be in the range [0, N) where N is partition number in the cache.
+ *
+ * @param part Partition number over which this query should iterate.
+ */
+ public void setPartition(@Nullable Integer part) {
+ this.part = part;
+ }
+
/** {@inheritDoc} */
@Override public ScanQuery<K, V> setPageSize(int pageSize) {
return (ScanQuery<K, V>)super.setPageSize(pageSize);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65db72eb/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65db72eb/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65db72eb/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --cc modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index c598e38,ed9fc9a..7fa038c
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@@ -138,9 -140,6 +138,11 @@@ public class IgniteCacheTestSuite4 exte
suite.addTestSuite(IgniteCacheManyClientsTest.class);
+ suite.addTestSuite(IgniteStartCacheInTransactionSelfTest.class);
+ suite.addTestSuite(IgniteStartCacheInTransactionAtomicSelfTest.class);
+
++ suite.addTestSuite(IgniteCacheManyClientsTest.class);
++
return suite;
}
}
[4/9] incubator-ignite git commit: # ignite-883
Posted by sb...@apache.org.
# ignite-883
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/8870a177
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/8870a177
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/8870a177
Branch: refs/heads/ignite-sprint-6
Commit: 8870a177acaa16f6757615cab6017b21aa274c01
Parents: 16f3d32
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jun 12 10:05:22 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jun 12 17:20:14 2015 +0300
----------------------------------------------------------------------
.../communication/tcp/TcpCommunicationSpi.java | 2 +-
.../ignite/spi/discovery/tcp/ClientImpl.java | 232 ++++++++++++-------
.../ignite/spi/discovery/tcp/ServerImpl.java | 170 +++++++++++---
.../ipfinder/TcpDiscoveryIpFinderAdapter.java | 34 ++-
.../TcpDiscoveryMulticastIpFinder.java | 19 +-
.../messages/TcpDiscoveryAbstractMessage.java | 10 +-
.../distributed/IgniteCacheManyClientsTest.java | 65 ++++--
.../tcp/TcpClientDiscoverySpiSelfTest.java | 42 +++-
8 files changed, 412 insertions(+), 162 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8870a177/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git 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
index f19e25b..9e38788 100644
--- 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
@@ -2028,7 +2028,7 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter
if (X.hasCause(e, SocketTimeoutException.class))
LT.warn(log, null, "Connect timed out (consider increasing 'connTimeout' " +
- "configuration property) [addr=" + addr + ", connTimeout" + connTimeout + ']');
+ "configuration property) [addr=" + addr + ", connTimeout=" + connTimeout + ']');
if (errs == null)
errs = new IgniteCheckedException("Failed to connect to node (is node still alive?). " +
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8870a177/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 23e6f88..d8108e5 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
@@ -311,7 +311,7 @@ class ClientImpl extends TcpDiscoveryImpl {
for (ClusterNode n : rmts) {
rmtNodes.remove(n.id());
- Collection<ClusterNode> top = updateTopologyHistory(topVer + 1);
+ Collection<ClusterNode> top = updateTopologyHistory(topVer + 1, null);
lsnr.onDiscovery(EVT_NODE_FAILED, topVer, n, top, new TreeMap<>(topHist), null);
}
@@ -348,13 +348,14 @@ class ClientImpl extends TcpDiscoveryImpl {
/**
* @param recon {@code True} if reconnects.
+ * @param timeout Timeout.
* @return Opened socket or {@code null} if timeout.
* @throws InterruptedException If interrupted.
* @throws IgniteSpiException If failed.
* @see TcpDiscoverySpi#joinTimeout
*/
@SuppressWarnings("BusyWait")
- @Nullable private Socket joinTopology(boolean recon) throws IgniteSpiException, InterruptedException {
+ @Nullable private Socket joinTopology(boolean recon, long timeout) throws IgniteSpiException, InterruptedException {
Collection<InetSocketAddress> addrs = null;
long startTime = U.currentTimeMillis();
@@ -371,11 +372,11 @@ class ClientImpl extends TcpDiscoveryImpl {
log.debug("Resolved addresses from IP finder: " + addrs);
}
else {
- U.warn(log, "No addresses registered in the IP finder (will retry in 2000ms): " + spi.ipFinder);
-
- if (spi.joinTimeout > 0 && (U.currentTimeMillis() - startTime) > spi.joinTimeout)
+ if (timeout > 0 && (U.currentTimeMillis() - startTime) > timeout)
return null;
+ U.warn(log, "No addresses registered in the IP finder (will retry in 2000ms): " + spi.ipFinder);
+
Thread.sleep(2000);
}
}
@@ -421,13 +422,13 @@ class ClientImpl extends TcpDiscoveryImpl {
}
if (addrs.isEmpty()) {
- if (spi.joinTimeout > 0 && (U.currentTimeMillis() - startTime) > spi.joinTimeout)
+ if (timeout > 0 && (U.currentTimeMillis() - startTime) > timeout)
return null;
- Thread.sleep(2000);
-
U.warn(log, "Failed to connect to any address from IP finder (will retry to join topology " +
"in 2000ms): " + addrs0);
+
+ Thread.sleep(2000);
}
}
}
@@ -541,16 +542,17 @@ class ClientImpl extends TcpDiscoveryImpl {
/**
* @param topVer New topology version.
+ * @param msg Discovery message.
* @return Latest topology snapshot.
*/
- private NavigableSet<ClusterNode> updateTopologyHistory(long topVer) {
+ private NavigableSet<ClusterNode> updateTopologyHistory(long topVer, @Nullable TcpDiscoveryAbstractMessage msg) {
this.topVer = topVer;
NavigableSet<ClusterNode> allNodes = allVisibleNodes();
if (!topHist.containsKey(topVer)) {
assert topHist.isEmpty() || topHist.lastKey() == topVer - 1 :
- "lastVer=" + topHist.lastKey() + ", newVer=" + topVer;
+ "lastVer=" + topHist.lastKey() + ", newVer=" + topVer + ", locNode=" + locNode + ", msg=" + msg;
topHist.put(topVer, allNodes);
@@ -619,6 +621,17 @@ class ClientImpl extends TcpDiscoveryImpl {
}
/**
+ * @param err Error.
+ */
+ private void joinError(IgniteSpiException err) {
+ assert err != null;
+
+ joinErr = err;
+
+ joinLatch.countDown();
+ }
+
+ /**
* Heartbeat sender.
*/
private class HeartbeatSender extends TimerTask {
@@ -727,7 +740,7 @@ class ClientImpl extends TcpDiscoveryImpl {
spi.stats.onMessageReceived(msg);
- if (spi.ensured(msg))
+ if (spi.ensured(msg) && joinLatch.getCount() == 0L)
lastMsgId = msg.id();
msgWorker.addMessage(msg);
@@ -866,11 +879,16 @@ class ClientImpl extends TcpDiscoveryImpl {
/** */
private volatile Socket sock;
+ /** */
+ private boolean join;
+
/**
- *
+ * @param join {@code True} if reconnects during join.
*/
- protected Reconnector() {
+ protected Reconnector(boolean join) {
super(spi.ignite().name(), "tcp-client-disco-msg-worker", log);
+
+ this.join = join;
}
/**
@@ -888,51 +906,94 @@ class ClientImpl extends TcpDiscoveryImpl {
boolean success = false;
+ Exception err = null;
+
+ long timeout = join ? spi.joinTimeout : spi.netTimeout;
+
+ long startTime = U.currentTimeMillis();
+
try {
- sock = joinTopology(true);
+ while (true) {
+ sock = joinTopology(true, timeout);
- if (sock == null) {
- U.error(log, "Failed to reconnect to cluster: timeout.");
+ if (sock == null) {
+ 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 + ']'));
+ }
+ else
+ U.error(log, "Failed to reconnect to cluster (consider increasing 'networkTimeout' " +
+ "configuration property) [networkTimeout=" + spi.netTimeout + ", sock=" + sock + ']');
- return;
- }
+ return;
+ }
- if (isInterrupted())
- throw new InterruptedException();
+ if (isInterrupted())
+ throw new InterruptedException();
- InputStream in = new BufferedInputStream(sock.getInputStream());
+ int oldTimeout = 0;
- sock.setKeepAlive(true);
- sock.setTcpNoDelay(true);
+ try {
+ oldTimeout = sock.getSoTimeout();
- // Wait for
- while (!isInterrupted()) {
- TcpDiscoveryAbstractMessage msg = spi.marsh.unmarshal(in, U.gridClassLoader());
+ sock.setSoTimeout((int)spi.netTimeout);
- if (msg instanceof TcpDiscoveryClientReconnectMessage) {
- TcpDiscoveryClientReconnectMessage res = (TcpDiscoveryClientReconnectMessage)msg;
+ InputStream in = new BufferedInputStream(sock.getInputStream());
- if (res.creatorNodeId().equals(getLocalNodeId())) {
- if (res.success()) {
- msgWorker.addMessage(res);
+ sock.setKeepAlive(true);
+ sock.setTcpNoDelay(true);
- success = true;
- }
+ // Wait for
+ while (!isInterrupted()) {
+ TcpDiscoveryAbstractMessage msg = spi.marsh.unmarshal(in, U.gridClassLoader());
- break;
+ if (msg instanceof TcpDiscoveryClientReconnectMessage) {
+ TcpDiscoveryClientReconnectMessage res = (TcpDiscoveryClientReconnectMessage)msg;
+
+ if (res.creatorNodeId().equals(getLocalNodeId())) {
+ if (res.success()) {
+ msgWorker.addMessage(res);
+
+ success = true;
+ }
+
+ return;
+ }
+ }
}
}
+ catch (IOException | IgniteCheckedException e) {
+ U.closeQuiet(sock);
+
+ if (log.isDebugEnabled())
+ log.error("Reconnect error [join=" + join + ", timeout=" + timeout + ']', e);
+ if (timeout > 0 && (U.currentTimeMillis() - startTime) > timeout)
+ throw e;
+ else
+ U.warn(log, "Failed to reconnect to cluster (will retry): " + e);
+ }
+ finally {
+ if (success)
+ sock.setSoTimeout(oldTimeout);
+ }
}
}
catch (IOException | IgniteCheckedException e) {
+ err = e;
+
U.error(log, "Failed to reconnect", e);
}
finally {
if (!success) {
U.closeQuiet(sock);
- msgWorker.addMessage(SPI_RECONNECT_FAILED);
+ if (join)
+ joinError(new IgniteSpiException("Failed to connect to cluster, connection failed and failed " +
+ "to reconnect.", err));
+ else
+ msgWorker.addMessage(SPI_RECONNECT_FAILED);
}
}
}
@@ -967,7 +1028,7 @@ class ClientImpl extends TcpDiscoveryImpl {
spi.stats.onJoinStarted();
try {
- final Socket sock = joinTopology(false);
+ final Socket sock = joinTopology(false, spi.joinTimeout);
if (sock == null) {
joinErr = new IgniteSpiException("Join process timed out.");
@@ -981,12 +1042,14 @@ class ClientImpl extends TcpDiscoveryImpl {
sockWriter.setSocket(sock);
- timer.schedule(new TimerTask() {
- @Override public void run() {
- if (joinLatch.getCount() > 0)
- queue.add(JOIN_TIMEOUT);
- }
- }, spi.netTimeout);
+ if (spi.joinTimeout > 0) {
+ timer.schedule(new TimerTask() {
+ @Override public void run() {
+ if (joinLatch.getCount() > 0)
+ queue.add(JOIN_TIMEOUT);
+ }
+ }, spi.joinTimeout);
+ }
sockReader.setSocket(sock, locNode.clientRouterNodeId());
@@ -996,8 +1059,8 @@ class ClientImpl extends TcpDiscoveryImpl {
if (msg == JOIN_TIMEOUT) {
if (joinLatch.getCount() > 0) {
joinErr = new IgniteSpiException("Join process timed out, did not receive response for " +
- "join request (consider increasing 'networkTimeout' configuration property) " +
- "[networkTimeout=" + spi.netTimeout + ", sock=" + sock +']');
+ "join request (consider increasing 'joinTimeout' configuration property) " +
+ "[joinTimeout=" + spi.joinTimeout + ", sock=" + sock +']');
joinLatch.countDown();
@@ -1021,30 +1084,23 @@ class ClientImpl extends TcpDiscoveryImpl {
if (((SocketClosedMessage)msg).sock == currSock) {
currSock = null;
- if (joinLatch.getCount() > 0) {
- joinErr = new IgniteSpiException("Failed to connect to cluster: socket closed.");
+ boolean join = joinLatch.getCount() > 0;
- joinLatch.countDown();
+ if (spi.getSpiContext().isStopping() || segmented) {
+ leaveLatch.countDown();
- break;
+ if (join) {
+ joinError(new IgniteSpiException("Failed to connect to cluster: socket closed."));
+
+ break;
+ }
}
else {
- if (spi.getSpiContext().isStopping() || segmented)
- leaveLatch.countDown();
- else {
- assert reconnector == null;
-
- final Reconnector reconnector = new Reconnector();
- this.reconnector = reconnector;
- reconnector.start();
-
- timer.schedule(new TimerTask() {
- @Override public void run() {
- if (reconnector.isAlive())
- reconnector.cancel();
- }
- }, spi.netTimeout);
- }
+ assert reconnector == null;
+
+ final Reconnector reconnector = new Reconnector(join);
+ this.reconnector = reconnector;
+ reconnector.start();
}
}
}
@@ -1208,7 +1264,7 @@ class ClientImpl extends TcpDiscoveryImpl {
locNode.order(topVer);
- notifyDiscovery(EVT_NODE_JOINED, topVer, locNode, updateTopologyHistory(topVer));
+ notifyDiscovery(EVT_NODE_JOINED, topVer, locNode, updateTopologyHistory(topVer, msg));
joinErr = null;
@@ -1230,6 +1286,14 @@ class ClientImpl extends TcpDiscoveryImpl {
return;
}
+ if (!topHist.isEmpty() && msg.topologyVersion() <= topHist.lastKey()) {
+ if (log.isDebugEnabled())
+ log.debug("Discarding node add finished message since topology already updated " +
+ "[msg=" + msg + ", lastHistKey=" + topHist.lastKey() + ", node=" + node + ']');
+
+ return;
+ }
+
long topVer = msg.topologyVersion();
node.order(topVer);
@@ -1238,7 +1302,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (spi.locNodeVer.equals(node.version()))
node.version(spi.locNodeVer);
- NavigableSet<ClusterNode> top = updateTopologyHistory(topVer);
+ NavigableSet<ClusterNode> top = updateTopologyHistory(topVer, msg);
if (!pending && joinLatch.getCount() > 0) {
if (log.isDebugEnabled())
@@ -1276,7 +1340,7 @@ class ClientImpl extends TcpDiscoveryImpl {
return;
}
- NavigableSet<ClusterNode> top = updateTopologyHistory(msg.topologyVersion());
+ NavigableSet<ClusterNode> top = updateTopologyHistory(msg.topologyVersion(), msg);
if (!pending && joinLatch.getCount() > 0) {
if (log.isDebugEnabled())
@@ -1319,7 +1383,7 @@ class ClientImpl extends TcpDiscoveryImpl {
return;
}
- NavigableSet<ClusterNode> top = updateTopologyHistory(msg.topologyVersion());
+ NavigableSet<ClusterNode> top = updateTopologyHistory(msg.topologyVersion(), msg);
if (!pending && joinLatch.getCount() > 0) {
if (log.isDebugEnabled())
@@ -1376,28 +1440,32 @@ class ClientImpl extends TcpDiscoveryImpl {
return;
if (getLocalNodeId().equals(msg.creatorNodeId())) {
- assert msg.success();
+ assert msg.success() : msg;
- currSock = reconnector.sock;
+ if (reconnector != null) {
+ currSock = reconnector.sock;
- sockWriter.setSocket(currSock);
- sockReader.setSocket(currSock, locNode.clientRouterNodeId());
+ sockWriter.setSocket(currSock);
+ sockReader.setSocket(currSock, locNode.clientRouterNodeId());
- reconnector = null;
+ reconnector = null;
- pending = true;
+ pending = true;
- try {
- for (TcpDiscoveryAbstractMessage pendingMsg : msg.pendingMessages()) {
- if (log.isDebugEnabled())
- log.debug("Process message on reconnect [msg=" + pendingMsg + ']');
+ try {
+ for (TcpDiscoveryAbstractMessage pendingMsg : msg.pendingMessages()) {
+ if (log.isDebugEnabled())
+ log.debug("Process message on reconnect [msg=" + pendingMsg + ']');
- processDiscoveryMessage(pendingMsg);
+ processDiscoveryMessage(pendingMsg);
+ }
+ }
+ finally {
+ pending = false;
}
}
- finally {
- pending = false;
- }
+ else if (log.isDebugEnabled())
+ log.debug("Discarding reconnect message, reconnect is completed: " + msg);
}
else if (log.isDebugEnabled())
log.debug("Discarding reconnect message for another client: " + msg);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8870a177/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 65bea9f..9041557 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
@@ -41,7 +41,6 @@ import org.jsr166.*;
import java.io.*;
import java.net.*;
-import java.text.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
@@ -1192,7 +1191,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (node.id().equals(destNodeId)) {
Collection<TcpDiscoveryNode> allNodes = ring.allNodes();
- Collection<TcpDiscoveryNode> topToSend = new ArrayList<>(allNodes.size());
+ Collection<TcpDiscoveryNode> topToSnd = new ArrayList<>(allNodes.size());
for (TcpDiscoveryNode n0 : allNodes) {
assert n0.internalOrder() != 0 : n0;
@@ -1202,10 +1201,10 @@ class ServerImpl extends TcpDiscoveryImpl {
// There will be separate messages for nodes with greater
// internal order.
if (n0.internalOrder() < nodeAddedMsg.node().internalOrder())
- topToSend.add(n0);
+ topToSnd.add(n0);
}
- nodeAddedMsg.topology(topToSend);
+ nodeAddedMsg.topology(topToSnd);
nodeAddedMsg.messages(msgs, discardMsgId);
Map<Long, Collection<ClusterNode>> hist;
@@ -1646,6 +1645,108 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/**
+ * Discovery messages history used for client reconnect.
+ */
+ private class EnsuredMessageHistory {
+ /** */
+ private static final int MAX = 1024;
+
+ /** Pending messages. */
+ private final ArrayDeque<TcpDiscoveryAbstractMessage> msgs = new ArrayDeque<>(MAX * 2);
+
+ /**
+ * @param msg Adds message.
+ */
+ void add(TcpDiscoveryAbstractMessage msg) {
+ assert spi.ensured(msg) : msg;
+
+ msgs.addLast(msg);
+
+ while (msgs.size() > MAX)
+ msgs.pollFirst();
+ }
+
+ /**
+ * Gets messages starting from provided ID (exclusive). If such
+ * message is not found, {@code null} is returned (this indicates
+ * a failure condition when it was already removed from queue).
+ *
+ * @param lastMsgId Last message ID received on client. {@code Null} if client did not finish connect procedure.
+ * @param node Client node.
+ * @return Collection of messages.
+ */
+ @Nullable Collection<TcpDiscoveryAbstractMessage> messages(@Nullable IgniteUuid lastMsgId,
+ TcpDiscoveryNode node)
+ {
+ assert node != null && node.isClient() : node;
+
+ if (lastMsgId == null) {
+ // Client connection failed before it received TcpDiscoveryNodeAddedMessage.
+ List<TcpDiscoveryAbstractMessage> res = null;
+
+ for (TcpDiscoveryAbstractMessage msg : msgs) {
+ if (msg instanceof TcpDiscoveryNodeAddedMessage) {
+ if (node.id().equals(((TcpDiscoveryNodeAddedMessage) msg).node().id()))
+ res = new ArrayList<>(msgs.size());
+ }
+
+ if (res != null)
+ res.add(prepare(msg, node.id()));
+ }
+
+ if (log.isDebugEnabled()) {
+ if (res == null)
+ log.debug("Failed to find node added message [node=" + node + ']');
+ else
+ log.debug("Found add added message [node=" + node + ", hist=" + res + ']');
+ }
+
+ return res;
+ }
+ else {
+ if (msgs.isEmpty())
+ return Collections.emptyList();
+
+ Collection<TcpDiscoveryAbstractMessage> cp = new ArrayList<>(msgs.size());
+
+ boolean skip = true;
+
+ for (TcpDiscoveryAbstractMessage msg : msgs) {
+ if (skip) {
+ if (msg.id().equals(lastMsgId))
+ skip = false;
+ }
+ else
+ cp.add(prepare(msg, node.id()));
+ }
+
+ cp = !skip ? cp : null;
+
+ if (log.isDebugEnabled()) {
+ if (cp == null)
+ log.debug("Failed to find messages history [node=" + node + ", lastMsgId" + lastMsgId + ']');
+ else
+ log.debug("Found messages history [node=" + node + ", hist=" + cp + ']');
+ }
+
+ return cp;
+ }
+ }
+
+ /**
+ * @param msg Message.
+ * @param destNodeId Client node ID.
+ * @return Prepared message.
+ */
+ private TcpDiscoveryAbstractMessage prepare(TcpDiscoveryAbstractMessage msg, UUID destNodeId) {
+ if (msg instanceof TcpDiscoveryNodeAddedMessage)
+ prepareNodeAddedMessage(msg, destNodeId, null, null);
+
+ return msg;
+ }
+ }
+
+ /**
* Pending messages container.
*/
private static class PendingMessages {
@@ -1678,33 +1779,27 @@ class ServerImpl extends TcpDiscoveryImpl {
}
/**
- * Gets messages starting from provided ID (exclusive). If such
- * message is not found, {@code null} is returned (this indicates
- * a failure condition when it was already removed from queue).
+ * Resets pending messages.
*
- * @param lastMsgId Last message ID.
- * @return Collection of messages.
+ * @param msgs Message.
+ * @param discardId Discarded message ID.
*/
- @Nullable Collection<TcpDiscoveryAbstractMessage> messages(IgniteUuid lastMsgId) {
- assert lastMsgId != null;
-
- if (msgs.isEmpty())
- return Collections.emptyList();
+ void reset(@Nullable Collection<TcpDiscoveryAbstractMessage> msgs, @Nullable IgniteUuid discardId) {
+ this.msgs.clear();
- Collection<TcpDiscoveryAbstractMessage> cp = new ArrayList<>(msgs.size());
+ if (msgs != null)
+ this.msgs.addAll(msgs);
- boolean skip = true;
+ this.discardId = discardId;
+ }
- for (TcpDiscoveryAbstractMessage msg : msgs) {
- if (skip) {
- if (msg.id().equals(lastMsgId))
- skip = false;
- }
- else
- cp.add(msg);
- }
+ /**
+ * Clears pending messages.
+ */
+ void clear() {
+ msgs.clear();
- return !skip ? cp : null;
+ discardId = null;
}
/**
@@ -1728,6 +1823,9 @@ class ServerImpl extends TcpDiscoveryImpl {
/** Pending messages. */
private final PendingMessages pendingMsgs = new PendingMessages();
+ /** Messages history used for client reconnect. */
+ private final EnsuredMessageHistory msgHist = new EnsuredMessageHistory();
+
/** Last message that updated topology. */
private TcpDiscoveryAbstractMessage lastMsg;
@@ -1794,6 +1892,9 @@ class ServerImpl extends TcpDiscoveryImpl {
else
assert false : "Unknown message type: " + msg.getClass().getSimpleName();
+ if (spi.ensured(msg))
+ msgHist.add(msg);
+
spi.stats.onMessageProcessingFinished(msg);
}
@@ -2130,6 +2231,8 @@ class ServerImpl extends TcpDiscoveryImpl {
onException("Failed to send message to next node [next=" + next.id() + ", msg=" + msg + ']',
e);
+ log.error("Will resend [msg=" + msg + ", e=" + e + ']');
+
if (e instanceof SocketTimeoutException || X.hasCause(e, SocketTimeoutException.class)) {
ackTimeout0 *= 2;
@@ -2619,11 +2722,15 @@ class ServerImpl extends TcpDiscoveryImpl {
node.aliveCheck(spi.maxMissedClientHbs);
if (isLocalNodeCoordinator()) {
- Collection<TcpDiscoveryAbstractMessage> pending = pendingMsgs.messages(msg.lastMessageId());
+ Collection<TcpDiscoveryAbstractMessage> pending = msgHist.messages(msg.lastMessageId(), node);
if (pending != null) {
msg.pendingMessages(pending);
msg.success(true);
+
+ if (log.isDebugEnabled())
+ log.debug("Accept client reconnect, restored pending messages " +
+ "[locNodeId=" + locNodeId + ", clientNodeId=" + nodeId + ']');
}
else {
if (log.isDebugEnabled())
@@ -2836,7 +2943,8 @@ class ServerImpl extends TcpDiscoveryImpl {
topHist.clear();
topHist.putAll(msg.topologyHistory());
- pendingMsgs.discard(msg.discardedMessageId());
+ // Restore pending messages.
+ pendingMsgs.reset(msg.messages(), msg.discardedMessageId());
// Clear data to minimize message size.
msg.messages(null, null);
@@ -3094,6 +3202,10 @@ class ServerImpl extends TcpDiscoveryImpl {
if (log.isDebugEnabled())
log.debug("Removed node from topology: " + leftNode);
+ // Clear pending messages map.
+ if (!ring.hasRemoteNodes())
+ pendingMsgs.clear();
+
long topVer;
if (locNodeCoord) {
@@ -3257,6 +3369,10 @@ class ServerImpl extends TcpDiscoveryImpl {
assert node != null;
+ // Clear pending messages map.
+ if (!ring.hasRemoteNodes())
+ pendingMsgs.clear();
+
long topVer;
if (locNodeCoord) {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8870a177/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinderAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinderAdapter.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinderAdapter.java
index 99a2cdc..4d62ff2 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinderAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/TcpDiscoveryIpFinderAdapter.java
@@ -17,9 +17,13 @@
package org.apache.ignite.spi.discovery.tcp.ipfinder;
+import org.apache.ignite.*;
import org.apache.ignite.internal.util.tostring.*;
import org.apache.ignite.internal.util.typedef.internal.*;
+import org.apache.ignite.resources.*;
import org.apache.ignite.spi.*;
+import org.apache.ignite.spi.discovery.*;
+import org.apache.ignite.spi.discovery.tcp.*;
import java.net.*;
import java.util.*;
@@ -35,6 +39,11 @@ public abstract class TcpDiscoveryIpFinderAdapter implements TcpDiscoveryIpFinde
@GridToStringExclude
private volatile IgniteSpiContext spiCtx;
+ /** Ignite instance . */
+ @IgniteInstanceResource
+ @GridToStringExclude
+ protected Ignite ignite;
+
/** {@inheritDoc} */
@Override public void onSpiContextInitialized(IgniteSpiContext spiCtx) throws IgniteSpiException {
this.spiCtx = spiCtx;
@@ -47,7 +56,8 @@ public abstract class TcpDiscoveryIpFinderAdapter implements TcpDiscoveryIpFinde
/** {@inheritDoc} */
@Override public void initializeLocalAddresses(Collection<InetSocketAddress> addrs) throws IgniteSpiException {
- registerAddresses(addrs);
+ if (!discoveryClientMode())
+ registerAddresses(addrs);
}
/** {@inheritDoc} */
@@ -77,6 +87,28 @@ public abstract class TcpDiscoveryIpFinderAdapter implements TcpDiscoveryIpFinde
}
/**
+ * @return {@code True} if TCP discovery works in client mode.
+ */
+ protected boolean discoveryClientMode() {
+ boolean clientMode;
+
+ Ignite ignite0 = ignite;
+
+ if (ignite0 != null) { // Can be null if used in tests without starting Ignite.
+ DiscoverySpi discoSpi = ignite0.configuration().getDiscoverySpi();
+
+ if (!(discoSpi instanceof TcpDiscoverySpi))
+ throw new IgniteSpiException("TcpDiscoveryIpFinder should be used with TcpDiscoverySpi: " + discoSpi);
+
+ clientMode = ignite0.configuration().isClientMode() && !((TcpDiscoverySpi)discoSpi).isForceServerMode();
+ }
+ else
+ clientMode = false;
+
+ return clientMode;
+ }
+
+ /**
* @return SPI context.
*/
protected IgniteSpiContext spiContext() {
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8870a177/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
index a992620..8e5a1fd 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ipfinder/multicast/TcpDiscoveryMulticastIpFinder.java
@@ -85,11 +85,6 @@ public class TcpDiscoveryMulticastIpFinder extends TcpDiscoveryVmIpFinder {
@LoggerResource
private IgniteLogger log;
- /** Ignite instance . */
- @IgniteInstanceResource
- @GridToStringExclude
- private Ignite ignite;
-
/** Multicast IP address as string. */
private String mcastGrp = DFLT_MCAST_GROUP;
@@ -256,19 +251,7 @@ public class TcpDiscoveryMulticastIpFinder extends TcpDiscoveryVmIpFinder {
"(it is recommended in production to specify at least one address in " +
"TcpDiscoveryMulticastIpFinder.getAddresses() configuration property)");
- boolean clientMode;
-
- if (ignite != null) { // Can be null if used in tests without starting Ignite.
- DiscoverySpi discoSpi = ignite.configuration().getDiscoverySpi();
-
- if (!(discoSpi instanceof TcpDiscoverySpi))
- throw new IgniteSpiException("TcpDiscoveryMulticastIpFinder should be used with " +
- "TcpDiscoverySpi: " + discoSpi);
-
- clientMode = ((TcpDiscoverySpi)discoSpi).isClientMode();
- }
- else
- clientMode = false;
+ boolean clientMode = discoveryClientMode();
InetAddress mcastAddr;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8870a177/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
index 145b518..21dbf4f 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractMessage.java
@@ -41,7 +41,7 @@ public abstract class TcpDiscoveryAbstractMessage implements Serializable {
protected static final int CLIENT_RECON_SUCCESS_FLAG_POS = 2;
/** Sender of the message (transient). */
- private transient UUID senderNodeId;
+ private transient UUID sndNodeId;
/** Message ID. */
private IgniteUuid id;
@@ -99,16 +99,16 @@ public abstract class TcpDiscoveryAbstractMessage implements Serializable {
* @return Sender node ID.
*/
public UUID senderNodeId() {
- return senderNodeId;
+ return sndNodeId;
}
/**
* Sets sender node ID.
*
- * @param senderNodeId Sender node ID.
+ * @param sndNodeId Sender node ID.
*/
- public void senderNodeId(UUID senderNodeId) {
- this.senderNodeId = senderNodeId;
+ public void senderNodeId(UUID sndNodeId) {
+ this.sndNodeId = sndNodeId;
}
/**
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8870a177/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
index 77ddd40..4fb4387 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
@@ -63,18 +63,13 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
((TcpCommunicationSpi)cfg.getCommunicationSpi()).setLocalPortRange(200);
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setIpFinder(ipFinder);
+ ((TcpDiscoverySpi)cfg.getDiscoverySpi()).setJoinTimeout(2 * 60_000);
if (!clientDiscovery)
((TcpDiscoverySpi)cfg.getDiscoverySpi()).setForceServerMode(true);
cfg.setClientMode(client);
- if (client) {
-// cfg.setPublicThreadPoolSize(1);
-// cfg.setPeerClassLoadingThreadPoolSize(1);
-// cfg.setIgfsThreadPoolSize(1);
- }
-
CacheConfiguration ccfg = new CacheConfiguration();
ccfg.setCacheMode(PARTITIONED);
@@ -197,43 +192,62 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
try {
IgniteInternalFuture<?> fut = GridTestUtils.runMultiThreadedAsync(new Callable<Object>() {
@Override public Object call() throws Exception {
- try (Ignite ignite = startGrid(idx.getAndIncrement())) {
- log.info("Started node: " + ignite.name());
+ boolean counted = false;
- assertTrue(ignite.configuration().isClientMode());
+ try {
+ int nodeIdx = idx.getAndIncrement();
- IgniteCache<Object, Object> cache = ignite.cache(null);
+ Thread.currentThread().setName("client-thread-node-" + nodeIdx);
- ThreadLocalRandom rnd = ThreadLocalRandom.current();
+ try (Ignite ignite = startGrid(nodeIdx)) {
+ log.info("Started node: " + ignite.name());
- int iter = 0;
+ assertTrue(ignite.configuration().isClientMode());
- Integer key = rnd.nextInt(0, 1000);
+ IgniteCache<Object, Object> cache = ignite.cache(null);
- cache.put(key, iter++);
+ ThreadLocalRandom rnd = ThreadLocalRandom.current();
- assertNotNull(cache.get(key));
+ int iter = 0;
- latch.countDown();
-
- while (!stop.get()) {
- key = rnd.nextInt(0, 1000);
+ Integer key = rnd.nextInt(0, 1000);
cache.put(key, iter++);
assertNotNull(cache.get(key));
- Thread.sleep(1);
+ latch.countDown();
+
+ counted = true;
+
+ while (!stop.get()) {
+ key = rnd.nextInt(0, 1000);
+
+ cache.put(key, iter++);
+
+ assertNotNull(cache.get(key));
+
+ Thread.sleep(1);
+ }
+
+ log.info("Stopping node: " + ignite.name());
}
- log.info("Stopping node: " + ignite.name());
+ return null;
}
+ catch (Throwable e) {
+ log.error("Unexpected error in client thread: " + e, e);
- return null;
+ throw e;
+ }
+ finally {
+ if (!counted)
+ latch.countDown();
+ }
}
}, THREADS, "client-thread");
- latch.await();
+ assertTrue(latch.await(getTestTimeout(), TimeUnit.MILLISECONDS));
log.info("All clients started.");
@@ -245,6 +259,11 @@ public class IgniteCacheManyClientsTest extends GridCommonAbstractTest {
fut.get();
}
+ catch (Throwable e) {
+ log.error("Unexpected error: " + e, e);
+
+ throw e;
+ }
finally {
stop.set(true);
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/8870a177/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
index 44fe299..8147958 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
@@ -193,8 +193,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
- *
- * @throws Exception
+ * @throws Exception If failed.
*/
public void testJoinTimeout() throws Exception {
clientIpFinder = new TcpDiscoveryVmIpFinder();
@@ -544,8 +543,6 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testClientReconnectTopologyChange2() throws Exception {
- fail("https://issues.apache.org/jira/browse/IGNITE-998");
-
maxMissedClientHbs = 100;
clientsPerSrv = 1;
@@ -1001,6 +998,24 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
+ * @throws Exception If failed.
+ */
+ public void testJoinError2() throws Exception {
+ startServerNodes(1);
+
+ Ignite ignite = G.ignite("server-0");
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)ignite.configuration().getDiscoverySpi());
+
+ srvSpi.failNodeAddedMessage();
+ srvSpi.failClientReconnectMessage();
+
+ startClientNodes(1);
+
+ checkNodes(1, 1);
+ }
+
+ /**
* @param clientIdx Client index.
* @param srvIdx Server index.
* @throws Exception In case of error.
@@ -1251,6 +1266,9 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
/** */
private AtomicInteger failNodeAdded = new AtomicInteger();
+ /** */
+ private AtomicInteger failClientReconnect = new AtomicInteger();
+
/**
* @param lock Lock.
*/
@@ -1276,6 +1294,13 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
+ *
+ */
+ void failClientReconnectMessage() {
+ failClientReconnect.set(1);
+ }
+
+ /**
* @param isPause Is lock.
* @param locks Locks.
*/
@@ -1293,7 +1318,14 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
GridByteArrayOutputStream bout) throws IOException, IgniteCheckedException {
waitFor(writeLock);
- if (msg instanceof TcpDiscoveryNodeAddedMessage && failNodeAdded.getAndDecrement() > 0) {
+ boolean fail = false;
+
+ if (msg instanceof TcpDiscoveryNodeAddedMessage)
+ fail = failNodeAdded.getAndDecrement() > 0;
+ else if (msg instanceof TcpDiscoveryClientReconnectMessage)
+ fail = failClientReconnect.getAndDecrement() > 0;
+
+ if (fail) {
log.info("Close socket on message write [msg=" + msg + "]");
sock.close();
[6/9] incubator-ignite git commit: Merge remote-tracking branch
'remotes/origin/ignite-883' into ignite-883-1
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/ignite-883' into ignite-883-1
Conflicts:
modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/GridCacheQueryManager.java
modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteCacheManyClientsTest.java
modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/GridCacheSwapScanQueryAbstractSelfTest.java
modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheCrossCacheQuerySelfTest.java
modules/scalar-2.10/pom.xml
modules/spark-2.10/pom.xml
modules/spark/pom.xml
modules/visor-console-2.10/pom.xml
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/f115c048
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/f115c048
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/f115c048
Branch: refs/heads/ignite-sprint-6
Commit: f115c048083cd36edd31b589a9c3389322e0220e
Parents: 65db72e
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jun 12 18:02:31 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jun 12 18:02:31 2015 +0300
----------------------------------------------------------------------
.../apache/ignite/cache/query/ScanQuery.java | 20 --------------------
.../ignite/spi/discovery/tcp/ClientImpl.java | 2 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 2 --
.../testsuites/IgniteCacheTestSuite4.java | 2 --
4 files changed, 1 insertion(+), 25 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f115c048/modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java b/modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
index 90000e8..6e02ff9 100644
--- a/modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
+++ b/modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
@@ -97,16 +97,6 @@ public final class ScanQuery<K, V> extends Query<Cache.Entry<K, V>> {
}
/**
- * Gets partition number over which this query should iterate. Will return {@code null} if partition was not
- * set. In this case query will iterate over all partitions in the cache.
- *
- * @return Partition number or {@code null}.
- */
- @Nullable public Integer getPartition() {
- return part;
- }
-
- /**
* Sets partition number over which this query should iterate. If {@code null}, query will iterate over
* all partitions in the cache. Must be in the range [0, N) where N is partition number in the cache.
*
@@ -129,16 +119,6 @@ public final class ScanQuery<K, V> extends Query<Cache.Entry<K, V>> {
return part;
}
- /**
- * Sets partition number over which this query should iterate. If {@code null}, query will iterate over
- * all partitions in the cache. Must be in the range [0, N) where N is partition number in the cache.
- *
- * @param part Partition number over which this query should iterate.
- */
- public void setPartition(@Nullable Integer part) {
- this.part = part;
- }
-
/** {@inheritDoc} */
@Override public ScanQuery<K, V> setPageSize(int pageSize) {
return (ScanQuery<K, V>)super.setPageSize(pageSize);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f115c048/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 d8108e5..a17296c 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
@@ -1145,7 +1145,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (joinLatch.getCount() > 0) {
// This should not occurs.
- joinErr = new IgniteSpiException("Some error occur in join process.");
+ joinErr = new IgniteSpiException("Some error in join process.");
joinLatch.countDown();
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f115c048/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 c92fcde..63f165d 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
@@ -2231,8 +2231,6 @@ class ServerImpl extends TcpDiscoveryImpl {
onException("Failed to send message to next node [next=" + next.id() + ", msg=" + msg + ']',
e);
- log.error("Will resend [msg=" + msg + ", e=" + e + ']');
-
if (e instanceof SocketTimeoutException || X.hasCause(e, SocketTimeoutException.class)) {
ackTimeout0 *= 2;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/f115c048/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index 7fa038c..fed5efe 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -136,8 +136,6 @@ public class IgniteCacheTestSuite4 extends TestSuite {
suite.addTestSuite(CacheReadOnlyTransactionalClientSelfTest.class);
- suite.addTestSuite(IgniteCacheManyClientsTest.class);
-
suite.addTestSuite(IgniteStartCacheInTransactionSelfTest.class);
suite.addTestSuite(IgniteStartCacheInTransactionAtomicSelfTest.class);
[2/9] incubator-ignite git commit: Merge remote-tracking branch
'remotes/origin/ignite-sprint-5' into ignite-883
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/ignite-sprint-5' into ignite-883
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/58fe3edf
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/58fe3edf
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/58fe3edf
Branch: refs/heads/ignite-sprint-6
Commit: 58fe3edfdb68fea7b045b3158dbfab9f9228cae7
Parents: de0d61f 4375529
Author: sboikov <sb...@gridgain.com>
Authored: Thu Jun 11 17:42:00 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Thu Jun 11 17:42:00 2015 +0300
----------------------------------------------------------------------
idea/ignite_codeStyle.xml | 147 +++++++++++++++++++
.../apache/ignite/cache/query/ScanQuery.java | 5 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 38 +----
.../tcp/TcpClientDiscoverySpiSelfTest.java | 114 +++++++++++++-
parent/pom.xml | 1 +
5 files changed, 267 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/58fe3edf/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
[8/9] incubator-ignite git commit: Merge branches 'ignite-883-1' and
'ignite-sprint-6' of https://git-wip-us.apache.org/repos/asf/incubator-ignite
into ignite-883-1
Posted by sb...@apache.org.
Merge branches 'ignite-883-1' and 'ignite-sprint-6' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-883-1
Conflicts:
modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/4a134912
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/4a134912
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/4a134912
Branch: refs/heads/ignite-sprint-6
Commit: 4a1349126eb3b7a78cb849f2e8d4b772ced43fe1
Parents: ca7032e 0907338
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Mon Jun 15 10:45:25 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Mon Jun 15 10:45:25 2015 +0300
----------------------------------------------------------------------
.../datastructures/DataStructuresProcessor.java | 67 +++++++++++++++++++-
.../rest/client/message/GridRouterRequest.java | 18 ++++++
.../rest/client/message/GridRouterResponse.java | 18 ++++++
.../testsuites/IgniteCacheTestSuite4.java | 4 +-
.../ignite/tools/classgen/ClassesGenerator.java | 12 ++++
5 files changed, 114 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
[3/9] incubator-ignite git commit: # ignite-883
Posted by sb...@apache.org.
# ignite-883
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/16f3d32e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/16f3d32e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/16f3d32e
Branch: refs/heads/ignite-sprint-6
Commit: 16f3d32ea2059efedb5fc24bbb24fce80da6a346
Parents: 58fe3ed
Author: sboikov <se...@inria.fr>
Authored: Thu Jun 11 21:53:18 2015 +0300
Committer: sboikov <se...@inria.fr>
Committed: Fri Jun 12 07:24:21 2015 +0300
----------------------------------------------------------------------
.../ignite/spi/discovery/tcp/ClientImpl.java | 13 ++++++--
.../tcp/TcpClientDiscoverySpiSelfTest.java | 33 ++++++++++++++++++++
2 files changed, 43 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/16f3d32e/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 23297ed..23e6f88 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
@@ -398,8 +398,7 @@ class ClientImpl extends TcpDiscoveryImpl {
continue;
}
- assert sockAndRes.get1() != null;
- assert sockAndRes.get2() != null;
+ assert sockAndRes.get1() != null && sockAndRes.get2() != null : sockAndRes;
Socket sock = sockAndRes.get1();
@@ -441,6 +440,10 @@ class ClientImpl extends TcpDiscoveryImpl {
@Nullable private T2<Socket, Integer> sendJoinRequest(boolean recon, InetSocketAddress addr) {
assert addr != null;
+ if (log.isDebugEnabled())
+ log.debug("Send join request [addr=" + addr + ", reconnect=" + recon +
+ ", locNodeId=" + getLocalNodeId() + ']');
+
Collection<Throwable> errs = null;
long ackTimeout0 = spi.ackTimeout;
@@ -1385,8 +1388,12 @@ class ClientImpl extends TcpDiscoveryImpl {
pending = true;
try {
- for (TcpDiscoveryAbstractMessage pendingMsg : msg.pendingMessages())
+ for (TcpDiscoveryAbstractMessage pendingMsg : msg.pendingMessages()) {
+ if (log.isDebugEnabled())
+ log.debug("Process message on reconnect [msg=" + pendingMsg + ']');
+
processDiscoveryMessage(pendingMsg);
+ }
}
finally {
pending = false;
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/16f3d32e/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
index 55a14e4..44fe299 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
@@ -984,6 +984,23 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
+ * @throws Exception If failed.
+ */
+ public void testJoinError() throws Exception {
+ startServerNodes(1);
+
+ Ignite ignite = G.ignite("server-0");
+
+ TestTcpDiscoverySpi srvSpi = ((TestTcpDiscoverySpi)ignite.configuration().getDiscoverySpi());
+
+ srvSpi.failNodeAddedMessage();
+
+ startClientNodes(1);
+
+ checkNodes(1, 1);
+ }
+
+ /**
* @param clientIdx Client index.
* @param srvIdx Server index.
* @throws Exception In case of error.
@@ -1231,6 +1248,9 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
/** */
private final AtomicBoolean openSockLock = new AtomicBoolean();
+ /** */
+ private AtomicInteger failNodeAdded = new AtomicInteger();
+
/**
* @param lock Lock.
*/
@@ -1249,6 +1269,13 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
}
/**
+ *
+ */
+ void failNodeAddedMessage() {
+ failNodeAdded.set(1);
+ }
+
+ /**
* @param isPause Is lock.
* @param locks Locks.
*/
@@ -1266,6 +1293,12 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
GridByteArrayOutputStream bout) throws IOException, IgniteCheckedException {
waitFor(writeLock);
+ if (msg instanceof TcpDiscoveryNodeAddedMessage && failNodeAdded.getAndDecrement() > 0) {
+ log.info("Close socket on message write [msg=" + msg + "]");
+
+ sock.close();
+ }
+
super.writeToSocket(sock, msg, bout);
}
[7/9] incubator-ignite git commit: Merge remote-tracking branch
'remotes/origin/ignite-sprint-6' into ignite-883-1
Posted by sb...@apache.org.
Merge remote-tracking branch 'remotes/origin/ignite-sprint-6' into ignite-883-1
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/ca7032ea
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/ca7032ea
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/ca7032ea
Branch: refs/heads/ignite-sprint-6
Commit: ca7032ea72410d14f6d376abca1a801495207b14
Parents: f115c04 2707194
Author: sboikov <sb...@gridgain.com>
Authored: Fri Jun 12 18:47:29 2015 +0300
Committer: sboikov <sb...@gridgain.com>
Committed: Fri Jun 12 18:47:29 2015 +0300
----------------------------------------------------------------------
RELEASE_NOTES.txt | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
[9/9] incubator-ignite git commit: Merge branches 'ignite-883-1' and
'ignite-sprint-6' of https://git-wip-us.apache.org/repos/asf/incubator-ignite
into ignite-883-1
Posted by sb...@apache.org.
Merge branches 'ignite-883-1' and 'ignite-sprint-6' of https://git-wip-us.apache.org/repos/asf/incubator-ignite into ignite-883-1
Conflicts:
modules/core/src/main/java/org/apache/ignite/cache/query/ScanQuery.java
modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/5d8a5e61
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/5d8a5e61
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/5d8a5e61
Branch: refs/heads/ignite-sprint-6
Commit: 5d8a5e61945436e6bcda31d9fb12edda2feac2a0
Parents: 4a13491
Author: Yakov Zhdanov <yz...@gridgain.com>
Authored: Mon Jun 15 10:45:38 2015 +0300
Committer: Yakov Zhdanov <yz...@gridgain.com>
Committed: Mon Jun 15 10:45:38 2015 +0300
----------------------------------------------------------------------
.../java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/5d8a5e61/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index c598e38..fed5efe 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -136,11 +136,11 @@ public class IgniteCacheTestSuite4 extends TestSuite {
suite.addTestSuite(CacheReadOnlyTransactionalClientSelfTest.class);
- suite.addTestSuite(IgniteCacheManyClientsTest.class);
-
suite.addTestSuite(IgniteStartCacheInTransactionSelfTest.class);
suite.addTestSuite(IgniteStartCacheInTransactionAtomicSelfTest.class);
+ suite.addTestSuite(IgniteCacheManyClientsTest.class);
+
return suite;
}
}