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 3cfacdd43b9a9..621a234ee5428 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.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 @@ private T2> sendJoinRequests( 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 @@ private T2> sendJoinRequests( case RES_CONTINUE_JOIN: case RES_WAIT: - U.closeQuiet(sock); + U.closeQuiet(ses); return new T2<>(true, null); @@ -645,7 +644,7 @@ private T2> sendJoinRequests( if (log.isDebugEnabled()) log.debug("Received unexpected response to join request: " + joinRes.get2()); - U.closeQuiet(sock); + U.closeQuiet(ses); } } @@ -696,18 +695,16 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws 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 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws Collection 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 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws 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 @@ private static void sleepEx(long millis, Runnable before, Runnable after) throws 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 @@ private NavigableSet allVisibleNodes() { TcpDiscoveryIoSession ses = msgWorker.currSes; if (ses != null) - U.closeQuiet(ses.socket()); + U.closeQuiet(ses); } /** {@inheritDoc} */ @@ -1133,7 +1130,7 @@ private void forceStopRead() throws InterruptedException { this.stopReadLatch = stopReadLatch = new CountDownLatch(1); - U.closeQuiet(ses.socket()); + U.closeQuiet(ses); this.ses = null; this.rmtNodeId = null; @@ -1170,26 +1167,25 @@ private void forceStopRead() throws InterruptedException { 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 @@ private void forceStopRead() throws InterruptedException { "(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 @@ private void forceStopRead() throws InterruptedException { 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 @@ void ackReceived(TcpDiscoveryClientAckResponse res) { } } - U.closeQuiet(ses.socket()); + U.closeQuiet(ses); this.ses = null; @@ -1479,7 +1475,7 @@ void ackReceived(TcpDiscoveryClientAckResponse res) { 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 @@ public void cancel() { TcpDiscoveryIoSession ses = this.ses; - if (ses != null) - U.closeQuiet(ses.socket()); + U.closeQuiet(ses); } /** {@inheritDoc} */ @@ -1578,26 +1573,19 @@ public void cancel() { 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 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 @@ else if (spi.ensured(msg)) { } } 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 @@ else if (spi.ensured(msg)) { 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 @@ else if (spi.ensured(msg)) { 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 @@ else if (msg instanceof TcpDiscoveryNodeFailedMessage && } 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 @@ else if (discoMsg instanceof TcpDiscoveryCheckFailedMessage) 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 fe86a1c789b42..045a68ebd8e9b 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 @@ private boolean pingNode(TcpDiscoveryNode node) { 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())); @@ -1497,13 +1495,12 @@ else if (U.millisSinceNanos(joinStartNanos) > spi.joinTimeout) boolean openSock = 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; @@ -1564,7 +1561,7 @@ else if (U.millisSinceNanos(joinStartNanos) > spi.joinTimeout) // 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)); @@ -1652,7 +1649,7 @@ else if (U.millisSinceNanos(joinStartNanos) > spi.joinTimeout) } } finally { - U.closeQuiet(sock); + U.closeQuiet(ses); } } @@ -1993,10 +1990,7 @@ private void clearNodeAddedMessage(TcpDiscoveryAbstractMessage msg) { /** {@inheritDoc} */ @Override public void brakeConnection() { - Socket sock = msgWorker.sock; - - if (sock != null) - U.closeQuiet(sock); + U.closeQuiet(msgWorker.ses); } /** {@inheritDoc} */ @@ -2864,9 +2858,6 @@ protected class RingMessageWorker extends MessageWorker 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 @@ else if (log.isInfoEnabled()) { } 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 @@ else if (log.isInfoEnabled()) { 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 @@ else if (e.hasCause(ObjectStreamException.class) || (!sock.isClosed() && !e.hasC if (msg instanceof TcpDiscoveryConnectionCheckMessage) { ringMessageReceived(); - spi.writeToSocket(sock, RES_OK, sockTimeout); + spi.writeReceipt(ses, RES_OK, sockTimeout); continue; } @@ -6887,7 +6834,7 @@ else if (msg instanceof TcpDiscoveryClientReconnectMessage) { 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 @@ else if (msg instanceof TcpDiscoveryClientReconnectMessage) { // 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 @@ else if (msg instanceof TcpDiscoveryDuplicateIdMessage) { } 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 @@ else if (msg instanceof TcpDiscoveryAuthFailedMessage) { } 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 @@ else if (msg instanceof TcpDiscoveryCheckFailedMessage) { } 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 @@ else if (msg instanceof TcpDiscoveryRingLatencyCheckMessage) { 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 @@ else if (msg instanceof TcpDiscoveryRingLatencyCheckMessage) { 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 @@ private boolean processJoinRequestMessage( // 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 @@ private boolean processJoinRequestMessage( // 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 @@ private boolean pingJoiningNode(TcpDiscoveryNode node) { @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 @@ private class ClientMessageWorker extends MessageWorker() ); - this.sock = sock; + this.ses = ses; this.clientNodeId = clientNodeId; clientMsgSer = new TcpDiscoveryMessageSerializer(spi); @@ -7659,14 +7606,14 @@ void addMessage(TcpDiscoveryAbstractMessage msg, @Nullable byte[] msgBytes) { 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 @@ else if (msgLog.isDebugEnabled()) { } 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 @@ else if (msgLog.isDebugEnabled()) { } 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 @@ else if (msgLog.isDebugEnabled()) { U.interrupt(runner()); - U.close(sock, log); + ses.close(log); } } } @@ -7719,7 +7666,7 @@ private void writeToSocket(T2 msgT, long ti 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 @@ public boolean ping(IgniteSpiOperationTimeoutHelper timeoutHelper) throws Interr @Override protected void tearDown() { pingResult(false); - U.closeQuiet(sock); + U.closeQuiet(ses); } /** {@inheritDoc} */ @@ -8007,24 +7954,6 @@ private static class GridPingFutureAdapter extends GridFutureAdapter { 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 8c34284e87fb7..71a4995eeabcb 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.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 @@ * * 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 @@ void writeMessage(TcpDiscoveryAbstractMessage msg) throws IgniteCheckedException } } + /** + * 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 Type of the expected message. + * @return Deserialized message instance. + * @throws IgniteCheckedException If deserialization fails. + */ + 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 @@ T readMessage() throws IgniteCheckedException, IOException { 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 @@ void serializeMessage(Message m, OutputStream out) throws IOException, IgniteChe } /** - * 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 @@ private void detectSslAlert(byte b0, byte b1, InputStream in) throws IOException 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 @@ private void attachByteArray(byte[] prefixData) { 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 @@ private void attachByteArray(byte[] prefixData) { @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 a20ac2d7f01b1..f3efcf3e9a334 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 long getCoordinatorSinceTimestamp() { } /** - * @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 @@ protected Socket openSocket( * 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 @@ protected Socket openSocket( 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 @@ void validateRemoteFeatures(IgniteNodeFeatureSet rmtFeatures) throws IgniteCheck } /** - * 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 @@ protected void writeMessage( 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 @@ protected void writeToSocket( * @throws IgniteCheckedException If unmarshalling failed. */ protected 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 @@ protected T readMessage(TcpDiscoveryIoSession ses, long time 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,64 @@ protected T readMessage(TcpDiscoveryIoSession ses, long time 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} an incoming connection is expected to start with. + * See {@link #openSession(Socket, InetSocketAddress, IgniteSpiOperationTimeoutHelper)} writing this prefix. * - * @param sock Socket. - * @param timeout Socket timeout for this operation. - * @return Receipt. + * @param ses IO session. + * @param timeout Operation timeout. + * @return {@code true} if the Ignite header was successfully read during the specified timeout, + * {@code false} otherwise. * @throws IOException If IO failed or read timed 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 +2490,9 @@ TcpDiscoveryHandshakeResponse readHandshakeResponse( } /** 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 +2520,7 @@ private class SocketTimeoutObject implements IgniteSpiTimeoutObject, AutoCloseab private final IgniteUuid id = IgniteUuid.randomUuid(); /** */ - private final Socket sock; + private final TcpDiscoveryIoSession ses; /** */ private final long endTime; @@ -2542,14 +2529,14 @@ private class SocketTimeoutObject implements IgniteSpiTimeoutObject, AutoCloseab 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 +2550,16 @@ boolean cancel() { /** {@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 23be4f0f8b0a7..e059f8e87a446 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 @@ private class TcpCommunicationSpi extends org.apache.ignite.spi.communication.tc */ 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 @@ private class DiscoverySpi extends TcpDiscoverySpi { } /** {@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 2ddece5dc923c..a12b701809d5d 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 void testRecoveryOnDisconnect() throws Exception { */ 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 @@ private boolean isDrop() { } /** {@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 f85acbd4a42da..a29986bfa2674 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 @@ protected void checkSegmented(InetSocketAddress sockAddr, long timeout) throws S } /** {@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 @@ protected void checkSegmented(InetSocketAddress sockAddr, long timeout) throws S } /** {@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 b473603babcac..04e228110470c 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.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 void testNodeStopOnDiscoverySpiFailTest() throws Exception { // 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 void testNodeStopOnDiscoverySpiFailTest() throws Exception { // 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 ba7507bfb64d5..ff83530400195 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 @@ private GridNioServer startServer() throws Exception { */ 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 @@ private class TestDiscoverySpi extends TcpDiscoverySpi { } /** {@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 5c6060fa60f5d..52e81b05df960 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 @@ private synchronized void apply(ClusterNode addr, TcpDiscoveryAbstractMessage ms } /** {@inheritDoc} */ - @Override protected void writeToSocket( - Socket sock, + @Override protected void write( + TcpDiscoveryIoSession ses, byte[] data, long timeout ) throws IOException, IgniteCheckedException { @@ -77,7 +76,7 @@ private synchronized void apply(ClusterNode addr, TcpDiscoveryAbstractMessage ms 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 b6c7d58225217..c041a9bc6085f 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 void testMetricOverflow() throws Exception { /** */ 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 @@ private class TestTcpDiscoverySpi extends TcpDiscoverySpi { } } - 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 96bff0e22b004..b429c35ba17c2 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 @@ private TestTcpDiscoverySpi( } /** {@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 f7b93eba19f03..fea9c7528023d 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 @@ /** */ public class ReceivedMessagesTracker { /** */ - private final Map msgs = Collections.synchronizedMap(new WeakHashMap<>()); + private final Map msgs = Collections.synchronizedMap(new WeakHashMap<>()); /** */ public 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 3fd9b589bc919..9efb3b52809dc 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 @@ private static class TestTcpDiscoverySpi2 extends TcpDiscoverySpi { 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 @@ private static class TestTcpDiscoverySpi2 extends TcpDiscoverySpi { } } - 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 @@ private static class TestTcpDiscoverySpi2 extends TcpDiscoverySpi { } /** */ - @Override protected void writeToSocket( - Socket sock, + @Override protected void writeReceipt( + TcpDiscoveryIoSession ses, int res, long timeout ) throws IOException, IgniteCheckedException { @@ -507,8 +506,8 @@ private static class TestTcpDiscoverySpi2 extends TcpDiscoverySpi { } } - 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 f8a0410bcfa61..ca96f33f159cf 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 @@ private void pauseResumeOperation(boolean isPause, AtomicBoolean... locks) { } /** {@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 @@ private void pauseResumeOperation(boolean isPause, AtomicBoolean... locks) { 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 @@ else if (msg instanceof TcpDiscoveryClientReconnectMessage) } /** {@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 void resumeAll() { } /** {@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 void resumeAll() { } } - 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 de4dd6f7b0a71..a613b758997ea 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 void test() throws Exception { //Ignore. } - assertEquals(0, spi.getSockets().size()); + assertFalse(spi.getSockets().isEmpty()); + + for (Socket sock : spi.getSockets()) + assertTrue(sock.isClosed()); } /** @@ -77,33 +80,23 @@ private static class TestTcpDiscoverySpi extends TcpDiscoverySpi { Set 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 43633ef547436..dad5c4c8fd891 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 void testClusterFailedNewCoordinatorInitialized() throws Exception { 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 @@ private static class StallingJoinDiscoverySpi extends TcpDiscoverySpi { 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 0b5eb6891e794..a53a814b6da3f 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 @@ private void assertStartFailed(final String name) { */ 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 @@ private static class FailTcpDiscoverySpi extends TcpDiscoverySpi { */ 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 @@ private static class DropTcpDiscoverySpi extends TcpDiscoverySpi { } /** {@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 37496f5d3b209..43996bbd6c438 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 void testServerGetsSegmentedOnBecomeDangling() throws Exception { 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 @@ private boolean dropMsg(Socket sock) { } /** {@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 @@ private boolean dropMsg(Socket sock) { } /** {@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 ec2f0c30ab2c1..4fabd12223492 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 @@ private class DyingThreadDiscoverySpi extends NoRingClosingTcpDiscoverySpi { */ 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 @@ private class DyingDiscoverySpi extends NoRingClosingTcpDiscoverySpi { } /** {@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 c71c26b054063..e7fccc4e86347 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 @@ private static class TestTcpDiscoverySpi extends TcpDiscoverySpi { private volatile IgniteSpiOperationTimeoutException err; /** {@inheritDoc} */ - @Override protected Socket openSocket( + @Override protected TcpDiscoveryIoSession openSession( Socket sock, InetSocketAddress sockAddr, IgniteSpiOperationTimeoutHelper timeoutHelper @@ -291,7 +291,7 @@ else if (openSockTimeoutWait) { } } - super.openSocket(sock, sockAddr, timeoutHelper); + TcpDiscoveryIoSession ses = super.openSession(sock, sockAddr, timeoutHelper); try { Thread.sleep(1500); @@ -300,7 +300,7 @@ else if (openSockTimeoutWait) { // 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 21c7207c6b04e..0b3f253bc4cc3 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 @@ private static class FailingTcpDiscoverySpi extends TcpDiscoverySpi { } /** {@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); } /**