You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by na...@apache.org on 2025/01/23 07:51:00 UTC
(ignite) branch ignite-2.17 updated: IGNITE-24224 Fixed NIO server broken message serialization with SSL enabled (#11819)
This is an automated email from the ASF dual-hosted git repository.
namelchev pushed a commit to branch ignite-2.17
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/ignite-2.17 by this push:
new a4bd886fea5 IGNITE-24224 Fixed NIO server broken message serialization with SSL enabled (#11819)
a4bd886fea5 is described below
commit a4bd886fea5ad635a1f5e7e859943a2af4b9dba5
Author: Mikhail Petrov <32...@users.noreply.github.com>
AuthorDate: Thu Jan 23 01:20:29 2025 +0300
IGNITE-24224 Fixed NIO server broken message serialization with SSL enabled (#11819)
(cherry picked from commit 70a8f43817b59b180256b28983c836ce429b184f)
---
.../ignite/internal/util/nio/GridNioServer.java | 2 +-
.../internal/util/nio/ssl/GridNioSslFilter.java | 29 ++--
.../internal/util/nio/ssl/GridNioSslHandler.java | 49 +++---
.../tcp/internal/TcpHandshakeExecutor.java | 13 +-
.../TcpCommunicationSpiSslVolatilePayloadTest.java | 184 +++++++++++++++++++++
.../GridAbstractCommunicationSelfTest.java | 180 ++------------------
.../communication/TestVolatilePayloadMessage.java | 145 ++++++++++++++++
.../tcp/GridTcpCommunicationSpiAbstractTest.java | 163 +++++++++++++++++-
.../tcp/GridTcpCommunicationSpiSslSelfTest.java | 6 +-
.../IgniteSpiCommunicationSelfTestSuite.java | 5 +-
10 files changed, 565 insertions(+), 211 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
index 7c86e5737a1..70562e3d4cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridNioServer.java
@@ -3321,7 +3321,7 @@ public class GridNioServer<T> {
/**
*
*/
- private static final class WriteRequestImpl implements SessionWriteRequest, SessionChangeRequest {
+ static final class WriteRequestImpl implements SessionWriteRequest, SessionChangeRequest {
/** */
private GridNioSession ses;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
index 0fcaca7a047..125aee429d9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslFilter.java
@@ -230,14 +230,15 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
}
try {
- GridNioSslHandler hnd = new GridNioSslHandler(this,
+ GridNioSslHandler hnd = new GridNioSslHandler(
+ this,
ses,
engine,
directBuf,
order,
log,
handshake,
- sslMeta.encodedBuffer());
+ sslMeta);
sslMeta.handler(hnd);
@@ -257,10 +258,7 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
hnd.handshake();
- ByteBuffer alreadyDecoded = sslMeta.decodedBuffer();
-
- if (alreadyDecoded != null)
- proceedMessageReceived(ses, alreadyDecoded);
+ processApplicationBuffer(ses, hnd.getApplicationBuffer());
}
catch (SSLException e) {
onSessionOpenedException = e;
@@ -400,14 +398,7 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
if (hnd.isHandshakeFinished())
hnd.flushDeferredWrites();
- ByteBuffer appBuf = hnd.getApplicationBuffer();
-
- appBuf.flip();
-
- if (appBuf.hasRemaining())
- proceedMessageReceived(ses, appBuf);
-
- appBuf.compact();
+ processApplicationBuffer(ses, hnd.getApplicationBuffer());
if (hnd.isInboundDone() && !hnd.isOutboundDone()) {
if (log.isDebugEnabled())
@@ -424,6 +415,16 @@ public class GridNioSslFilter extends GridNioFilterAdapter {
}
}
+ /** */
+ private void processApplicationBuffer(GridNioSession ses, ByteBuffer appBuffer) throws IgniteCheckedException {
+ appBuffer.flip();
+
+ if (appBuffer.hasRemaining())
+ proceedMessageReceived(ses, appBuffer);
+
+ appBuffer.compact();
+ }
+
/** {@inheritDoc} */
@Override public GridNioFuture<Boolean> onSessionClose(GridNioSession ses) throws IgniteCheckedException {
GridNioSslHandler hnd = sslHandler(ses);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslHandler.java
index 33000c71ed3..9ece470653d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/ssl/GridNioSslHandler.java
@@ -36,6 +36,7 @@ import org.apache.ignite.internal.util.nio.GridNioFutureImpl;
import org.apache.ignite.internal.util.nio.GridNioSession;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteInClosure;
+import org.jetbrains.annotations.Nullable;
import static javax.net.ssl.SSLEngineResult.HandshakeStatus;
import static javax.net.ssl.SSLEngineResult.HandshakeStatus.FINISHED;
@@ -109,17 +110,19 @@ class GridNioSslHandler extends ReentrantLock {
* @param directBuf Direct buffer flag.
* @param order Byte order.
* @param handshake is handshake required.
- * @param encBuf encoded buffer to be used.
+ * @param sslMeta SSL meta.
* @throws SSLException If exception occurred when starting SSL handshake.
*/
- GridNioSslHandler(GridNioSslFilter parent,
+ GridNioSslHandler(
+ GridNioSslFilter parent,
GridNioSession ses,
SSLEngine engine,
boolean directBuf,
ByteOrder order,
IgniteLogger log,
boolean handshake,
- ByteBuffer encBuf) throws SSLException {
+ GridSslMeta sslMeta
+ ) throws SSLException {
assert parent != null;
assert ses != null;
assert engine != null;
@@ -145,32 +148,21 @@ class GridNioSslHandler extends ReentrantLock {
// Allocate a little bit more so SSL engine would not return buffer overflow status.
int netBufSize = sslEngine.getSession().getPacketBufferSize() + 50;
- outNetBuf = directBuf ? ByteBuffer.allocateDirect(netBufSize) : ByteBuffer.allocate(netBufSize);
-
- outNetBuf.order(order);
-
- inNetBuf = directBuf ? ByteBuffer.allocateDirect(netBufSize) : ByteBuffer.allocate(netBufSize);
-
- inNetBuf.order(order);
-
- if (encBuf != null) {
- encBuf.flip();
-
- inNetBuf.put(encBuf); // Buffer contains bytes read but not handled by sslEngine at BlockingSslHandler.
- }
+ outNetBuf = createBuffer(netBufSize, null);
// Initially buffer is empty.
outNetBuf.position(0);
outNetBuf.limit(0);
- int appBufSize = Math.max(sslEngine.getSession().getApplicationBufferSize() + 50, netBufSize * 2);
-
- appBuf = directBuf ? ByteBuffer.allocateDirect(appBufSize) : ByteBuffer.allocate(appBufSize);
+ inNetBuf = createBuffer(netBufSize, sslMeta.encodedBuffer());
- appBuf.order(order);
+ appBuf = createBuffer(
+ Math.max(sslEngine.getSession().getApplicationBufferSize() + 50, netBufSize * 2),
+ sslMeta.decodedBuffer()
+ );
if (log.isDebugEnabled())
- log.debug("Started SSL session [netBufSize=" + netBufSize + ", appBufSize=" + appBufSize + ']');
+ log.debug("Started SSL session [netBufSize=" + outNetBuf.capacity() + ", appBufSize=" + appBuf.capacity() + ']');
}
/**
@@ -682,6 +674,21 @@ class GridNioSslHandler extends ReentrantLock {
return cp;
}
+ /** */
+ private ByteBuffer createBuffer(int size, @Nullable ByteBuffer init) {
+ if (init != null)
+ size = Math.max(size, init.remaining());
+
+ ByteBuffer buf = directBuf ? ByteBuffer.allocateDirect(size) : ByteBuffer.allocate(size);
+
+ buf.order(order);
+
+ if (init != null)
+ buf.put(init);
+
+ return buf;
+ }
+
/**
* Write request for cases while handshake is not finished yet.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpHandshakeExecutor.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpHandshakeExecutor.java
index 387ceed0fdb..bd6cbc7cd73 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpHandshakeExecutor.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/internal/TcpHandshakeExecutor.java
@@ -81,7 +81,7 @@ public class TcpHandshakeExecutor {
BlockingTransport transport = stateProvider.isSslEnabled() ?
new SslTransport(sslMeta, ch, directBuffer, log) : new TcpTransport(ch);
- ByteBuffer buf = transport.recieveNodeId();
+ ByteBuffer buf = transport.receiveNodeId();
if (buf == null)
return NEED_WAIT;
@@ -98,7 +98,7 @@ public class TcpHandshakeExecutor {
transport.sendHandshake(msg);
- buf = transport.recieveAcknowledge();
+ buf = transport.receiveAcknowledge();
long rcvCnt = buf.getLong(DIRECT_TYPE_SIZE);
@@ -125,7 +125,7 @@ public class TcpHandshakeExecutor {
* @return Buffer with {@link NodeIdMessage}.
* @throws IgniteCheckedException If failed.
*/
- ByteBuffer recieveNodeId() throws IgniteCheckedException {
+ ByteBuffer receiveNodeId() throws IgniteCheckedException {
ByteBuffer buf = ByteBuffer.allocate(NodeIdMessage.MESSAGE_FULL_SIZE)
.order(ByteOrder.LITTLE_ENDIAN);
@@ -171,7 +171,7 @@ public class TcpHandshakeExecutor {
* @return Buffer with message.
* @throws IgniteCheckedException If failed.
*/
- ByteBuffer recieveAcknowledge() throws IgniteCheckedException {
+ ByteBuffer receiveAcknowledge() throws IgniteCheckedException {
ByteBuffer buf = ByteBuffer.allocate(RecoveryLastReceivedMessage.MESSAGE_FULL_SIZE)
.order(ByteOrder.LITTLE_ENDIAN);
@@ -333,8 +333,11 @@ public class TcpHandshakeExecutor {
ByteBuffer inBuf = handler.inputBuffer();
- if (inBuf.position() > 0)
+ if (inBuf.position() > 0) {
+ inBuf.flip();
+
sslMeta.encodedBuffer(inBuf);
+ }
}
/**
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/nio/TcpCommunicationSpiSslVolatilePayloadTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/TcpCommunicationSpiSslVolatilePayloadTest.java
new file mode 100644
index 00000000000..08a5f2586c3
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/nio/TcpCommunicationSpiSslVolatilePayloadTest.java
@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.util.nio;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.internal.util.nio.ssl.BlockingSslHandler;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteRunnable;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.communication.CommunicationListener;
+import org.apache.ignite.spi.communication.CommunicationSpi;
+import org.apache.ignite.spi.communication.GridAbstractCommunicationSelfTest;
+import org.apache.ignite.spi.communication.TestVolatilePayloadMessage;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.spi.communication.tcp.internal.GridNioServerWrapper;
+import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.junit.Test;
+
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/**
+ * Tests the case when regular communications messages are sent along with the last handshake messages and SSL is enabled.
+ * It asserts that if not all received by network bytes are processed by {@link BlockingSslHandler} during the handshake
+ * phase, then all remaining bytes are properly copied to {@code GridNioSslHandler}, which replaces
+ * {@link BlockingSslHandler} after the handshake phase.
+ * The steps that can lead to mentioned above conditions:
+ * <p>
+ * 1. Node B sends a MESSAGE to Node A and stores it in the local recovery descriptor until an acknowledgment is received
+ * from Node A.
+ * <p>
+ * 2. Node A, for whatever reason, reestablishes connection with node B and starts handshake negotiation.
+ * <p>
+ * 3. Node B during the final phase of handshake sends {@link RecoveryLastReceivedMessage} and resends not acknowledged
+ * MESSAGE from step 1. But all sent bytes are divided into two network packets. Let's assume that the first packet
+ * contains all bytes related to {@link RecoveryLastReceivedMessage} and only half of the MESSAGE bytes.
+ * <p>
+ * 4. Node A decodes {@link RecoveryLastReceivedMessage} from the received network packet and finishes the handshake.
+ * But the MESSAGE cannot be processed because not enough bytes were received. So we must save remaining bytes from the
+ * first network packet, wait for the next network packet and finish MESSAGE deserialization.
+ *
+ */
+public class TcpCommunicationSpiSslVolatilePayloadTest extends GridAbstractCommunicationSelfTest<CommunicationSpi<Message>> {
+ /** */
+ private static final int TEST_ITERATION_CNT = 1000;
+
+ /** The number of messages intended to fill the network buffer during last handshake message sending. */
+ private static final int RECOVERY_DESCRIPTOR_QUEUE_MESSAGE_CNT = 50;
+
+ /** */
+ private static final AtomicInteger msgCreatedCntr = new AtomicInteger();
+
+ /** */
+ private static final AtomicInteger msgReceivedCntr = new AtomicInteger();
+
+ /** */
+ private static final Map<Integer, TestVolatilePayloadMessage> messages = new ConcurrentHashMap<>();
+
+ /** {@inheritDoc} */
+ @Override protected CommunicationSpi<Message> getSpi(int idx) {
+ return new TcpCommunicationSpi().setLocalPort(GridTestUtils.getNextCommPort(getClass()))
+ .setIdleConnectionTimeout(2000)
+ .setTcpNoDelay(true);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected CommunicationListener<Message> createMessageListener(UUID nodeId) {
+ return new TestCommunicationListener();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Map<Short, Supplier<Message>> customMessageTypes() {
+ return Collections.singletonMap(TestVolatilePayloadMessage.DIRECT_TYPE, TestVolatilePayloadMessage::new);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected boolean isSslEnabled() {
+ return true;
+ }
+
+ /** */
+ @Test
+ public void test() throws Exception {
+ ClusterNode from = nodes.get(0);
+ ClusterNode to = nodes.get(1);
+
+ for (int i = 0; i < TEST_ITERATION_CNT; i++) {
+ // Force connection to be established.
+ sendMessage(from, to, createMessage());
+
+ GridNioRecoveryDescriptor fromDesc = extractRecoveryDescriptor(from);
+ GridNioRecoveryDescriptor toDesc = extractRecoveryDescriptor(to);
+
+ // Stores multiple dummy messages in a recovery descriptor. When the connection is restored, they will be
+ // written to the network buffer along with the last handshake message.
+ // See TcpHandshakeExecutor#receiveAcknowledge
+ for (int j = 0; j < RECOVERY_DESCRIPTOR_QUEUE_MESSAGE_CNT; j++)
+ toDesc.add(new GridNioServer.WriteRequestImpl(toDesc.session(), createMessage(), false, null));
+
+ // Close connection to re-initiate handshake between nodes.
+ if (fromDesc.session() != null)
+ fromDesc.session().close();
+ }
+
+ assertTrue(waitForCondition(() -> msgCreatedCntr.get() == msgReceivedCntr.get(), 5000));
+ }
+
+ /** */
+ public GridNioRecoveryDescriptor extractRecoveryDescriptor(ClusterNode node) throws Exception {
+ CommunicationSpi<Message> spi = spis.get(node.id());
+
+ GridNioServerWrapper wrapper = U.field(spi, "nioSrvWrapper");
+
+ assertTrue(waitForCondition(() -> !wrapper.recoveryDescs().values().isEmpty(), getTestTimeout()));
+
+ return wrapper.recoveryDescs().values().stream().findFirst().get();
+ }
+
+ /** */
+ private Message createMessage() {
+ byte[] payload = new byte[ThreadLocalRandom.current().nextInt(10, 1024)];
+
+ ThreadLocalRandom.current().nextBytes(payload);
+
+ TestVolatilePayloadMessage msg = new TestVolatilePayloadMessage(msgCreatedCntr.getAndIncrement(), payload);
+
+ messages.put(msg.index(), msg);
+
+ return msg;
+ }
+
+ /** */
+ private void sendMessage(ClusterNode from, ClusterNode to, Message msg) {
+ spis.get(from.id()).sendMessage(to, msg);
+ }
+
+ /** */
+ private static class TestCommunicationListener implements CommunicationListener<Message> {
+ /** {@inheritDoc} */
+ @Override public void onMessage(UUID nodeId, Message msg, IgniteRunnable msgC) {
+ msgC.run();
+
+ if (msg instanceof TestVolatilePayloadMessage) {
+ TestVolatilePayloadMessage testMsg = (TestVolatilePayloadMessage)msg;
+
+ TestVolatilePayloadMessage expMsg = messages.get(testMsg.index());
+
+ assertNotNull(expMsg);
+
+ assertTrue(Arrays.equals(expMsg.payload(), testMsg.payload()));
+
+ msgReceivedCntr.incrementAndGet();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onDisconnected(UUID nodeId) {
+ // No-op.
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
index 80726824a78..3427867c84f 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/GridAbstractCommunicationSelfTest.java
@@ -21,11 +21,11 @@ import java.net.BindException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
-import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
-import java.util.Set;
import java.util.UUID;
+import java.util.function.Supplier;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -34,7 +34,6 @@ import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImp
import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
@@ -46,7 +45,6 @@ import org.apache.ignite.testframework.junits.GridTestKernalContext;
import org.apache.ignite.testframework.junits.IgniteMock;
import org.apache.ignite.testframework.junits.IgniteTestResources;
import org.apache.ignite.testframework.junits.spi.GridSpiAbstractTest;
-import org.junit.Test;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MACS;
@@ -55,186 +53,40 @@ import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MACS;
* @param <T> Type of communication SPI.
*/
public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationSpi<Message>> extends GridSpiAbstractTest<T> {
- /** */
- private static long msgId = 1;
-
/** */
private static final Collection<IgniteTestResources> spiRsrcs = new ArrayList<>();
- /** */
- private static final Map<UUID, Set<UUID>> msgDestMap = new HashMap<>();
-
/** */
protected static final Map<UUID, CommunicationSpi<Message>> spis = new HashMap<>();
/** */
- protected static final Collection<ClusterNode> nodes = new ArrayList<>();
-
- /** */
- private static final Object mux = new Object();
+ protected static final List<ClusterNode> nodes = new ArrayList<>();
/** */
private static GridTimeoutProcessor timeoutProcessor;
- /** */
- protected boolean useSsl;
-
- /** */
- private class MessageListener implements CommunicationListener<Message> {
- /** */
- private final UUID locNodeId;
-
- /**
- * @param locNodeId Local node ID.
- */
- MessageListener(UUID locNodeId) {
- assert locNodeId != null;
-
- this.locNodeId = locNodeId;
- }
-
- /** {@inheritDoc} */
- @Override public void onMessage(UUID nodeId, Message msg, IgniteRunnable msgC) {
- info("Received message [locNodeId=" + locNodeId + ", nodeId=" + nodeId +
- ", msg=" + msg + ']');
-
- msgC.run();
-
- if (msg instanceof GridTestMessage) {
- GridTestMessage testMsg = (GridTestMessage)msg;
-
- if (!testMsg.getSourceNodeId().equals(nodeId))
- fail("Listener nodeId not equals to message nodeId.");
-
- synchronized (mux) {
- // Get list of all recipients for the message.
- Set<UUID> recipients = msgDestMap.get(testMsg.getSourceNodeId());
-
- if (recipients != null) {
- // Remove this node from a list of recipients.
- if (!recipients.remove(locNodeId))
- fail("Received unknown message [locNodeId=" + locNodeId + ", msg=" + testMsg + ']');
-
- // If all recipients received their messages,
- // remove source nodes from sent messages map.
- if (recipients.isEmpty())
- msgDestMap.remove(testMsg.getSourceNodeId());
-
- if (msgDestMap.isEmpty())
- mux.notifyAll();
- }
- else
- fail("Received unknown message [locNodeId=" + locNodeId + ", msg=" + testMsg + ']');
- }
- }
- }
-
- /** {@inheritDoc} */
- @Override public void onDisconnected(UUID nodeId) {
- // No-op.
- }
- }
-
/** */
protected GridAbstractCommunicationSelfTest() {
super(false);
}
/**
- * @throws Exception If failed.
- */
- @Test
- public void testSendToOneNode() throws Exception {
- info(">>> Starting send to one node test. <<<");
-
- msgDestMap.clear();
-
- for (Entry<UUID, CommunicationSpi<Message>> entry : spis.entrySet()) {
- for (ClusterNode node : nodes) {
- synchronized (mux) {
- if (!msgDestMap.containsKey(entry.getKey()))
- msgDestMap.put(entry.getKey(), new HashSet<>());
-
- msgDestMap.get(entry.getKey()).add(node.id());
- }
-
- entry.getValue().sendMessage(node, new GridTestMessage(entry.getKey(), msgId++, 0));
- }
- }
-
- long now = System.currentTimeMillis();
- long endTime = now + getMaxTransmitMessagesTime();
-
- synchronized (mux) {
- while (now < endTime && !msgDestMap.isEmpty()) {
- mux.wait(endTime - now);
-
- now = System.currentTimeMillis();
- }
-
- if (!msgDestMap.isEmpty()) {
- for (Entry<UUID, Set<UUID>> entry : msgDestMap.entrySet()) {
- error("Failed to receive all messages [sender=" + entry.getKey() +
- ", dest=" + entry.getValue() + ']');
- }
- }
-
- assert msgDestMap.isEmpty() : "Some messages were not received.";
- }
- }
-
- /**
- * @throws Exception If failed.
+ * @param idx Node index.
+ * @return Spi.
*/
- @Test
- public void testSendToManyNodes() throws Exception {
- msgDestMap.clear();
-
- // Send message from each SPI to all SPI's, including itself.
- for (Entry<UUID, CommunicationSpi<Message>> entry : spis.entrySet()) {
- UUID sndId = entry.getKey();
-
- CommunicationSpi<Message> commSpi = entry.getValue();
-
- for (ClusterNode node : nodes) {
- synchronized (mux) {
- if (!msgDestMap.containsKey(sndId))
- msgDestMap.put(sndId, new HashSet<>());
-
- msgDestMap.get(sndId).add(node.id());
- }
-
- commSpi.sendMessage(node, new GridTestMessage(sndId, msgId++, 0));
- }
- }
-
- long now = System.currentTimeMillis();
- long endTime = now + getMaxTransmitMessagesTime();
-
- synchronized (mux) {
- while (now < endTime && !msgDestMap.isEmpty()) {
- mux.wait(endTime - now);
+ protected abstract CommunicationSpi<Message> getSpi(int idx);
- now = System.currentTimeMillis();
- }
+ /** */
+ protected abstract CommunicationListener<Message> createMessageListener(UUID nodeId);
- if (!msgDestMap.isEmpty()) {
- for (Entry<UUID, Set<UUID>> entry : msgDestMap.entrySet()) {
- error("Failed to receive all messages [sender=" + entry.getKey() +
- ", dest=" + entry.getValue() + ']');
- }
- }
+ /** */
+ protected abstract Map<Short, Supplier<Message>> customMessageTypes();
- assert msgDestMap.isEmpty() : "Some messages were not received.";
- }
+ /** */
+ protected boolean isSslEnabled() {
+ return false;
}
- /**
- * @param idx Node index.
- * @return Spi.
- */
- protected abstract CommunicationSpi<Message> getSpi(int idx);
-
/**
* @return Spi count.
*/
@@ -304,7 +156,7 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
MessageFactoryProvider testMsgFactory = new MessageFactoryProvider() {
@Override public void registerAll(MessageFactory factory) {
- factory.register(GridTestMessage.DIRECT_TYPE, GridTestMessage::new);
+ customMessageTypes().forEach(factory::register);
}
};
@@ -322,7 +174,7 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
GridTestUtils.setFieldValue(spi, IgniteSpiAdapter.class, "igniteInstanceName", "grid-" + i);
- if (useSsl) {
+ if (isSslEnabled()) {
IgniteMock ignite = GridTestUtils.getFieldValue(spi, IgniteSpiAdapter.class, "ignite");
IgniteConfiguration cfg = ignite.configuration()
@@ -331,7 +183,7 @@ public abstract class GridAbstractCommunicationSelfTest<T extends CommunicationS
ignite.setStaticCfg(cfg);
}
- spi.setListener(new MessageListener(rsrcs.getNodeId()));
+ spi.setListener(createMessageListener(rsrcs.getNodeId()));
node.setAttribute(ATTR_MACS, F.concat(U.allLocalMACs(), ", "));
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/TestVolatilePayloadMessage.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/TestVolatilePayloadMessage.java
new file mode 100644
index 00000000000..ffc06df72eb
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/TestVolatilePayloadMessage.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.spi.communication;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+
+/** */
+public class TestVolatilePayloadMessage implements Message {
+ /** */
+ public static final short DIRECT_TYPE = 210;
+
+ /** */
+ private int idx;
+
+ /** */
+ private byte[] payload;
+
+ /** */
+ private int payloadLen;
+
+ /** */
+ public TestVolatilePayloadMessage() {
+ // No-op.
+ }
+
+ /** */
+ public TestVolatilePayloadMessage(int idx, byte[] payload) {
+ this.idx = idx;
+ this.payload = payload;
+ this.payloadLen = payload.length;
+ }
+
+ /** */
+ public int index() {
+ return idx;
+ }
+
+ /**
+ * @return Network payload.
+ */
+ public byte[] payload() {
+ return payload;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onAckReceived() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ writer.setBuffer(buf);
+
+ if (!writer.isHeaderWritten()) {
+ if (!writer.writeHeader(directType(), fieldsCount()))
+ return false;
+
+ writer.onHeaderWritten();
+ }
+
+ switch (writer.state()) {
+ case 0:
+ if (!writer.writeInt(null, idx))
+ return false;
+
+ writer.incrementState();
+ case 1:
+ if (!writer.writeInt(null, payloadLen))
+ return false;
+
+ writer.incrementState();
+ case 2:
+ if (!writer.writeByteArray(null, payload))
+ return false;
+
+ writer.incrementState();
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ reader.setBuffer(buf);
+
+ switch (reader.state()) {
+ case 0:
+ idx = reader.readInt(null);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 1:
+ payloadLen = reader.readInt(null);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 2:
+ if (buf.remaining() < payloadLen)
+ return false;
+
+ payload = reader.readByteArray(null);
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+ }
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return DIRECT_TYPE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 2;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java
index 5752f790957..d9fe1c6f811 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiAbstractTest.java
@@ -19,22 +19,30 @@ package org.apache.ignite.spi.communication.tcp;
import java.util.ArrayList;
import java.util.BitSet;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CyclicBarrier;
+import java.util.function.Supplier;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.util.nio.GridCommunicationClient;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
+import org.apache.ignite.lang.IgniteRunnable;
import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.IgniteSpiAdapter;
+import org.apache.ignite.spi.communication.CommunicationListener;
import org.apache.ignite.spi.communication.CommunicationSpi;
import org.apache.ignite.spi.communication.GridAbstractCommunicationSelfTest;
+import org.apache.ignite.spi.communication.GridTestMessage;
import org.apache.ignite.testframework.GridTestUtils;
import org.junit.Test;
@@ -42,6 +50,9 @@ import org.junit.Test;
* Test for {@link TcpCommunicationSpi}
*/
abstract class GridTcpCommunicationSpiAbstractTest extends GridAbstractCommunicationSelfTest<CommunicationSpi<Message>> {
+ /** */
+ private static long msgId = 1;
+
/** */
private static final int SPI_COUNT = 3;
@@ -70,9 +81,98 @@ abstract class GridTcpCommunicationSpiAbstractTest extends GridAbstractCommunica
}
/** {@inheritDoc} */
+ @Override protected CommunicationListener<Message> createMessageListener(UUID nodeId) {
+ return new MessageListener(nodeId);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected Map<Short, Supplier<Message>> customMessageTypes() {
+ return Collections.singletonMap(GridTestMessage.DIRECT_TYPE, GridTestMessage::new);
+ }
+
+ /** */
+ @Test
+ public void testSendToOneNode() throws Exception {
+ info(">>> Starting send to one node test. <<<");
+
+ MessageListener.msgDestMap.clear();
+
+ for (Map.Entry<UUID, CommunicationSpi<Message>> entry : spis.entrySet()) {
+ for (ClusterNode node : nodes) {
+ synchronized (MessageListener.mux) {
+ if (!MessageListener.msgDestMap.containsKey(entry.getKey()))
+ MessageListener.msgDestMap.put(entry.getKey(), new HashSet<>());
+
+ MessageListener.msgDestMap.get(entry.getKey()).add(node.id());
+ }
+
+ entry.getValue().sendMessage(node, new GridTestMessage(entry.getKey(), msgId++, 0));
+ }
+ }
+
+ long now = System.currentTimeMillis();
+ long endTime = now + getMaxTransmitMessagesTime();
+
+ synchronized (MessageListener.mux) {
+ while (now < endTime && !MessageListener.msgDestMap.isEmpty()) {
+ MessageListener.mux.wait(endTime - now);
+
+ now = System.currentTimeMillis();
+ }
+
+ if (!MessageListener.msgDestMap.isEmpty()) {
+ for (Map.Entry<UUID, Set<UUID>> entry : MessageListener.msgDestMap.entrySet()) {
+ error("Failed to receive all messages [sender=" + entry.getKey() +
+ ", dest=" + entry.getValue() + ']');
+ }
+ }
+
+ assert MessageListener.msgDestMap.isEmpty() : "Some messages were not received.";
+ }
+ }
+
+ /** */
@Test
- @Override public void testSendToManyNodes() throws Exception {
- super.testSendToManyNodes();
+ public void testSendToManyNodes() throws Exception {
+ MessageListener.msgDestMap.clear();
+
+ // Send message from each SPI to all SPI's, including itself.
+ for (Map.Entry<UUID, CommunicationSpi<Message>> entry : spis.entrySet()) {
+ UUID sndId = entry.getKey();
+
+ CommunicationSpi<Message> commSpi = entry.getValue();
+
+ for (ClusterNode node : nodes) {
+ synchronized (MessageListener.mux) {
+ if (!MessageListener.msgDestMap.containsKey(sndId))
+ MessageListener.msgDestMap.put(sndId, new HashSet<>());
+
+ MessageListener.msgDestMap.get(sndId).add(node.id());
+ }
+
+ commSpi.sendMessage(node, new GridTestMessage(sndId, msgId++, 0));
+ }
+ }
+
+ long now = System.currentTimeMillis();
+ long endTime = now + getMaxTransmitMessagesTime();
+
+ synchronized (MessageListener.mux) {
+ while (now < endTime && !MessageListener.msgDestMap.isEmpty()) {
+ MessageListener.mux.wait(endTime - now);
+
+ now = System.currentTimeMillis();
+ }
+
+ if (!MessageListener.msgDestMap.isEmpty()) {
+ for (Map.Entry<UUID, Set<UUID>> entry : MessageListener.msgDestMap.entrySet()) {
+ error("Failed to receive all messages [sender=" + entry.getKey() +
+ ", dest=" + entry.getValue() + ']');
+ }
+ }
+
+ assert MessageListener.msgDestMap.isEmpty() : "Some messages were not received.";
+ }
// Test idle clients remove.
for (CommunicationSpi<Message> spi : spis.values()) {
@@ -184,4 +284,63 @@ abstract class GridTcpCommunicationSpiAbstractTest extends GridAbstractCommunica
fail("Failed to wait when clients are closed.");
}
}
+
+ /** */
+ private static class MessageListener implements CommunicationListener<Message> {
+ /** */
+ private static final Map<UUID, Set<UUID>> msgDestMap = new HashMap<>();
+
+ /** */
+ private static final Object mux = new Object();
+
+ /** */
+ private final UUID locNodeId;
+
+ /**
+ * @param locNodeId Local node ID.
+ */
+ MessageListener(UUID locNodeId) {
+ assert locNodeId != null;
+
+ this.locNodeId = locNodeId;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onMessage(UUID nodeId, Message msg, IgniteRunnable msgC) {
+ msgC.run();
+
+ if (msg instanceof GridTestMessage) {
+ GridTestMessage testMsg = (GridTestMessage)msg;
+
+ if (!testMsg.getSourceNodeId().equals(nodeId))
+ fail("Listener nodeId not equals to message nodeId.");
+
+ synchronized (mux) {
+ // Get list of all recipients for the message.
+ Set<UUID> recipients = msgDestMap.get(testMsg.getSourceNodeId());
+
+ if (recipients != null) {
+ // Remove this node from a list of recipients.
+ if (!recipients.remove(locNodeId))
+ fail("Received unknown message [locNodeId=" + locNodeId + ", msg=" + testMsg + ']');
+
+ // If all recipients received their messages,
+ // remove source nodes from sent messages map.
+ if (recipients.isEmpty())
+ msgDestMap.remove(testMsg.getSourceNodeId());
+
+ if (msgDestMap.isEmpty())
+ mux.notifyAll();
+ }
+ else
+ fail("Received unknown message [locNodeId=" + locNodeId + ", msg=" + testMsg + ']');
+ }
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onDisconnected(UUID nodeId) {
+ // No-op.
+ }
+ }
}
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiSslSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiSslSelfTest.java
index a80a15d771c..5c0d5253cae 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiSslSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationSpiSslSelfTest.java
@@ -24,9 +24,9 @@ import org.apache.ignite.testframework.junits.spi.GridSpiTest;
*/
@GridSpiTest(spi = TcpCommunicationSpi.class, group = "Communication SPI")
public class GridTcpCommunicationSpiSslSelfTest extends GridTcpCommunicationSpiAbstractTest {
- /** */
- public GridTcpCommunicationSpiSslSelfTest() {
- this.useSsl = true;
+ /** {@inheritDoc} */
+ @Override protected boolean isSslEnabled() {
+ return true;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiCommunicationSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiCommunicationSelfTestSuite.java
index be98eec6e00..e231898adf6 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiCommunicationSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiCommunicationSelfTestSuite.java
@@ -17,6 +17,7 @@
package org.apache.ignite.testsuites;
+import org.apache.ignite.internal.util.nio.TcpCommunicationSpiSslVolatilePayloadTest;
import org.apache.ignite.spi.communication.tcp.ClientExceptionsUtilsTest;
import org.apache.ignite.spi.communication.tcp.GridCacheDhtLockBackupSelfTest;
import org.apache.ignite.spi.communication.tcp.GridSandboxedClientWithoutNetworkTest;
@@ -115,7 +116,9 @@ import org.junit.runners.Suite;
TcpCommunicationSpiNodeLeftLoggingTest.class,
TcpCommunicationSpiInverseConnectionLoggingTest.class,
- ClientExceptionsUtilsTest.class
+ ClientExceptionsUtilsTest.class,
+
+ TcpCommunicationSpiSslVolatilePayloadTest.class,
})
public class IgniteSpiCommunicationSelfTestSuite {
}