This is an automated email from the ASF dual-hosted git repository.
petrov-mg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new e195b83e83e IGNITE-29005 Refactored Discovery network operations to
rely on session not plain socket (#13509) (#13509)
e195b83e83e is described below
commit e195b83e83e8f071dca3c643574852bcd801c462
Author: Mikhail Petrov <[email protected]>
AuthorDate: Thu Sep 3 16:37:29 2026 +0300
IGNITE-29005 Refactored Discovery network operations to rely on session not
plain socket (#13509) (#13509)
---
.../ignite/spi/discovery/tcp/ClientImpl.java | 95 ++++-----
.../ignite/spi/discovery/tcp/ServerImpl.java | 203 +++++++------------
.../spi/discovery/tcp/TcpDiscoveryIoSession.java | 141 +++++++++++++-
.../ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 214 ++++++++++-----------
.../ignite/internal/IgniteClientRejoinTest.java | 22 +--
.../IgniteDiscoveryMassiveNodeFailTest.java | 16 +-
.../dht/IgniteCacheTopologySplitAbstractTest.java | 20 +-
.../processors/rest/RestProcessorHangTest.java | 8 +-
.../IgniteTcpCommunicationConnectOnInitTest.java | 28 +--
.../spi/discovery/tcp/BlockTcpDiscoverySpi.java | 7 +-
.../discovery/tcp/IgniteMetricsOverflowTest.java | 7 +-
.../discovery/tcp/MultiDataCenterSplitTest.java | 6 +-
.../spi/discovery/tcp/ReceivedMessagesTracker.java | 9 +-
...cpClientDiscoverySpiFailureTimeoutSelfTest.java | 17 +-
.../tcp/TcpClientDiscoverySpiSelfTest.java | 27 +--
.../tcp/TcpClientDiscoveryUnresolvedHostTest.java | 23 +--
.../tcp/TcpDiscoveryCoordinatorFailureTest.java | 18 +-
.../discovery/tcp/TcpDiscoveryFailedJoinTest.java | 33 ++--
.../tcp/TcpDiscoveryNetworkIssuesTest.java | 27 +--
.../TcpDiscoveryPendingMessageDeliveryTest.java | 13 +-
.../tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java | 6 +-
.../tcp/TcpDiscoverySpiReconnectDelayTest.java | 8 +-
22 files changed, 470 insertions(+), 478 deletions(-)
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 3cfacdd43b9..621a234ee54 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
@@ -21,7 +21,6 @@ import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.StreamCorruptedException;
import java.net.InetSocketAddress;
-import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayDeque;
import java.util.ArrayList;
@@ -626,7 +625,7 @@ class ClientImpl extends TcpDiscoveryImpl {
assert joinRes.get1() != null && joinRes.get2() != null : joinRes;
- Socket sock = joinRes.get1().socket();
+ TcpDiscoveryIoSession ses = joinRes.get1();
if (log.isDebugEnabled())
log.debug("Received response to join request [addr=" + addr +
", res=" + joinRes.get2() + ']');
@@ -637,7 +636,7 @@ class ClientImpl extends TcpDiscoveryImpl {
case RES_CONTINUE_JOIN:
case RES_WAIT:
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
return new T2<>(true, null);
@@ -645,7 +644,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (log.isDebugEnabled())
log.debug("Received unexpected response to join
request: " + joinRes.get2());
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
}
}
@@ -696,18 +695,16 @@ class ClientImpl extends TcpDiscoveryImpl {
DiscoveryDataPacket discoveryData = null;
while (true) {
- boolean openSock = false;
+ boolean openSes = false;
- Socket sock = null;
+ TcpDiscoveryIoSession ses = null;
try {
long tsNanos = System.nanoTime();
- sock = spi.openSocket(addr, timeoutHelper);
+ ses = spi.openSession(addr, timeoutHelper);
- openSock = true;
-
- TcpDiscoveryIoSession ses = createSession(sock);
+ openSes = true;
TcpDiscoveryHandshakeRequest req = new
TcpDiscoveryHandshakeRequest(locNodeId, locNode.features());
@@ -724,9 +721,9 @@ class ClientImpl extends TcpDiscoveryImpl {
Collection<InetSocketAddress> redirectAddrs =
res.redirectAddresses();
if (redirectAddrs != null) {
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
- openSock = false;
+ openSes = false;
if (log.isInfoEnabled())
log.info("Reconnecting to the addresses of a proper DC
[addrs=" + redirectAddrs + ']');
@@ -783,10 +780,10 @@ class ClientImpl extends TcpDiscoveryImpl {
log.debug("Message has been sent to address [msg=" + msg +
", addr=" + addr +
", rmtNodeId=" + rmtNodeId + ']');
- return new T2<>(ses, spi.readReceipt(sock,
timeoutHelper.nextTimeoutChunk(ackTimeout0)));
+ return new T2<>(ses, spi.readReceipt(ses,
timeoutHelper.nextTimeoutChunk(ackTimeout0)));
}
catch (IOException | IgniteCheckedException e) {
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
if (log.isDebugEnabled())
log.error("Exception on joining: " + e.getMessage(), e);
@@ -836,7 +833,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (!spi.failureDetectionTimeoutEnabled() && ++reconCnt ==
spi.getReconnectCount())
break;
- if (!openSock) {
+ if (!openSes) {
// Reconnect for the second time, if connection is not
established.
if (connectAttempts < 2) {
connectAttempts++;
@@ -971,7 +968,7 @@ class ClientImpl extends TcpDiscoveryImpl {
TcpDiscoveryIoSession ses = msgWorker.currSes;
if (ses != null)
- U.closeQuiet(ses.socket());
+ U.closeQuiet(ses);
}
/** {@inheritDoc} */
@@ -1133,7 +1130,7 @@ class ClientImpl extends TcpDiscoveryImpl {
this.stopReadLatch = stopReadLatch = new CountDownLatch(1);
- U.closeQuiet(ses.socket());
+ U.closeQuiet(ses);
this.ses = null;
this.rmtNodeId = null;
@@ -1170,26 +1167,25 @@ class ClientImpl extends TcpDiscoveryImpl {
rmtNodeId = this.rmtNodeId;
}
- Socket sock = ses.socket();
-
U.enhanceThreadName(U.id8(rmtNodeId)
- + ' ' + sock.getInetAddress().getHostAddress()
- + ":" + sock.getPort());
+ + ' ' + ses.socket().getInetAddress().getHostAddress()
+ + ":" + ses.socket().getPort());
try {
- assert sock.getKeepAlive() && sock.getTcpNoDelay() :
"Socket wasn't configured properly:" +
- " KeepAlive " + sock.getKeepAlive() +
- " TcpNoDelay " + sock.getTcpNoDelay();
+ assert ses.socket().getKeepAlive() &&
ses.socket().getTcpNoDelay() :
+ "Socket wasn't configured properly:" +
+ " KeepAlive " + ses.socket().getKeepAlive() +
+ " TcpNoDelay " + ses.socket().getTcpNoDelay();
while (!isInterrupted()) {
TcpDiscoveryAbstractMessage msg;
try {
- msg = spi.readMessage(ses, sock.getSoTimeout());
+ msg = spi.readMessage(ses,
ses.socket().getSoTimeout());
}
catch (IgniteCheckedException e) {
if (log.isDebugEnabled())
- U.error(log, "Failed to read message [sock=" +
sock + ", " +
+ U.error(log, "Failed to read message [ses=" +
ses + ", " +
"locNodeId=" + getLocalNodeId() + ",
rmtNodeId=" + rmtNodeId + ']', e);
// Exists possibility that exception raised on
interruption.
@@ -1209,7 +1205,7 @@ class ClientImpl extends TcpDiscoveryImpl {
"(make sure same versions of all classes
are available on all nodes) " +
"[rmtNodeId=" + rmtNodeId + ", err=" +
clsNotFoundEx.getMessage() + ']');
else
- LT.error(log, e, "Failed to read message
[sock=" + sock + ", locNodeId=" +
+ LT.error(log, e, "Failed to read message
[ses=" + ses + ", locNodeId=" +
getLocalNodeId() + ", rmtNodeId=" +
rmtNodeId + ']');
continue;
@@ -1236,10 +1232,10 @@ class ClientImpl extends TcpDiscoveryImpl {
msgWorker.addMessage(new SessionClosedMessage(ses));
if (log.isDebugEnabled())
- U.error(log, "Connection failed [sock=" + sock + ",
locNodeId=" + getLocalNodeId() + ']', e);
+ U.error(log, "Connection failed [ses=" + ses + ",
locNodeId=" + getLocalNodeId() + ']', e);
}
finally {
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
synchronized (mux) {
if (this.ses == ses) {
@@ -1386,7 +1382,7 @@ class ClientImpl extends TcpDiscoveryImpl {
}
}
- U.closeQuiet(ses.socket());
+ U.closeQuiet(ses);
this.ses = null;
@@ -1479,7 +1475,7 @@ class ClientImpl extends TcpDiscoveryImpl {
else
U.error(log, "Failed to send message: " + msg, e);
- U.closeQuiet(ses.socket());
+ U.closeQuiet(ses);
synchronized (mux) {
if (ses == this.ses)
@@ -1542,8 +1538,7 @@ class ClientImpl extends TcpDiscoveryImpl {
TcpDiscoveryIoSession ses = this.ses;
- if (ses != null)
- U.closeQuiet(ses.socket());
+ U.closeQuiet(ses);
}
/** {@inheritDoc} */
@@ -1578,26 +1573,19 @@ class ClientImpl extends TcpDiscoveryImpl {
this.ses = ses;
- Socket sock = ses.socket();
-
if (isInterrupted())
throw new InterruptedException();
- int oldTimeout = 0;
-
try {
- oldTimeout = sock.getSoTimeout();
-
- sock.setSoTimeout((int)spi.netTimeout);
-
- assert sock.getKeepAlive() && sock.getTcpNoDelay() :
"Socket wasn't configured properly:" +
- " KeepAlive " + sock.getKeepAlive() +
- " TcpNoDelay " + sock.getTcpNoDelay();
+ assert ses.socket().getKeepAlive() &&
ses.socket().getTcpNoDelay() :
+ "Socket wasn't configured properly:" +
+ " KeepAlive " + ses.socket().getKeepAlive() +
+ " TcpNoDelay " + ses.socket().getTcpNoDelay();
List<TcpDiscoveryAbstractMessage> msgs = null;
while (!isInterrupted()) {
- TcpDiscoveryAbstractMessage msg =
spi.readMessage(ses, sock.getSoTimeout());
+ TcpDiscoveryAbstractMessage msg =
spi.readMessage(ses, spi.netTimeout);
if (msg instanceof
TcpDiscoveryClientReconnectMessage) {
TcpDiscoveryClientReconnectMessage res =
(TcpDiscoveryClientReconnectMessage)msg;
@@ -1632,7 +1620,7 @@ class ClientImpl extends TcpDiscoveryImpl {
}
}
catch (IOException | IgniteCheckedException e) {
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
if (log.isDebugEnabled())
log.error("Reconnect error [join=" + join + ",
timeout=" + timeout + ']', e);
@@ -1650,10 +1638,6 @@ class ClientImpl extends TcpDiscoveryImpl {
else
U.warn(log, "Failed to reconnect to cluster (will
retry): " + e);
}
- finally {
- if (success)
- sock.setSoTimeout(oldTimeout);
- }
}
}
catch (IOException | IgniteCheckedException e) {
@@ -1667,8 +1651,7 @@ class ClientImpl extends TcpDiscoveryImpl {
if (!success) {
TcpDiscoveryIoSession ses = this.ses;
- if (ses != null)
- U.closeQuiet(ses.socket());
+ U.closeQuiet(ses);
if (join)
joinError(new IgniteSpiException("Failed to connect to
cluster, connection failed and failed " +
@@ -1814,9 +1797,9 @@ class ClientImpl extends TcpDiscoveryImpl {
}
else if (msg instanceof SessionClosedMessage sesClosedMsg)
{
if (sesClosedMsg.ses == currSes) {
- Socket sock = currSes.socket();
-
- InetSocketAddress prevAddr = new
InetSocketAddress(sock.getInetAddress(), sock.getPort());
+ InetSocketAddress prevAddr = new InetSocketAddress(
+ currSes.socket().getInetAddress(),
+ currSes.socket().getPort());
currSes = null;
@@ -1981,7 +1964,7 @@ class ClientImpl extends TcpDiscoveryImpl {
TcpDiscoveryIoSession ses = this.currSes;
if (ses != null)
- U.closeQuiet(ses.socket());
+ U.closeQuiet(ses);
if (joinLatch.getCount() > 0)
joinError(new IgniteSpiException("Some error in join
process.")); // This should not occur.
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 fe86a1c789b..e8a9e5c2394 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
@@ -950,9 +950,7 @@ class ServerImpl extends TcpDiscoveryImpl {
fut.sock = sock;
- sock = spi.openSocket(sock, addr, timeoutHelper);
-
- TcpDiscoveryIoSession ses = createSession(sock);
+ TcpDiscoveryIoSession ses = spi.openSession(sock,
addr, timeoutHelper);
spi.writeMessage(ses, new
TcpDiscoveryPingRequest(locNodeId, clientNodeId),
timeoutHelper.nextTimeoutChunk(spi.getSocketTimeout()));
@@ -1495,17 +1493,16 @@ class ServerImpl extends TcpDiscoveryImpl {
// since remote node may leave in the middle of the first
iteration.
joinReqSent = false;
- boolean openSock = false;
+ boolean openSes = false;
- Socket sock = null;
+ TcpDiscoveryIoSession ses = null;
try {
long tsNanos = System.nanoTime();
- sock = spi.openSocket(addr, timeoutHelper);
- TcpDiscoveryIoSession ses = createSession(sock);
+ ses = spi.openSession(addr, timeoutHelper);
- openSock = true;
+ openSes = true;
TcpDiscoveryHandshakeRequest req = new
TcpDiscoveryHandshakeRequest(locNodeId, locNode.features());
@@ -1564,7 +1561,7 @@ class ServerImpl extends TcpDiscoveryImpl {
// E.g. due to class not found issue.
joinReqSent = msg instanceof TcpDiscoveryJoinRequestMessage;
- int receipt = spi.readReceipt(sock,
timeoutHelper.nextTimeoutChunk(ackTimeout0));
+ int receipt = spi.readReceipt(ses,
timeoutHelper.nextTimeoutChunk(ackTimeout0));
spi.stats.onMessageSent(msg, U.nanosToMillis(tsNanos0 -
tsNanos));
@@ -1632,7 +1629,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (!spi.failureDetectionTimeoutEnabled() && ++reconCnt ==
spi.getReconnectCount())
break;
- if (!openSock) {
+ if (!openSes) {
// Reconnect for the second time, if connection is not
established.
if (connectAttempts < 2) {
connectAttempts++;
@@ -1652,7 +1649,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
}
finally {
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
}
}
@@ -1993,10 +1990,7 @@ class ServerImpl extends TcpDiscoveryImpl {
/** {@inheritDoc} */
@Override public void brakeConnection() {
- Socket sock = msgWorker.sock;
-
- if (sock != null)
- U.closeQuiet(sock);
+ U.closeQuiet(msgWorker.ses);
}
/** {@inheritDoc} */
@@ -2864,9 +2858,6 @@ class ServerImpl extends TcpDiscoveryImpl {
/** Force pending messages send. */
private boolean forceSndPending;
- /** Socket. */
- private Socket sock;
-
// This serializer is used exclusively for serializing messages sent
to clients,
// as it represents a special case within the RingMessageWorker
workflow.
// Generally, both serialization and deserialization of messages
should be handled by TcpDiscoveryIoSession.
@@ -3188,12 +3179,12 @@ class ServerImpl extends TcpDiscoveryImpl {
onMessageExchanged();
}
- if (next != null && sock != null) {
+ if (next != null && ses != null) {
// Messages that change topology.
if (msg instanceof TcpDiscoveryNodeLeftMessage || msg
instanceof TcpDiscoveryNodeFailedMessage ||
msg instanceof TcpDiscoveryNodeAddFinishedMessage || msg
instanceof TcpDiscoveryNodeAddedMessage) {
- U.enhanceThreadName(U.id8(next.id()) + ' ' +
sock.getInetAddress().getHostAddress()
- + ":" + sock.getPort() + (isLocalNodeCoordinator() ? "
crd" : ""));
+ U.enhanceThreadName(U.id8(next.id()) + ' ' +
ses.socket().getInetAddress().getHostAddress()
+ + ":" + ses.socket().getPort() +
(isLocalNodeCoordinator() ? " crd" : ""));
}
}
@@ -3362,9 +3353,9 @@ class ServerImpl extends TcpDiscoveryImpl {
debugLog(msg, "New next node [newNext=" + newNext + ",
formerNext=" + next +
", ring=" + ring + ", failedNodes=" + failedNodes
+ ']');
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
- sock = null;
+ ses = null;
newNextNode(newNext);
@@ -3398,7 +3389,7 @@ class ServerImpl extends TcpDiscoveryImpl {
IgniteSpiOperationTimeoutHelper timeoutHelper = null;
while (true) {
- if (sock == null) {
+ if (ses == null) {
// We re-create the helper here because it could
be created earlier with wrong timeout on
// message sending like
IgniteConfiguration.failureDetectionTimeout. Here we are in the
// state of conenction recovering and have to work
with
@@ -3408,15 +3399,13 @@ class ServerImpl extends TcpDiscoveryImpl {
boolean success = false;
- boolean openSock = false;
+ boolean openSes = false;
// Restore ring.
try {
- sock = spi.openSocket(addr, timeoutHelper);
-
- ses = createSession(sock);
+ ses = spi.openSession(addr, timeoutHelper);
- openSock = true;
+ openSes = true;
// Handshake.
TcpDiscoveryHandshakeRequest hndMsg = new
TcpDiscoveryHandshakeRequest(locNodeId, locNode.features());
@@ -3464,9 +3453,9 @@ class ServerImpl extends TcpDiscoveryImpl {
newNextNode(ring.nextNode(failedNodes));
}
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
- sock = null;
+ ses = null;
if
(checkConnectionRecoveryFailed(sndState, failedNodes))
return; // Nothing to do here.
@@ -3482,9 +3471,9 @@ class ServerImpl extends TcpDiscoveryImpl {
if (log.isDebugEnabled())
log.debug("Handshake response from
local node: " + res);
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
- sock = null;
+ ses = null;
break;
}
@@ -3565,7 +3554,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (sndState != null &&
checkConnectionRecoveryFailed(sndState, failedNodes))
return; // Nothing to do here.
- if (!openSock)
+ if (!openSes)
break; // Don't retry if we can not
establish connection.
if (!spi.failureDetectionTimeoutEnabled() &&
++reconCnt == spi.getReconnectCount())
@@ -3588,9 +3577,9 @@ class ServerImpl extends TcpDiscoveryImpl {
if (log.isDebugEnabled())
log.debug("Closing socket to next: " +
next);
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
- sock = null;
+ ses = null;
}
else {
// Resetting timeout control object to let
the code below to use a new one
@@ -3643,7 +3632,7 @@ class ServerImpl extends TcpDiscoveryImpl {
long tsNanos0 = System.nanoTime();
- int res = spi.readReceipt(sock,
timeoutHelper.nextTimeoutChunk(ackTimeout0));
+ int res = spi.readReceipt(ses,
timeoutHelper.nextTimeoutChunk(ackTimeout0));
updateLastSentMessageTime();
@@ -3687,7 +3676,7 @@ class ServerImpl extends TcpDiscoveryImpl {
long tsNanos0 = System.nanoTime();
- int res = spi.readReceipt(sock,
timeoutHelper.nextTimeoutChunk(ackTimeout0));
+ int res = spi.readReceipt(ses,
timeoutHelper.nextTimeoutChunk(ackTimeout0));
updateLastSentMessageTime();
@@ -3757,9 +3746,9 @@ class ServerImpl extends TcpDiscoveryImpl {
if (log.isDebugEnabled())
log.debug("Closing socket to next (not
sent): " + next);
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
- sock = null;
+ ses = null;
if (log.isDebugEnabled()) {
log.debug("Message has not been sent
[next=" + next.id() + ", msg=" + msg +
@@ -5337,7 +5326,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (wrk != null)
wrk.addMessage(msg);
}
- else if (leftNode.equals(next) && sock != null) {
+ else if (leftNode.equals(next) && ses != null) {
try {
spi.writeMessage(ses, msg,
spi.failureDetectionTimeoutEnabled() ?
spi.failureDetectionTimeout() :
spi.getSocketTimeout());
@@ -5358,7 +5347,7 @@ class ServerImpl extends TcpDiscoveryImpl {
newNextNode(null);
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
}
}
@@ -5395,7 +5384,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (log.isDebugEnabled())
log.debug("Unable to send message across the ring
(topology has no remote nodes): " + msg);
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
processPendingMessagesLocally(msg);
}
@@ -5575,7 +5564,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (log.isDebugEnabled())
log.debug("Unable to send message across the ring
(topology has no remote nodes): " + msg);
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
processPendingMessagesLocally(msg);
}
@@ -6498,53 +6487,11 @@ class ServerImpl extends TcpDiscoveryImpl {
// Set socket options.
spi.configureSocketOptions(sock);
- int timeout = sock.getSoTimeout();
-
- sock.setSoTimeout((int)spi.netTimeout);
-
for (IgniteInClosure<Socket> connLsnr :
spi.incomeConnLsnrs)
connLsnr.apply(sock);
- byte[] buf = new byte[4];
- int read = 0;
-
- while (read < buf.length) {
- int r = sock.getInputStream().read(buf, read,
buf.length - read);
-
- if (r >= 0)
- read += r;
- else {
- if (log.isDebugEnabled())
- log.debug("Failed to read magic header (too
few bytes received) " +
- "[rmtAddr=" + rmtAddr +
- ", locAddr=" +
sock.getLocalSocketAddress() + ']');
-
- LT.warn(log, "Failed to read magic header (too few
bytes received) [rmtAddr=" +
- rmtAddr + ", locAddr=" +
sock.getLocalSocketAddress() + ']');
-
- return;
- }
- }
-
- if (!Arrays.equals(buf, U.IGNITE_HEADER)) {
- if (log.isDebugEnabled())
- log.debug("Unknown connection detected (possible
reasons: an incompatible Ignite node or " +
- "other software connecting to this Ignite
port" +
- (!spi.isSslEnabled() ? ", or missing SSL
configuration on remote node" : "") +
- ") [rmtAddr=" + rmtAddr +
- ", locAddr=" + sock.getLocalSocketAddress() +
- ", rcvdHdr=" + U.byteArray2HexString(buf) +
']');
-
- LT.warn(log, "Unknown connection detected (possible
reasons: an incompatible Ignite node or " +
- "other software connecting to this Ignite port" +
- (!spi.isSslEnabled() ? ", or missing SSL
configuration on remote node" : "") +
- ") [rmtAddr=" + sock.getInetAddress() + ",
rcvdHdr=" + U.byteArray2HexString(buf) + ']', true);
-
+ if (!spi.readMagicHeader(ses, spi.netTimeout))
return;
- }
-
- // Restore timeout.
- sock.setSoTimeout(timeout);
TcpDiscoveryAbstractMessage msg = spi.readMessage(ses,
spi.netTimeout);
@@ -6731,7 +6678,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
if (req.client()) {
- ClientMessageWorker clientMsgWrk0 = new
ClientMessageWorker(sock, nodeId, log);
+ ClientMessageWorker clientMsgWrk0 = new
ClientMessageWorker(ses, nodeId, log);
while (true) {
ClientMessageWorker old =
clientMsgWorkers.putIfAbsent(nodeId, clientMsgWrk0);
@@ -6757,7 +6704,7 @@ class ServerImpl extends TcpDiscoveryImpl {
log.debug("Already have client message worker,
closing connection " +
"[locNodeId=" + locNodeId +
", rmtNodeId=" + nodeId +
- ", workerSock=" + old.sock +
+ ", workerSes=" + old.ses +
", sock=" + sock + ']');
return;
@@ -6861,7 +6808,7 @@ class ServerImpl extends TcpDiscoveryImpl {
if (msg instanceof TcpDiscoveryConnectionCheckMessage)
{
ringMessageReceived();
- spi.writeToSocket(sock, RES_OK, sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
continue;
}
@@ -6887,7 +6834,7 @@ class ServerImpl extends TcpDiscoveryImpl {
TcpDiscoverySpiState state = spiStateCopy();
if (state == CONNECTED) {
- spi.writeToSocket(sock, RES_OK, sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
if (clientMsgWrk != null &&
clientMsgWrk.runner() == null && !clientMsgWrk.isDone())
new
MessageWorkerThreadWithCleanup<>(clientMsgWrk, log).start();
@@ -6901,21 +6848,21 @@ class ServerImpl extends TcpDiscoveryImpl {
// If message is received from previous node
and node is connecting forward to next node.
if
(!getLocalNodeId().equals(msg0.routerNodeId()) && state == CONNECTING) {
- spi.writeToSocket(sock, RES_OK,
sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
msgWorker.addMessage(msg);
continue;
}
- spi.writeToSocket(sock, RES_CONTINUE_JOIN,
sockTimeout);
+ spi.writeReceipt(ses, RES_CONTINUE_JOIN,
sockTimeout);
break;
}
}
else if (msg instanceof
TcpDiscoveryDuplicateIdMessage) {
// Send receipt back.
- spi.writeToSocket(sock, RES_OK, sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
boolean ignored = false;
@@ -6944,7 +6891,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else if (msg instanceof TcpDiscoveryAuthFailedMessage)
{
// Send receipt back.
- spi.writeToSocket(sock, RES_OK, sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
synchronized (mux) {
if (spiState == CONNECTING) {
@@ -6972,7 +6919,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else if (msg instanceof
TcpDiscoveryCheckFailedMessage) {
// Send receipt back.
- spi.writeToSocket(sock, RES_OK, sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
boolean ignored = false;
@@ -7015,7 +6962,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else if (msg instanceof
TcpDiscoveryLoopbackProblemMessage) {
// Send receipt back.
- spi.writeToSocket(sock, RES_OK, sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
boolean ignored = false;
@@ -7080,7 +7027,7 @@ class ServerImpl extends TcpDiscoveryImpl {
clientMsgWrk.addMessage(ack);
}
else
- spi.writeToSocket(sock, RES_OK, sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
if (metricsUpdateMsg != null)
processClientMetricsUpdateMessage(metricsUpdateMsg);
@@ -7155,7 +7102,7 @@ class ServerImpl extends TcpDiscoveryImpl {
U.interrupt(clientMsgWrk.runner());
}
- U.close(sock, log);
+ ses.close(log);
if (log.isInfoEnabled()) {
log.info("Finished serving remote node connection
[rmtAddr=" + rmtAddr +
@@ -7398,13 +7345,13 @@ class ServerImpl extends TcpDiscoveryImpl {
// Check that joining node can accept incoming connections.
if (node.clientRouterNodeId() == null) {
if (!pingJoiningNode(node)) {
- spi.writeToSocket(sock, RES_JOIN_IMPOSSIBLE,
sockTimeout);
+ spi.writeReceipt(ses, RES_JOIN_IMPOSSIBLE,
sockTimeout);
return false;
}
}
- spi.writeToSocket(sock, RES_OK, sockTimeout);
+ spi.writeReceipt(ses, RES_OK, sockTimeout);
if (log.isDebugEnabled())
log.debug("Responded to join request message [msg=" + msg
+ ", res=" + RES_OK + ']');
@@ -7441,7 +7388,7 @@ class ServerImpl extends TcpDiscoveryImpl {
// Local node is stopping. Remote node should try next one.
res = RES_CONTINUE_JOIN;
- spi.writeToSocket(sock, res, sockTimeout);
+ spi.writeReceipt(ses, res, sockTimeout);
if (log.isDebugEnabled())
log.debug("Responded to join request message [msg=" + msg
+ ", res=" + res + ']');
@@ -7486,14 +7433,14 @@ class ServerImpl extends TcpDiscoveryImpl {
@Override public void interrupt() {
super.interrupt();
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
}
/** {@inheritDoc} */
@Override protected void cleanup() {
super.cleanup();
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
synchronized (mux) {
readers.remove(this);
@@ -7553,8 +7500,8 @@ class ServerImpl extends TcpDiscoveryImpl {
/** */
private final TcpDiscoveryMessageSerializer clientMsgSer;
- /** Socket. */
- private final Socket sock;
+ /** Session shared with the socket reader serving the same client
connection. */
+ private final TcpDiscoveryIoSession ses;
/** Current client metrics. */
private volatile ClusterMetrics metrics;
@@ -7569,22 +7516,22 @@ class ServerImpl extends TcpDiscoveryImpl {
private IgniteProductVersion clientVer;
/**
- * @param sock Socket.
+ * @param ses IO session of the connection to the client.
* @param clientNodeId Node ID.
* @param log Logger.
*/
- private ClientMessageWorker(Socket sock, UUID clientNodeId,
IgniteLogger log) {
+ private ClientMessageWorker(TcpDiscoveryIoSession ses, UUID
clientNodeId, IgniteLogger log) {
super(
"tcp-disco-client-message-worker-[" + U.id8(clientNodeId) +
- ' ' + sock.getInetAddress().getHostAddress() +
- ":" + sock.getPort() + ']',
+ ' ' + ses.socket().getInetAddress().getHostAddress() +
+ ":" + ses.socket().getPort() + ']',
log,
Math.max(spi.metricsUpdateFreq, 10),
null,
new LinkedBlockingDeque<>()
);
- this.sock = sock;
+ this.ses = ses;
this.clientNodeId = clientNodeId;
clientMsgSer = new TcpDiscoveryMessageSerializer(spi);
@@ -7659,14 +7606,14 @@ class ServerImpl extends TcpDiscoveryImpl {
clientVer = IgniteUtils.productVersion(node);
else if (msgLog.isDebugEnabled()) {
msgLog.debug("Skip sending message ack to client,
fail to get client node " +
- "[sock=" + sock + ", locNodeId=" +
getLocalNodeId() +
+ "[ses=" + ses + ", locNodeId=" +
getLocalNodeId() +
", rmtNodeId=" + clientNodeId + ", msg=" + msg
+ ']');
}
}
if (clientVer != null) {
if (msgLog.isDebugEnabled()) {
- msgLog.debug("Sending message ack to client
[sock=" + sock + ", locNodeId="
+ msgLog.debug("Sending message ack to client [ses="
+ ses + ", locNodeId="
+ getLocalNodeId() + ", rmtNodeId=" +
clientNodeId + ", msg=" + msg + ']');
}
@@ -7676,7 +7623,7 @@ class ServerImpl extends TcpDiscoveryImpl {
}
else {
if (msgLog.isDebugEnabled()) {
- msgLog.debug("Redirecting message to client [sock=" +
sock + ", locNodeId="
+ msgLog.debug("Redirecting message to client [ses=" +
ses + ", locNodeId="
+ getLocalNodeId() + ", rmtNodeId=" + clientNodeId
+ ", msg=" + msg + ']');
}
@@ -7694,10 +7641,10 @@ class ServerImpl extends TcpDiscoveryImpl {
}
catch (IgniteCheckedException | IOException e) {
if (log.isDebugEnabled())
- U.error(log, "Client connection failed [sock=" + sock + ",
locNodeId="
+ U.error(log, "Client connection failed [ses=" + ses + ",
locNodeId="
+ getLocalNodeId() + ", rmtNodeId=" + clientNodeId +
", msg=" + msg + ']', e);
- onException("Client connection failed [sock=" + sock + ",
locNodeId="
+ onException("Client connection failed [ses=" + ses + ",
locNodeId="
+ getLocalNodeId() + ", rmtNodeId=" + clientNodeId + ",
msg=" + msg + ']', e);
}
finally {
@@ -7706,7 +7653,7 @@ class ServerImpl extends TcpDiscoveryImpl {
U.interrupt(runner());
- U.close(sock, log);
+ ses.close(log);
}
}
}
@@ -7719,7 +7666,7 @@ class ServerImpl extends TcpDiscoveryImpl {
throws IgniteCheckedException, IOException {
byte[] msgBytes = msgT.get2() == null ?
clientMsgSer.serializeMessage(msgT.get1()) : msgT.get2();
- spi.writeToSocket(sock, msgBytes, timeout);
+ spi.write(ses, msgBytes, timeout);
}
/**
@@ -7799,7 +7746,7 @@ class ServerImpl extends TcpDiscoveryImpl {
@Override protected void tearDown() {
pingResult(false);
- U.closeQuiet(sock);
+ U.closeQuiet(ses);
}
/** {@inheritDoc} */
@@ -8007,24 +7954,6 @@ class ServerImpl extends TcpDiscoveryImpl {
GridPingFutureAdapter(@Nullable UUID nodeId) {
this.nodeId = nodeId;
}
-
- /**
- * Returns socket associated with this ping future.
- *
- * @return Socket or {@code null} if no socket associated.
- */
- public Socket sock() {
- return sock;
- }
-
- /**
- * Associates socket with this ping future.
- *
- * @param sock Socket.
- */
- public void sock(Socket sock) {
- this.sock = sock;
- }
}
/**
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
index 8c34284e87f..71a4995eeab 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryIoSession.java
@@ -26,12 +26,14 @@ import java.io.InputStream;
import java.io.OutputStream;
import java.io.StreamCorruptedException;
import java.net.Socket;
+import java.net.SocketException;
import java.nio.ByteBuffer;
import java.security.cert.Certificate;
import javax.net.ssl.SSLPeerUnverifiedException;
import javax.net.ssl.SSLSocket;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteException;
+import org.apache.ignite.IgniteLogger;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.direct.DirectMessageReader;
@@ -60,7 +62,7 @@ import org.jetbrains.annotations.Nullable;
* </ul>
* A leading byte is used to distinguish between the modes. The byte will be
removed in future.
*/
-public class TcpDiscoveryIoSession {
+public class TcpDiscoveryIoSession implements AutoCloseable {
/** Default size of buffer used for buffering socket in/out. */
private static final int DFLT_SOCK_BUFFER_SIZE = 8192;
@@ -68,7 +70,7 @@ public class TcpDiscoveryIoSession {
private static final int MSG_BUFFER_SIZE = 100;
/** */
- final TcpDiscoverySpi spi;
+ private final TcpDiscoverySpi spi;
/** */
private final Socket sock;
@@ -145,6 +147,20 @@ public class TcpDiscoveryIoSession {
}
}
+ /**
+ * Reads the next discovery message from the socket input stream limiting
read time.
+ *
+ * @param timeout Socket read timeout for this operation, {@code 0} means
infinite.
+ * @param <T> Type of the expected message.
+ * @return Deserialized message instance.
+ * @throws IgniteCheckedException If deserialization fails.
+ */
+ <T extends Message> T readMessage(long timeout) throws
IgniteCheckedException, IOException {
+ try (SocketTimeoutScope ignored = withTimeout(timeout)) {
+ return readMessage();
+ }
+ }
+
/**
* Reads the next discovery message from the socket input stream.
*
@@ -165,7 +181,7 @@ public class TcpDiscoveryIoSession {
msg = spi.messageFactory().create(msgType);
}
catch (IgniteException e) {
- detectSslAlert(b0, b1, in);
+ detectSslAlert(b0, b1);
// 'Invalid message type' should not be lost.
throw e;
@@ -269,11 +285,94 @@ public class TcpDiscoveryIoSession {
}
/**
- * Checks wheter input stream contains SSL alert.
+ * Writes raw data to the underlying socket output stream.
+ *
+ * @param data Raw data to write.
+ * @throws IOException If failed.
+ */
+ void write(byte[] data) throws IOException {
+ out.write(data);
+
+ out.flush();
+ }
+
+ /**
+ * Writes a single byte response to the underlying socket output stream.
+ *
+ * @param b Integer response.
+ * @throws IOException If failed.
+ */
+ void write(int b) throws IOException {
+ out.write(b);
+
+ out.flush();
+ }
+
+ /**
+ * Reads a single byte from the underlying socket input stream limiting
read time.
+ *
+ * @param timeout Socket read timeout for this operation, {@code 0} means
infinite.
+ * @return Receipt.
+ * @throws IOException If failed.
+ * @throws EOFException If the connection has been closed.
+ */
+ int read(long timeout) throws IOException {
+ try (SocketTimeoutScope ignored = withTimeout(timeout)) {
+ int res = in.read();
+
+ if (res == -1)
+ throw new EOFException();
+
+ return res;
+ }
+ }
+
+ /**
+ * Reads {@code data.length} bytes from the underlying socket stream into
the given array limiting
+ * read time.
+ *
+ * @param data Array to read the data into.
+ * @param timeout Socket read timeout for this operation, {@code 0} means
infinite.
+ * @return Number of bytes read, less than {@code data.length} only if the
connection has been closed.
+ * @throws IOException If failed.
+ */
+ int read(byte[] data, long timeout) throws IOException {
+ try (SocketTimeoutScope ignored = withTimeout(timeout)) {
+ return in.readNBytes(data, 0, data.length);
+ }
+ }
+
+ /**
+ * Applies the given read timeout to the session socket until the returned
scope is closed.
+ *
+ * @param timeout Socket read timeout, {@code 0} means infinite.
+ * @return Scope restoring the previous socket read timeout when closed.
+ * @throws SocketException If the timeout can not be applied.
+ */
+ private SocketTimeoutScope withTimeout(long timeout) throws
SocketException {
+ SocketTimeoutScope scope = new SocketTimeoutScope(sock.getSoTimeout());
+
+ sock.setSoTimeout((int)timeout);
+
+ return scope;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() {
+ U.closeQuiet(sock);
+ }
+
+ /** */
+ void close(IgniteLogger log) {
+ U.close(sock, log);
+ }
+
+ /**
+ * Checks whether input stream contains SSL alert.
* See handling {@code StreamCorruptedException} in {@link #readMessage()}.
* Keeps logic similar to {@link
java.io.ObjectInputStream#readStreamHeader}.
*/
- private void detectSslAlert(byte b0, byte b1, InputStream in) throws
IOException {
+ private void detectSslAlert(byte b0, byte b1) throws IOException {
byte[] hdr = new byte[4];
hdr[0] = b0;
hdr[1] = b1;
@@ -293,6 +392,27 @@ public class TcpDiscoveryIoSession {
return "TcpDiscoveryIoSession [sock=" + sock + ']';
}
+ /** Restores the socket read timeout changed for the duration of a single
operation. */
+ private final class SocketTimeoutScope implements AutoCloseable {
+ /** */
+ private final int oldTimeout;
+
+ /** */
+ private SocketTimeoutScope(int oldTimeout) {
+ this.oldTimeout = oldTimeout;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void close() {
+ try {
+ sock.setSoTimeout(oldTimeout);
+ }
+ catch (SocketException ignored) {
+ // No-op.
+ }
+ }
+ }
+
/**
* Input stream implementation that combines a byte array and a regular
InputStream allowing to read bytes
* from the array first and then proceed with reading from InputStream.
@@ -336,7 +456,12 @@ public class TcpDiscoveryIoSession {
if (len0 == len)
return len0;
- return len0 + super.read(b, off + len0, len - len0);
+ int read = super.read(b, off + len0, len - len0);
+
+ if (read < 0)
+ return len0 > 0 ? len0 : read;
+
+ return len0 + read;
}
/** {@inheritDoc} */
@@ -348,7 +473,9 @@ public class TcpDiscoveryIoSession {
@Override public int readNBytes(byte[] b, int off, int len) throws
IOException {
int len0 = readPrefixBuffer(b, off, len);
- return super.readNBytes(b, off + len0, len - len0);
+ assert len0 <= len;
+
+ return len0 + super.readNBytes(b, off + len0, len - len0);
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index a20ac2d7f01..c4c74289af3 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -17,9 +17,7 @@
package org.apache.ignite.spi.discovery.tcp;
-import java.io.EOFException;
import java.io.IOException;
-import java.io.OutputStream;
import java.io.Serializable;
import java.io.StreamCorruptedException;
import java.net.InetAddress;
@@ -1575,58 +1573,62 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
}
/**
- * @param sockAddr Remote address.
+ * @param rmtAddr Remote address.
* @param timeoutHelper Timeout helper.
- * @return Opened socket.
+ * @return Session bound to the connected socket.
* @throws IOException If failed.
* @throws IgniteSpiOperationTimeoutException In case of timeout.
* @throws IgniteCheckedException If node is not yet initialized or is
stopping.
*/
- protected Socket openSocket(
- InetSocketAddress sockAddr,
+ protected TcpDiscoveryIoSession openSession(
+ InetSocketAddress rmtAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteSpiOperationTimeoutException,
IgniteCheckedException {
- return openSocket(createSocket(), sockAddr, timeoutHelper);
+ Socket sock = createSocket();
+
+ try {
+ return openSession(sock, rmtAddr, timeoutHelper);
+ }
+ catch (IOException | IgniteCheckedException | IgniteException e) {
+ U.closeQuiet(sock);
+
+ throw e;
+ }
}
/**
* Connects to remote address sending {@code U.IGNITE_HEADER} when
connection is established.
*
- * @param sock Socket bound to a local host address.
+ * @param sock Socket bound to a local host address, not connected yet.
* @param remAddr Remote address.
* @param timeoutHelper Timeout helper.
- * @return Connected socket.
+ * @return Session bound to the connected socket.
* @throws IOException If failed.
* @throws IgniteSpiOperationTimeoutException In case of timeout.
* @throws IgniteCheckedException If node is not yet initialized or is
stopping.
*/
- protected Socket openSocket(
+ protected TcpDiscoveryIoSession openSession(
Socket sock,
InetSocketAddress remAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteSpiOperationTimeoutException,
IgniteCheckedException {
+ assert sock != null;
assert remAddr != null;
- try {
- InetSocketAddress resolved = remAddr.isUnresolved() ?
- new
InetSocketAddress(InetAddress.getByName(remAddr.getHostName()),
remAddr.getPort()) : remAddr;
+ InetSocketAddress resolved = remAddr.isUnresolved() ?
+ new
InetSocketAddress(InetAddress.getByName(remAddr.getHostName()),
remAddr.getPort()) : remAddr;
- InetAddress addr = resolved.getAddress();
+ InetAddress addr = resolved.getAddress();
- assert addr != null;
+ assert addr != null;
- sock.connect(resolved,
(int)timeoutHelper.nextTimeoutChunk(sockTimeout));
+ sock.connect(resolved,
(int)timeoutHelper.nextTimeoutChunk(sockTimeout));
- writeToSocket(sock, U.IGNITE_HEADER,
timeoutHelper.nextTimeoutChunk(sockTimeout));
+ TcpDiscoveryIoSession ses = new TcpDiscoveryIoSession(sock, this);
- return sock;
- }
- catch (IOException | IgniteCheckedException e) {
- if (sock != null)
- U.closeQuiet(sock);
+ write(ses, U.IGNITE_HEADER,
timeoutHelper.nextTimeoutChunk(sockTimeout));
- throw e;
- }
+ return ses;
}
/**
@@ -1634,11 +1636,11 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
* before, on SSL handshake, and doesn't accept new messages. In a such
case it's possible to check the original error
* by reading the socket input stream.
*
- * @param sock Socket to check.
+ * @param ses Session to check.
* @param writeErr Error on writing a message to the socket.
* @return {@code SSLException} in case of SSL error, or {@code null}
otherwise.
*/
- private @Nullable SSLException checkSslException(Socket sock, Exception
writeErr) {
+ private @Nullable SSLException checkSslException(TcpDiscoveryIoSession
ses, Exception writeErr) {
if (!sslEnable)
return null;
@@ -1650,7 +1652,7 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
try {
// Set timeout to 1ms, in this case of closed socket it should
return fast.
if (X.hasCause(writeErr, SocketException.class))
- readReceipt(sock, 1);
+ readReceipt(ses, 1);
}
catch (SSLException sslErr) {
return sslErr;
@@ -1722,31 +1724,26 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
}
/**
- * Writes raw data to the socket.
+ * Writes raw data to the session socket.
*
- * @param sock Socket.
+ * @param ses IO session.
* @param data Raw data to write.
* @param timeout Socket write timeout.
* @throws IOException If IO failed or write timed out.
* @throws IgniteCheckedException If node is not yet initialized or is
stopping.
*/
- protected void writeToSocket(
- Socket sock,
+ protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
- assert sock != null;
assert data != null;
- try (SocketTimeoutObject ignored = startTimer(sock, timeout)) {
- OutputStream out = sock.getOutputStream();
-
- out.write(data);
-
- out.flush();
+ try (SocketTimeoutObject ignored = startTimer(ses, timeout)) {
+ ses.write(data);
}
catch (IOException e) {
- SSLException sslEx = checkSslException(sock, e);
+ SSLException sslEx = checkSslException(ses, e);
throw sslEx == null ? e : sslEx;
}
@@ -1788,46 +1785,37 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
TcpDiscoveryAbstractMessage msg,
long timeout
) throws IOException, IgniteCheckedException {
- Socket sock = ses.socket();
-
- assert sock != null;
assert msg != null;
- try (SocketTimeoutObject ignored = startTimer(sock, timeout)) {
+ try (SocketTimeoutObject ignored = startTimer(ses, timeout)) {
ses.writeMessage(msg);
}
catch (IgniteCheckedException e) {
- SSLException sslEx = checkSslException(sock, e);
+ SSLException sslEx = checkSslException(ses, e);
throw sslEx == null ? e : new IgniteCheckedException(sslEx);
}
}
/**
- * Writes response to the socket.
+ * Writes response to the session socket.
*
- * @param sock Socket.
+ * @param ses IO session.
* @param res Integer response.
* @param timeout Socket timeout.
* @throws IOException If IO failed or write timed out.
* @throws IgniteCheckedException If node is not yet initialized or is
stopping.
*/
- protected void writeToSocket(
- Socket sock,
+ protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
- assert sock != null;
-
- try (SocketTimeoutObject ignored = startTimer(sock, timeout)) {
- OutputStream out = sock.getOutputStream();
-
- out.write(res);
-
- out.flush();
+ try (SocketTimeoutObject ignored = startTimer(ses, timeout)) {
+ ses.write(res);
}
catch (IOException e) {
- SSLException sslEx = checkSslException(sock, e);
+ SSLException sslEx = checkSslException(ses, e);
throw (sslEx == null) ? e : sslEx;
}
@@ -1843,22 +1831,15 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
* @throws IgniteCheckedException If unmarshalling failed.
*/
protected <T extends Message> T readMessage(TcpDiscoveryIoSession ses,
long timeout) throws IOException, IgniteCheckedException {
- Socket sock = ses.socket();
-
- assert sock != null;
-
- int oldTimeout = sock.getSoTimeout();
-
try {
- sock.setSoTimeout((int)timeout);
-
- return ses.readMessage();
+ return ses.readMessage(timeout);
}
catch (IOException | IgniteCheckedException e) {
if (X.hasCause(e, SocketTimeoutException.class))
LT.warn(log, "Timed out waiting for message to be read (most
probably, the reason is " +
"long GC pauses on remote node) [curTimeout=" + timeout +
- ", rmtAddr=" + sock.getRemoteSocketAddress() + ",
rmtPort=" + sock.getPort() + ']');
+ ", rmtAddr=" + ses.socket().getRemoteSocketAddress() +
+ ", rmtPort=" + ses.socket().getPort() + ']');
StreamCorruptedException streamCorruptedCause = X.cause(e,
StreamCorruptedException.class);
@@ -1880,7 +1861,7 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
if (X.hasCause(e, ClassNotFoundException.class)) {
LT.error(log, e, "Failed to read message due to an unknown
class to unmarshal received. Unable to " +
"process the Discovery protocol. Stopping the Discovery
SPI and invoking the failure handler. " +
- "RmtAddr=" + sock.getRemoteSocketAddress() + ", rmtPort="
+ sock.getPort() + ']');
+ "RmtAddr=" + ses.socket().getRemoteSocketAddress() + ",
rmtPort=" + ses.socket().getPort() + ']');
ignite.context().failure().process(new
FailureContext(FailureType.CRITICAL_ERROR, e));
@@ -1890,58 +1871,66 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
throw e;
}
- finally {
- // Quietly restore timeout.
- try {
- sock.setSoTimeout(oldTimeout);
- }
- catch (SocketException ignored) {
- // No-op.
- }
- }
}
/**
- * Reads message delivery receipt from the socket.
+ * Reads and verifies the {@code U.IGNITE_HEADER} with which an incoming
connection is expected to start.
+ * See {@link #openSession(Socket, InetSocketAddress,
IgniteSpiOperationTimeoutHelper)} for writing this prefix.
*
- * @param sock Socket.
- * @param timeout Socket timeout for this operation.
- * @return Receipt.
- * @throws IOException If IO failed or read timed out.
+ * <p>If the {@code U.IGNITE_HEADER} is not received within the specified
timeout or its value is unexpected, the
+ * IO session remains open and its handling is left to the caller.</p>
+ *
+ * @param ses IO session.
+ * @param timeout Operation timeout.
+ * @return {@code true} if the Ignite header was successfully read within
the specified timeout, {@code false} otherwise.
+ * @throws IOException If an I/O error occurs or the read times out.
*/
- protected int readReceipt(Socket sock, long timeout) throws IOException {
- assert sock != null;
+ protected boolean readMagicHeader(TcpDiscoveryIoSession ses, long timeout)
throws IOException {
+ byte[] buf = new byte[U.IGNITE_HEADER.length];
- int oldTimeout = sock.getSoTimeout();
+ if (ses.read(buf, timeout) < buf.length) {
+ LT.warn(log, "Failed to read magic header (too few bytes received)
" +
+ "[rmtAddr=" + ses.socket().getRemoteSocketAddress() +
+ ", locAddr=" + ses.socket().getLocalSocketAddress() + ']');
- try {
- sock.setSoTimeout((int)timeout);
+ return false;
+ }
- int res = sock.getInputStream().read();
+ if (!Arrays.equals(buf, U.IGNITE_HEADER)) {
+ LT.warn(log, "Unknown connection detected (possible reasons: an
incompatible Ignite node or " +
+ "other software connecting to this Ignite port" +
+ (!isSslEnabled() ? ", or missing SSL configuration on remote
node" : "") +
+ ") [rmtAddr=" + ses.socket().getRemoteSocketAddress() +
+ ", locAddr=" + ses.socket().getLocalSocketAddress() +
+ ", rcvdHdr=" + U.byteArray2HexString(buf) + ']', true);
- if (res == -1)
- throw new EOFException();
+ return false;
+ }
+
+ return true;
+ }
- return res;
+ /**
+ * Reads message delivery receipt from the session socket.
+ *
+ * @param ses IO session.
+ * @param timeout Socket timeout for this operation.
+ * @return Receipt.
+ * @throws IOException If IO failed or read timed out.
+ */
+ protected int readReceipt(TcpDiscoveryIoSession ses, long timeout) throws
IOException {
+ try {
+ return ses.read(timeout);
}
catch (SocketTimeoutException e) {
LT.warn(log, "Timed out waiting for message delivery receipt (most
probably, the reason is " +
"in long GC pauses on remote node; consider tuning GC and
increasing 'ackTimeout' " +
"configuration property). Will retry to send message with
increased timeout " +
- "[currentTimeout=" + timeout + ", rmtAddr=" +
sock.getRemoteSocketAddress() +
- ", rmtPort=" + sock.getPort() + ']');
+ "[currentTimeout=" + timeout + ", rmtAddr=" +
ses.socket().getRemoteSocketAddress() +
+ ", rmtPort=" + ses.socket().getPort() + ']');
throw e;
}
- finally {
- // Quietly restore timeout.
- try {
- sock.setSoTimeout(oldTimeout);
- }
- catch (SocketException ignored) {
- // No-op.
- }
- }
}
/**
@@ -2503,9 +2492,9 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
}
/** Starts a timer for a socket operation. */
- private SocketTimeoutObject startTimer(Socket sock, long timeout) throws
IgniteCheckedException {
+ private SocketTimeoutObject startTimer(TcpDiscoveryIoSession ses, long
timeout) throws IgniteCheckedException {
try {
- SocketTimeoutObject obj = new SocketTimeoutObject(sock,
U.currentTimeMillis() + timeout);
+ SocketTimeoutObject obj = new SocketTimeoutObject(ses,
U.currentTimeMillis() + timeout);
addTimeoutObject(obj);
@@ -2533,7 +2522,7 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
private final IgniteUuid id = IgniteUuid.randomUuid();
/** */
- private final Socket sock;
+ private final TcpDiscoveryIoSession ses;
/** */
private final long endTime;
@@ -2542,14 +2531,14 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
private final AtomicBoolean done = new AtomicBoolean();
/**
- * @param sock Socket.
+ * @param ses IO session.
* @param endTime End time.
*/
- SocketTimeoutObject(Socket sock, long endTime) {
- assert sock != null;
+ SocketTimeoutObject(TcpDiscoveryIoSession ses, long endTime) {
+ assert ses != null;
assert endTime > 0;
- this.sock = sock;
+ this.ses = ses;
this.endTime = endTime;
}
@@ -2563,15 +2552,16 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
/** {@inheritDoc} */
@Override public void onTimeout() {
if (done.compareAndSet(false, true)) {
- // Close socket - timeout occurred.
- U.closeQuiet(sock);
+ // Close session - timeout occurred.
+ ses.close();
LT.warn(log, "Socket write has timed out (consider increasing
" +
(failureDetectionTimeoutEnabled() ?
"'IgniteConfiguration.failureDetectionTimeout'
configuration property) [" +
"failureDetectionTimeout=" +
failureDetectionTimeout() :
"'sockTimeout' configuration property)
[sockTimeout=" + sockTimeout) +
- ", rmtAddr=" + sock.getRemoteSocketAddress() + ",
rmtPort=" + sock.getPort() +
+ ", rmtAddr=" + ses.socket().getRemoteSocketAddress() +
+ ", rmtPort=" + ses.socket().getPort() +
", sockTimeout=" + sockTimeout + ']');
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientRejoinTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientRejoinTest.java
index 23be4f0f8b0..e059f8e87a4 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientRejoinTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/IgniteClientRejoinTest.java
@@ -353,15 +353,15 @@ public class IgniteClientRejoinTest extends
GridCommonAbstractTest {
*/
private class DiscoverySpi extends TcpDiscoverySpi {
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
- if (blockAll || block && sock.getPort() == 47500)
+ if (blockAll || block && ses.socket().getPort() == 47500)
throw new SocketException("Test discovery exception");
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
}
/** {@inheritDoc} */
@@ -377,27 +377,27 @@ public class IgniteClientRejoinTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
- if (blockAll || block && sock.getPort() == 47500)
+ if (blockAll || block && ses.socket().getPort() == 47500)
throw new SocketException("Test discovery exception");
- super.writeToSocket(sock, res, timeout);
+ super.writeReceipt(ses, res, timeout);
}
/** {@inheritDoc} */
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
Socket sock,
InetSocketAddress remAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteCheckedException {
- if (blockAll || block && sock.getPort() == 47500)
+ if (blockAll || block && remAddr.getPort() == 47500)
throw new SocketException("Test discovery exception");
- return super.openSocket(sock, remAddr, timeoutHelper);
+ return super.openSession(sock, remAddr, timeoutHelper);
}
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/IgniteDiscoveryMassiveNodeFailTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/IgniteDiscoveryMassiveNodeFailTest.java
index 2ddece5dc92..a12b701809d 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/IgniteDiscoveryMassiveNodeFailTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/IgniteDiscoveryMassiveNodeFailTest.java
@@ -302,17 +302,17 @@ public class IgniteDiscoveryMassiveNodeFailTest extends
GridCommonAbstractTest {
*/
private class FailDiscoverySpi extends TcpDiscoverySpi {
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
- assertNotFailedNode(sock);
+ assertNotFailedNode(ses.socket());
if (isDrop())
return;
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
}
/** {@inheritDoc} */
@@ -335,17 +335,17 @@ public class IgniteDiscoveryMassiveNodeFailTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
- assertNotFailedNode(sock);
+ assertNotFailedNode(ses.socket());
if (isDrop())
return;
- super.writeToSocket(sock, res, timeout);
+ super.writeReceipt(ses, res, timeout);
}
/**
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTopologySplitAbstractTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTopologySplitAbstractTest.java
index f85acbd4a42..a29986bfa26 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTopologySplitAbstractTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTopologySplitAbstractTest.java
@@ -207,25 +207,25 @@ public abstract class
IgniteCacheTopologySplitAbstractTest extends GridCommonAbs
}
/** {@inheritDoc} */
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
Socket sock,
InetSocketAddress remAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteCheckedException {
checkSegmented(remAddr,
timeoutHelper.nextTimeoutChunk(getSocketTimeout()));
- return super.openSocket(sock, remAddr, timeoutHelper);
+ return super.openSession(sock, remAddr, timeoutHelper);
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
- checkSegmented((InetSocketAddress)sock.getRemoteSocketAddress(),
timeout);
+
checkSegmented((InetSocketAddress)ses.socket().getRemoteSocketAddress(),
timeout);
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
}
/** {@inheritDoc} */
@@ -238,14 +238,14 @@ public abstract class
IgniteCacheTopologySplitAbstractTest extends GridCommonAbs
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
- checkSegmented((InetSocketAddress)sock.getRemoteSocketAddress(),
timeout);
+
checkSegmented((InetSocketAddress)ses.socket().getRemoteSocketAddress(),
timeout);
- super.writeToSocket(sock, res, timeout);
+ super.writeReceipt(ses, res, timeout);
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/rest/RestProcessorHangTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/rest/RestProcessorHangTest.java
index b473603babc..04e22811047 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/rest/RestProcessorHangTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/rest/RestProcessorHangTest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.internal.processors.rest;
import java.io.IOException;
-import java.net.Socket;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CountDownLatch;
@@ -30,6 +29,7 @@ import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.IgnitionEx;
import org.apache.ignite.internal.processors.rest.request.GridRestCacheRequest;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryIoSession;
import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
@@ -71,8 +71,8 @@ public class RestProcessorHangTest extends
GridCommonAbstractTest {
// Discovery spi that never allows connecting.
TestTcpDiscoverySpi discoSpi = new TestTcpDiscoverySpi() {
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
@@ -84,7 +84,7 @@ public class RestProcessorHangTest extends
GridCommonAbstractTest {
// No-op.
}
- super.writeToSocket(sock, 255, timeout);
+ super.writeReceipt(ses, 255, timeout);
}
};
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationConnectOnInitTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationConnectOnInitTest.java
index ba7507bfb64..ff835304001 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationConnectOnInitTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationConnectOnInitTest.java
@@ -179,35 +179,25 @@ public class IgniteTcpCommunicationConnectOnInitTest
extends GridCommonAbstractT
*/
private class TestDiscoverySpi extends TcpDiscoverySpi {
/** {@inheritDoc} */
- @Override protected Socket openSocket(
- InetSocketAddress sockAddr,
- IgniteSpiOperationTimeoutHelper timeoutHelper
- ) throws IOException, IgniteCheckedException {
- awaitLatch();
-
- return super.openSocket(sockAddr, timeoutHelper);
- }
-
- /** {@inheritDoc} */
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
Socket sock,
- InetSocketAddress remAddr,
+ InetSocketAddress sockAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteCheckedException {
awaitLatch();
- return super.openSocket(sock, remAddr, timeoutHelper);
+ return super.openSession(sock, sockAddr, timeoutHelper);
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
awaitLatch();
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
}
/** {@inheritDoc} */
@@ -222,14 +212,14 @@ public class IgniteTcpCommunicationConnectOnInitTest
extends GridCommonAbstractT
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
awaitLatch();
- super.writeToSocket(sock, res, timeout);
+ super.writeReceipt(ses, res, timeout);
}
/**
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/BlockTcpDiscoverySpi.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/BlockTcpDiscoverySpi.java
index 5c6060fa60f..52e81b05df9 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/BlockTcpDiscoverySpi.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/BlockTcpDiscoverySpi.java
@@ -18,7 +18,6 @@
package org.apache.ignite.spi.discovery.tcp;
import java.io.IOException;
-import java.net.Socket;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
@@ -65,8 +64,8 @@ public class BlockTcpDiscoverySpi extends TestTcpDiscoverySpi
{
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
@@ -77,7 +76,7 @@ public class BlockTcpDiscoverySpi extends TestTcpDiscoverySpi
{
apply(spiCtx.localNode(), msg);
}
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteMetricsOverflowTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteMetricsOverflowTest.java
index b6c7d582252..c041a9bc608 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteMetricsOverflowTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/IgniteMetricsOverflowTest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.spi.discovery.tcp;
import java.io.IOException;
-import java.net.Socket;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.Ignite;
@@ -47,7 +46,7 @@ public class IgniteMetricsOverflowTest extends
GridCommonAbstractTest {
private CountDownLatch slowDownLatch;
/**
- * Period of time, for which {@link TcpDiscoverySpi#readReceipt(Socket,
long)} execution is delayed on the node
+ * Period of time, for which {@link
TcpDiscoverySpi#readReceipt(TcpDiscoveryIoSession, long)} execution is delayed
on the node
* with a slow {@link DiscoverySpi}.
*/
private volatile int readReceiptDelay;
@@ -153,7 +152,7 @@ public class IgniteMetricsOverflowTest extends
GridCommonAbstractTest {
/** */
private class TestTcpDiscoverySpi extends TcpDiscoverySpi {
/** {@inheritDoc} */
- @Override protected int readReceipt(Socket sock, long timeout) throws
IOException {
+ @Override protected int readReceipt(TcpDiscoveryIoSession ses, long
timeout) throws IOException {
if (readReceiptDelay > 0) {
slowDownLatch.countDown();
@@ -165,7 +164,7 @@ public class IgniteMetricsOverflowTest extends
GridCommonAbstractTest {
}
}
- return super.readReceipt(sock, timeout);
+ return super.readReceipt(ses, timeout);
}
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java
index 96bff0e22b0..b429c35ba17 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java
@@ -351,10 +351,10 @@ public class MultiDataCenterSplitTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(Socket sock, byte[] data, long
timeout) throws IOException, IgniteCheckedException {
- tryToBlock(sock, data, timeout);
+ @Override protected void write(TcpDiscoveryIoSession ses, byte[] data,
long timeout) throws IOException, IgniteCheckedException {
+ tryToBlock(ses.socket(), data, timeout);
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
}
/** */
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/ReceivedMessagesTracker.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/ReceivedMessagesTracker.java
index f7b93eba19f..fea9c752802 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/ReceivedMessagesTracker.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/ReceivedMessagesTracker.java
@@ -17,7 +17,6 @@
package org.apache.ignite.spi.discovery.tcp;
-import java.net.Socket;
import java.util.Collections;
import java.util.Map;
import java.util.WeakHashMap;
@@ -28,18 +27,18 @@ import org.jetbrains.annotations.Nullable;
/** */
public class ReceivedMessagesTracker {
/** */
- private final Map<Socket, TcpDiscoveryAbstractMessage> msgs =
Collections.synchronizedMap(new WeakHashMap<>());
+ private final Map<TcpDiscoveryIoSession, TcpDiscoveryAbstractMessage> msgs
= Collections.synchronizedMap(new WeakHashMap<>());
/** */
public <T extends Message> T track(TcpDiscoveryIoSession ses, T msg) {
if (msg instanceof TcpDiscoveryAbstractMessage)
- msgs.put(ses.socket(), (TcpDiscoveryAbstractMessage)msg);
+ msgs.put(ses, (TcpDiscoveryAbstractMessage)msg);
return msg;
}
/** */
- public @Nullable TcpDiscoveryAbstractMessage lastFor(Socket sock) {
- return msgs.get(sock);
+ public @Nullable TcpDiscoveryAbstractMessage lastFor(TcpDiscoveryIoSession
ses) {
+ return msgs.get(ses);
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureTimeoutSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureTimeoutSelfTest.java
index 3fd9b589bc9..9efb3b52809 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureTimeoutSelfTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiFailureTimeoutSelfTest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.spi.discovery.tcp;
import java.io.IOException;
-import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Collections;
@@ -445,8 +444,8 @@ public class TcpClientDiscoverySpiFailureTimeoutSelfTest
extends TcpClientDiscov
private Exception err;
/** */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
@@ -461,8 +460,8 @@ public class TcpClientDiscoverySpiFailureTimeoutSelfTest
extends TcpClientDiscov
}
}
- if (sock.getSoTimeout() >= writeToSocketDelay)
- super.writeToSocket(sock, data, timeout);
+ if (ses.socket().getSoTimeout() >= writeToSocketDelay)
+ super.write(ses, data, timeout);
else
throw new SocketTimeoutException("Write to socket delay
timeout exception.");
}
@@ -491,8 +490,8 @@ public class TcpClientDiscoverySpiFailureTimeoutSelfTest
extends TcpClientDiscov
}
/** */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
@@ -507,8 +506,8 @@ public class TcpClientDiscoverySpiFailureTimeoutSelfTest
extends TcpClientDiscov
}
}
- if (sock.getSoTimeout() >= writeToSocketDelay)
- super.writeToSocket(sock, res, timeout);
+ if (ses.socket().getSoTimeout() >= writeToSocketDelay)
+ super.writeReceipt(ses, res, timeout);
else
throw new SocketTimeoutException("Write to socket delay
timeout exception.");
}
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 f8a0410bcfa..ca96f33f159 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
@@ -2591,11 +2591,13 @@ public class TcpClientDiscoverySpiSelfTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
+ Socket sock = ses.socket();
+
waitFor(writeLock);
TcpDiscoveryAbstractMessage msg = decodeMessage(this, data);
@@ -2603,7 +2605,7 @@ public class TcpClientDiscoverySpiSelfTest extends
GridCommonAbstractTest {
if (msg != null && !onMessage(sock, msg))
return;
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
if (msg != null && afterWrite != null)
afterWrite.apply(msg, sock);
@@ -2642,13 +2644,14 @@ public class TcpClientDiscoverySpiSelfTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
+ Socket sock,
InetSocketAddress sockAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteCheckedException {
waitFor(openSockLock);
- return super.openSocket(sockAddr, timeoutHelper);
+ return super.openSession(sock, sockAddr, timeoutHelper);
}
/**
@@ -2689,12 +2692,14 @@ public class TcpClientDiscoverySpiSelfTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
- TcpDiscoveryAbstractMessage msg = msgTracker.lastFor(sock);
+ Socket sock = ses.socket();
+
+ TcpDiscoveryAbstractMessage msg = msgTracker.lastFor(ses);
if (delayJoinAckFor != null && msg instanceof
TcpDiscoveryJoinRequestMessage) {
TcpDiscoveryJoinRequestMessage msg0 =
(TcpDiscoveryJoinRequestMessage)msg;
@@ -2713,12 +2718,12 @@ public class TcpClientDiscoverySpiSelfTest extends
GridCommonAbstractTest {
}
}
- super.writeToSocket(sock, res, timeout);
+ super.writeReceipt(ses, res, timeout);
}
/** {@inheritDoc} */
- @Override protected int readReceipt(Socket sock, long timeout) throws
IOException {
- int res = super.readReceipt(sock, timeout);
+ @Override protected int readReceipt(TcpDiscoveryIoSession ses, long
timeout) throws IOException {
+ int res = super.readReceipt(ses, timeout);
if (res != TcpDiscoveryImpl.RES_OK) {
invalidRes = true;
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoveryUnresolvedHostTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoveryUnresolvedHostTest.java
index de4dd6f7b0a..a613b758997 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoveryUnresolvedHostTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoveryUnresolvedHostTest.java
@@ -66,7 +66,10 @@ public class TcpClientDiscoveryUnresolvedHostTest extends
GridCommonAbstractTest
//Ignore.
}
- assertEquals(0, spi.getSockets().size());
+ assertFalse(spi.getSockets().isEmpty());
+
+ for (Socket sock : spi.getSockets())
+ assertTrue(sock.isClosed());
}
/**
@@ -77,33 +80,23 @@ public class TcpClientDiscoveryUnresolvedHostTest extends
GridCommonAbstractTest
Set<Socket> sockets = new HashSet<>();
/** {@inheritDoc} */
- @Override Socket createSocket() throws IOException {
- Socket sock = super.createSocket();
-
- sockets.add(sock);
-
- return sock;
- }
-
- /** {@inheritDoc} */
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
Socket sock,
InetSocketAddress remAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteCheckedException {
try {
- return super.openSocket(sock, remAddr, timeoutHelper);
+ return super.openSession(sock, remAddr, timeoutHelper);
}
catch (IgniteSpiOperationTimeoutException | IOException e) {
- if (sock.isClosed())
- sockets.remove(sock);
+ sockets.add(sock);
throw e;
}
}
/**
- * Gets list of sockets opened by this discovery spi.
+ * Gets sockets of the failed connection attempts.
*
* @return List of sockets.
*/
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryCoordinatorFailureTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryCoordinatorFailureTest.java
index 43633ef5474..dad5c4c8fd8 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryCoordinatorFailureTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryCoordinatorFailureTest.java
@@ -148,7 +148,7 @@ public class TcpDiscoveryCoordinatorFailureTest extends
GridCommonAbstractTest {
stallSpi.startStall();
- // At this point startGrid(3) cannot proceed as well because
openSocket() is blocked.
+ // At this point startGrid(3) cannot proceed as well because
openSession() is blocked.
assertFalse(fut3.isDone());
fut4.get();
@@ -205,24 +205,14 @@ public class TcpDiscoveryCoordinatorFailureTest extends
GridCommonAbstractTest {
private volatile CountDownLatch stallLatch;
/** {@inheritDoc} */
- @Override protected Socket openSocket(
- InetSocketAddress sockAddr,
- IgniteSpiOperationTimeoutHelper timeoutHelper
- ) throws IOException, IgniteCheckedException {
- checkStall();
-
- return super.openSocket(sockAddr, timeoutHelper);
- }
-
- /** {@inheritDoc} */
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
Socket sock,
- InetSocketAddress remAddr,
+ InetSocketAddress sockAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteCheckedException {
checkStall();
- return super.openSocket(sock, remAddr, timeoutHelper);
+ return super.openSession(sock, sockAddr, timeoutHelper);
}
/**
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryFailedJoinTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryFailedJoinTest.java
index 0b5eb6891e7..a53a814b6da 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryFailedJoinTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryFailedJoinTest.java
@@ -167,26 +167,15 @@ public class TcpDiscoveryFailedJoinTest extends
GridCommonAbstractTest {
*/
private static class FailTcpDiscoverySpi extends TcpDiscoverySpi {
/** {@inheritDoc} */
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
+ Socket sock,
InetSocketAddress sockAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteCheckedException {
if (sockAddr.getPort() == FAIL_PORT)
throw new SocketException("Connection refused");
- return super.openSocket(sockAddr, timeoutHelper);
- }
-
- /** {@inheritDoc} */
- @Override protected Socket openSocket(
- Socket sock,
- InetSocketAddress remAddr,
- IgniteSpiOperationTimeoutHelper timeoutHelper
- ) throws IOException, IgniteCheckedException {
- if (remAddr.getPort() == FAIL_PORT)
- throw new SocketException("Connection refused");
-
- return super.openSocket(sock, remAddr, timeoutHelper);
+ return super.openSession(sock, sockAddr, timeoutHelper);
}
}
@@ -195,13 +184,13 @@ public class TcpDiscoveryFailedJoinTest extends
GridCommonAbstractTest {
*/
private static class DropTcpDiscoverySpi extends TcpDiscoverySpi {
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
- if (sock.getPort() != FAIL_PORT)
- super.writeToSocket(sock, data, timeout);
+ if (ses.socket().getPort() != FAIL_PORT)
+ super.write(ses, data, timeout);
}
/** {@inheritDoc} */
@@ -212,13 +201,13 @@ public class TcpDiscoveryFailedJoinTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
- if (sock.getPort() != FAIL_PORT)
- super.writeToSocket(sock, res, timeout);
+ if (ses.socket().getPort() != FAIL_PORT)
+ super.writeReceipt(ses, res, timeout);
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java
index 37496f5d3b2..43996bbd6c4 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java
@@ -191,21 +191,22 @@ public class TcpDiscoveryNetworkIssuesTest extends
GridCommonAbstractTest {
IgniteEx ig1 = startGrid(NODE_1_NAME);
specialSpi = new TcpDiscoverySpi() {
- @Override protected int readReceipt(Socket sock, long timeout)
throws IOException {
- if (netBroken.get() && sock.getPort() == NODE_3_PORT)
+ @Override protected int readReceipt(TcpDiscoveryIoSession ses,
long timeout) throws IOException {
+ if (netBroken.get() && ses.socket().getPort() == NODE_3_PORT)
throw new SocketTimeoutException("Read timed out");
- return super.readReceipt(sock, timeout);
+ return super.readReceipt(ses, timeout);
}
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
+ Socket sock,
InetSocketAddress sockAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
) throws IOException, IgniteCheckedException {
if (netBroken.get() && sockAddr.getPort() == NODE_4_PORT)
throw new SocketTimeoutException("connect timed out");
- return super.openSocket(sockAddr, timeoutHelper);
+ return super.openSession(sock, sockAddr, timeoutHelper);
}
};
@@ -615,15 +616,15 @@ public class TcpDiscoveryNetworkIssuesTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
- if (dropMsg(sock))
+ if (dropMsg(ses.socket()))
return;
- super.writeToSocket(sock, res, timeout);
+ super.writeReceipt(ses, res, timeout);
}
/** {@inheritDoc} */
@@ -645,15 +646,15 @@ public class TcpDiscoveryNetworkIssuesTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
- if (dropMsg(sock))
+ if (dropMsg(ses.socket()))
return;
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
}
/**
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryPendingMessageDeliveryTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryPendingMessageDeliveryTest.java
index ec2f0c30ab2..4fabd122234 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryPendingMessageDeliveryTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryPendingMessageDeliveryTest.java
@@ -18,7 +18,6 @@
package org.apache.ignite.spi.discovery.tcp;
import java.io.IOException;
-import java.net.Socket;
import java.util.Set;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
@@ -268,13 +267,13 @@ public class TcpDiscoveryPendingMessageDeliveryTest
extends GridCommonAbstractTe
*/
private class DyingDiscoverySpi extends NoRingClosingTcpDiscoverySpi {
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void write(
+ TcpDiscoveryIoSession ses,
byte[] data,
long timeout
) throws IOException, IgniteCheckedException {
if (!blockMsgs)
- super.writeToSocket(sock, data, timeout);
+ super.write(ses, data, timeout);
}
/** {@inheritDoc} */
@@ -285,13 +284,13 @@ public class TcpDiscoveryPendingMessageDeliveryTest
extends GridCommonAbstractTe
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
if (!blockMsgs)
- super.writeToSocket(sock, res, timeout);
+ super.writeReceipt(ses, res, timeout);
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
index c71c26b0540..e7fccc4e863 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
@@ -263,7 +263,7 @@ public class TcpDiscoverySpiFailureTimeoutSelfTest extends
AbstractDiscoverySelf
private volatile IgniteSpiOperationTimeoutException err;
/** {@inheritDoc} */
- @Override protected Socket openSocket(
+ @Override protected TcpDiscoveryIoSession openSession(
Socket sock,
InetSocketAddress sockAddr,
IgniteSpiOperationTimeoutHelper timeoutHelper
@@ -291,7 +291,7 @@ public class TcpDiscoverySpiFailureTimeoutSelfTest extends
AbstractDiscoverySelf
}
}
- super.openSocket(sock, sockAddr, timeoutHelper);
+ TcpDiscoveryIoSession ses = super.openSession(sock, sockAddr,
timeoutHelper);
try {
Thread.sleep(1500);
@@ -300,7 +300,7 @@ public class TcpDiscoverySpiFailureTimeoutSelfTest extends
AbstractDiscoverySelf
// No-op.
}
- return sock;
+ return ses;
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiReconnectDelayTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiReconnectDelayTest.java
index 21c7207c6b0..0b3f253bc4c 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiReconnectDelayTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiReconnectDelayTest.java
@@ -421,15 +421,15 @@ public class TcpDiscoverySpiReconnectDelayTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void writeToSocket(
- Socket sock,
+ @Override protected void writeReceipt(
+ TcpDiscoveryIoSession ses,
int res,
long timeout
) throws IOException, IgniteCheckedException {
- if (msgTracker.lastFor(sock) instanceof
TcpDiscoveryJoinRequestMessage && failJoinReqRes.getAndDecrement() > 0)
+ if (msgTracker.lastFor(ses) instanceof
TcpDiscoveryJoinRequestMessage && failJoinReqRes.getAndDecrement() > 0)
res = RES_WAIT;
- super.writeToSocket(sock, res, timeout);
+ super.writeReceipt(ses, res, timeout);
}
/**
