You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ag...@apache.org on 2018/12/10 12:11:21 UTC
ignite git commit: IGNITE-4111 Send a correct response when
communication tries to send a message to a not yet joined node - Fixes #4685.
Repository: ignite
Updated Branches:
refs/heads/master 2bf36865f -> a4a66d6e2
IGNITE-4111 Send a correct response when communication tries to send a message to a not yet joined node - Fixes #4685.
Signed-off-by: Alexey Goncharuk <al...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/a4a66d6e
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/a4a66d6e
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/a4a66d6e
Branch: refs/heads/master
Commit: a4a66d6e20e2c3fda009220b53c6ee520bf3289f
Parents: 2bf3686
Author: NSAmelchev <ns...@gmail.com>
Authored: Mon Dec 10 15:09:20 2018 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Mon Dec 10 15:09:20 2018 +0300
----------------------------------------------------------------------
.../apache/ignite/internal/IgniteFeatures.java | 101 ++++++++++++++
.../apache/ignite/internal/IgniteKernal.java | 4 +
.../ignite/internal/IgniteNodeAttributes.java | 3 +
.../communication/GridIoMessageFactory.java | 8 +-
.../internal/util/nio/GridDirectParser.java | 6 +-
.../communication/tcp/TcpCommunicationSpi.java | 86 ++++++++++--
.../tcp/messages/HandshakeWaitMessage.java | 81 ++++++++++++
.../tcp/internal/TcpDiscoveryNodesRing.java | 2 +-
...iteTcpCommunicationHandshakeWaitSslTest.java | 35 +++++
...IgniteTcpCommunicationHandshakeWaitTest.java | 131 +++++++++++++++++++
.../IgniteSpiCommunicationSelfTestSuite.java | 5 +
11 files changed, 449 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java
new file mode 100644
index 0000000..4802dce
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteFeatures.java
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal;
+
+import java.util.BitSet;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+import org.apache.ignite.spi.communication.tcp.messages.HandshakeWaitMessage;
+
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_FEATURES;
+
+/**
+ * Defines supported features and check its on other nodes.
+ */
+public enum IgniteFeatures {
+ /**
+ * Support of {@link HandshakeWaitMessage} by {@link TcpCommunicationSpi}.
+ */
+ TCP_COMMUNICATION_SPI_HANDSHAKE_WAIT_MESSAGE(0);
+
+ /**
+ * Unique feature identifier.
+ */
+ private final int featureId;
+
+ /**
+ * @param featureId Feature ID.
+ */
+ IgniteFeatures(int featureId) {
+ this.featureId = featureId;
+ }
+
+ /**
+ * @return Feature ID.
+ */
+ public int getFeatureId() {
+ return featureId;
+ }
+
+ /**
+ * Checks that feature supported by node.
+ *
+ * @param clusterNode Cluster node to check.
+ * @return {@code True} if feature is declared to be supported by remote node.
+ */
+ public static boolean nodeSupports(ClusterNode clusterNode, IgniteFeatures feature) {
+ final byte[] features = clusterNode.attribute(ATTR_IGNITE_FEATURES);
+
+ return features != null && BitSet.valueOf(features).get(feature.getFeatureId());
+ }
+
+ /**
+ * Checks that feature supported by all nodes.
+ *
+ * @param nodes cluster nodes to check their feature support.
+ * @return if feature is declared to be supported by all nodes
+ */
+ public static boolean allNodesSupports(Iterable<ClusterNode> nodes, IgniteFeatures feature) {
+ for (ClusterNode next : nodes) {
+ if (!nodeSupports(next, feature))
+ return false;
+ }
+
+ return true;
+ }
+
+ /**
+ * Features supported by the current node.
+ *
+ * @return Byte array representing all supported features by current node.
+ */
+ public static byte[] allFeatures() {
+ final BitSet set = new BitSet();
+
+ for (IgniteFeatures value : IgniteFeatures.values()) {
+ final int featureId = value.getFeatureId();
+
+ assert !set.get(featureId) : "Duplicate feature ID found for [" + value + "] having same ID ["
+ + featureId + "]";
+
+ set.set(featureId);
+ }
+
+ return set.toByteArray();
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
index 15f3c6a..a78037b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java
@@ -249,6 +249,7 @@ import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_STORAGE_
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DATA_STREAMER_POOL_SIZE;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DEPLOYMENT_MODE;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_DYNAMIC_CACHE_START_ROLLBACK_SUPPORTED;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_FEATURES;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IPS;
import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_JIT_NAME;
@@ -1620,6 +1621,9 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable {
// Save transactions configuration.
add(ATTR_TX_CONFIG, cfg.getTransactionConfiguration());
+ // Supported features.
+ add(ATTR_IGNITE_FEATURES, IgniteFeatures.allFeatures());
+
// Stick in SPI versions and classes attributes.
addSpiAttributes(cfg.getCollisionSpi());
addSpiAttributes(cfg.getDiscoverySpi());
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
index 7db7fde..45ca234 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java
@@ -205,6 +205,9 @@ public final class IgniteNodeAttributes {
/** Internal attribute indicates that incoming cache requests should be validated on primary node as well. */
public static final String ATTR_VALIDATE_CACHE_REQUESTS = ATTR_CACHE + ".validate.cache.requests";
+ /** Supported features. */
+ public static final String ATTR_IGNITE_FEATURES = ATTR_PREFIX + ".features";
+
/**
* Enforces singleton.
*/
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 3f4eb18..b1c023a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -196,6 +196,7 @@ import org.apache.ignite.spi.collision.jobstealing.JobStealingRequest;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeMessage;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeMessage2;
+import org.apache.ignite.spi.communication.tcp.messages.HandshakeWaitMessage;
import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
@@ -348,6 +349,11 @@ public class GridIoMessageFactory implements MessageFactory {
break;
+ case TcpCommunicationSpi.HANDSHAKE_WAIT_MSG_TYPE:
+ msg = new HandshakeWaitMessage();
+
+ break;
+
case 0:
msg = new GridJobCancelRequest();
@@ -1114,7 +1120,7 @@ public class GridIoMessageFactory implements MessageFactory {
break;
- // [-3..119] [124..129] [-23..-27] [-36..-55]- this
+ // [-3..119] [124..129] [-23..-28] [-36..-55] - this
// [120..123] - DR
// [-4..-22, -30..-35] - SQL
// [2048..2053] - Snapshots
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
index 9e423bb..152ffe6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/nio/GridDirectParser.java
@@ -28,6 +28,8 @@ import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageReader;
import org.jetbrains.annotations.Nullable;
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.makeMessageType;
+
/**
* Parser for direct messages.
*/
@@ -76,9 +78,7 @@ public class GridDirectParser implements GridNioParser {
byte b0 = buf.get();
byte b1 = buf.get();
- short type = (short)((b1 & 0xFF) << 8 | b0 & 0xFF);
-
- msg = msgFactory.create(type);
+ msg = msgFactory.create(makeMessageType(b0, b1));
}
boolean finished = false;
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
index 3e3f7f9..044ffef 100755
--- a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/TcpCommunicationSpi.java
@@ -66,6 +66,7 @@ import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.failure.FailureType;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteFeatures;
import org.apache.ignite.internal.IgniteInternalFuture;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteKernal;
@@ -145,6 +146,7 @@ import org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationConnecti
import org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationNodeConnectionCheckFuture;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeMessage;
import org.apache.ignite.spi.communication.tcp.messages.HandshakeMessage2;
+import org.apache.ignite.spi.communication.tcp.messages.HandshakeWaitMessage;
import org.apache.ignite.spi.communication.tcp.messages.NodeIdMessage;
import org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage;
import org.apache.ignite.spi.discovery.DiscoverySpi;
@@ -157,6 +159,7 @@ import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static org.apache.ignite.failure.FailureType.CRITICAL_ERROR;
import static org.apache.ignite.failure.FailureType.SYSTEM_WORKER_TERMINATION;
import static org.apache.ignite.internal.util.nio.GridNioSessionMetaKey.SSL_META;
+import static org.apache.ignite.plugin.extensions.communication.Message.DIRECT_TYPE_SIZE;
import static org.apache.ignite.spi.communication.tcp.internal.TcpCommunicationConnectionCheckFuture.SES_FUT_META;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.ALREADY_CONNECTED;
import static org.apache.ignite.spi.communication.tcp.messages.RecoveryLastReceivedMessage.NEED_WAIT;
@@ -373,6 +376,9 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
/** Handshake message type. */
public static final short HANDSHAKE_MSG_TYPE = -3;
+ /** Handshake wait message type. */
+ public static final short HANDSHAKE_WAIT_MSG_TYPE = -28;
+
/** */
private ConnectGateway connectGate;
@@ -408,11 +414,19 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
log.info("Accepted incoming communication connection [locAddr=" + ses.localAddress() +
", rmtAddr=" + ses.remoteAddress() + ']');
- if (log.isDebugEnabled())
- log.debug("Sending local node ID to newly accepted session: " + ses);
-
try {
- ses.sendNoFuture(nodeIdMessage(), null);
+ if (ctxInitLatch.getCount() == 0 || !isHandshakeWaitSupported()) {
+ if (log.isDebugEnabled())
+ log.debug("Sending local node ID to newly accepted session: " + ses);
+
+ ses.sendNoFuture(nodeIdMessage(), null);
+ }
+ else {
+ if (log.isDebugEnabled())
+ log.debug("Sending handshake wait message to newly accepted session: " + ses);
+
+ ses.sendNoFuture(new HandshakeWaitMessage(), null);
+ }
}
catch (IgniteCheckedException e) {
U.error(log, "Failed to send message: " + e, e);
@@ -2309,12 +2323,19 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
};
GridNioMessageReaderFactory readerFactory = new GridNioMessageReaderFactory() {
+ private IgniteSpiContext context;
+
private MessageFormatter formatter;
@Override public MessageReader reader(GridNioSession ses, MessageFactory msgFactory)
throws IgniteCheckedException {
- if (formatter == null)
- formatter = getSpiContext().messageFormatter();
+ final IgniteSpiContext ctx = TcpCommunicationSpi.super.getSpiContext();
+
+ if (formatter == null || context != ctx) {
+ context = ctx;
+
+ formatter = context.messageFormatter();
+ }
assert formatter != null;
@@ -2325,11 +2346,18 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
};
GridNioMessageWriterFactory writerFactory = new GridNioMessageWriterFactory() {
+ private IgniteSpiContext context;
+
private MessageFormatter formatter;
@Override public MessageWriter writer(GridNioSession ses) throws IgniteCheckedException {
- if (formatter == null)
- formatter = getSpiContext().messageFormatter();
+ final IgniteSpiContext ctx = TcpCommunicationSpi.super.getSpiContext();
+
+ if (formatter == null || context != ctx) {
+ context = ctx;
+
+ formatter = context.messageFormatter();
+ }
assert formatter != null;
@@ -3654,6 +3682,13 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
ByteBuffer handBuff = sslHnd.applicationBuffer();
+ if (handBuff.remaining() >= DIRECT_TYPE_SIZE) {
+ short msgType = makeMessageType(handBuff.get(0), handBuff.get(1));
+
+ if (msgType == HANDSHAKE_WAIT_MSG_TYPE)
+ return NEED_WAIT;
+ }
+
if (handBuff.remaining() < NodeIdMessage.MESSAGE_FULL_SIZE) {
buf = ByteBuffer.allocate(1000);
@@ -3665,6 +3700,13 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
buf.flip();
buf = sslHnd.decode(buf);
+
+ if (handBuff.remaining() >= DIRECT_TYPE_SIZE) {
+ short msgType = makeMessageType(handBuff.get(0), handBuff.get(1));
+
+ if (msgType == HANDSHAKE_WAIT_MSG_TYPE)
+ return NEED_WAIT;
+ }
}
else
buf = handBuff;
@@ -3678,6 +3720,13 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
if (read == -1)
throw new HandshakeException("Failed to read remote node ID (connection closed).");
+ if (read >= DIRECT_TYPE_SIZE) {
+ short msgType = makeMessageType(buf.get(0), buf.get(1));
+
+ if (msgType == HANDSHAKE_WAIT_MSG_TYPE)
+ return NEED_WAIT;
+ }
+
i += read;
}
}
@@ -4036,6 +4085,17 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
return this;
}
+ /**
+ * Checks whether remote nodes support {@link HandshakeWaitMessage}.
+ *
+ * @return {@code True} if remote nodes support {@link HandshakeWaitMessage}.
+ */
+ private boolean isHandshakeWaitSupported() {
+ Collection<ClusterNode> nodes = ignite().configuration().getDiscoverySpi().getRemoteNodes();
+
+ return IgniteFeatures.allNodesSupports(nodes, IgniteFeatures.TCP_COMMUNICATION_SPI_HANDSHAKE_WAIT_MESSAGE);
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return S.toString(TcpCommunicationSpi.class, this);
@@ -4136,6 +4196,16 @@ public class TcpCommunicationSpi extends IgniteSpiAdapter implements Communicati
}
/**
+ * Concatenates the two parameter bytes to form a message type value.
+ *
+ * @param b0 The first byte.
+ * @param b1 The second byte.
+ */
+ public static short makeMessageType(byte b0, byte b1) {
+ return (short)((b1 & 0xFF) << 8 | b0 & 0xFF);
+ }
+
+ /**
* @param ignite Ignite.
*/
private static WorkersRegistry getWorkersRegistry(Ignite ignite) {
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/messages/HandshakeWaitMessage.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/messages/HandshakeWaitMessage.java b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/messages/HandshakeWaitMessage.java
new file mode 100644
index 0000000..ae29d99
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/spi/communication/tcp/messages/HandshakeWaitMessage.java
@@ -0,0 +1,81 @@
+/*
+ * 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.tcp.messages;
+
+import java.nio.ByteBuffer;
+import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageReader;
+import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
+
+import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.HANDSHAKE_WAIT_MSG_TYPE;
+
+/**
+ * Message requesting to wait until node's SPI context initialize.
+ */
+public class HandshakeWaitMessage implements Message {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Full message size (with message type) in bytes. */
+ public static final int MESSAGE_FULL_SIZE = DIRECT_TYPE_SIZE;
+
+ /**
+ * Default constructor required by {@link Message}.
+ */
+ public HandshakeWaitMessage() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onAckReceived() {
+ // No-op.
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
+ if (buf.remaining() < MESSAGE_FULL_SIZE)
+ return false;
+
+ TcpCommunicationSpi.writeMessageType(buf, directType());
+
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
+ return true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return HANDSHAKE_WAIT_MSG_TYPE;
+ }
+
+ /** {@inheritDoc} */
+ @Override public byte fieldsCount() {
+ return 0;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(HandshakeWaitMessage.class, this);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java
index 3a8ded7..e0a067f 100644
--- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java
+++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/internal/TcpDiscoveryNodesRing.java
@@ -162,7 +162,7 @@ public class TcpDiscoveryNodesRing {
* @return Collection of visible remote nodes.
*/
public Collection<TcpDiscoveryNode> visibleRemoteNodes() {
- return nodes(VISIBLE_NODES, F.remoteNodes(locNode.id()));
+ return nodes(F.remoteNodes(locNode.id()), VISIBLE_NODES);
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitSslTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitSslTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitSslTest.java
new file mode 100644
index 0000000..01a6fa4
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitSslTest.java
@@ -0,0 +1,35 @@
+/*
+ * 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.tcp;
+
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.testframework.GridTestUtils;
+
+/**
+ * Testing {@link IgniteTcpCommunicationHandshakeWaitTest} with SSL enabled.
+ */
+public class IgniteTcpCommunicationHandshakeWaitSslTest extends IgniteTcpCommunicationHandshakeWaitTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setSslContextFactory(GridTestUtils.sslFactory());
+
+ return cfg;
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitTest.java
new file mode 100644
index 0000000..f48f364
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitTest.java
@@ -0,0 +1,131 @@
+/*
+ * 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.tcp;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.IgniteInterruptedCheckedException;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddFinishedMessage;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ * Testing {@link TcpCommunicationSpi} that will send the wait handshake message on received connections until SPI
+ * context initialized.
+ */
+public class IgniteTcpCommunicationHandshakeWaitTest extends GridCommonAbstractTest {
+ /** */
+ private static final long COMMUNICATION_TIMEOUT = 1000;
+
+ /** */
+ private static final long DISCOVERY_MESSAGE_DELAY = 500;
+
+ /** */
+ private final AtomicBoolean slowNet = new AtomicBoolean();
+
+ /** */
+ private final CountDownLatch latch = new CountDownLatch(1);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setPeerClassLoadingEnabled(false);
+
+ TcpDiscoverySpi discoSpi = new SlowTcpDiscoverySpi();
+
+ cfg.setDiscoverySpi(discoSpi);
+
+ TcpCommunicationSpi commSpi = new TcpCommunicationSpi();
+
+ commSpi.setConnectTimeout(COMMUNICATION_TIMEOUT);
+ commSpi.setMaxConnectTimeout(COMMUNICATION_TIMEOUT);
+ commSpi.setReconnectCount(1);
+
+ cfg.setCommunicationSpi(commSpi);
+
+ return cfg;
+ }
+
+ /**
+ * Test that joining node will send the wait handshake message on received connections until SPI context
+ * initialized.
+ *
+ * @throws Exception If failed.
+ */
+ public void testHandshakeOnNodeJoining() throws Exception {
+ System.setProperty(IgniteSystemProperties.IGNITE_ENABLE_FORCIBLE_NODE_KILL, "true");
+
+ IgniteEx ignite = startGrid("srv1");
+
+ startGrid("srv2");
+
+ slowNet.set(true);
+
+ IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ latch.await(2 * COMMUNICATION_TIMEOUT, TimeUnit.MILLISECONDS);
+
+ Collection<ClusterNode> nodes = ignite.context().discovery().aliveServerNodes();
+
+ assertEquals(3, nodes.size());
+
+ return ignite.context().io().sendIoTest(new ArrayList<>(nodes), null, true).get();
+ });
+
+ startGrid("srv3");
+
+ fut.get();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ System.clearProperty(IgniteSystemProperties.IGNITE_ENABLE_FORCIBLE_NODE_KILL);
+ }
+
+ /** */
+ private class SlowTcpDiscoverySpi extends TcpDiscoverySpi {
+ /** {@inheritDoc} */
+ @Override protected boolean ensured(TcpDiscoveryAbstractMessage msg) {
+ if (slowNet.get() && msg instanceof TcpDiscoveryNodeAddFinishedMessage) {
+ try {
+ if (igniteInstanceName.contains("srv2") && msg.verified())
+ latch.countDown();
+
+ U.sleep(DISCOVERY_MESSAGE_DELAY);
+ }
+ catch (IgniteInterruptedCheckedException e) {
+ throw new IgniteSpiException("Thread has been interrupted.", e);
+ }
+ }
+
+ return super.ensured(msg);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/a4a66d6e/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiCommunicationSelfTestSuite.java
----------------------------------------------------------------------
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 50c0412..3bce671 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
@@ -35,6 +35,8 @@ import org.apache.ignite.spi.communication.tcp.GridTcpCommunicationSpiStartStopS
import org.apache.ignite.spi.communication.tcp.GridTcpCommunicationSpiTcpFailureDetectionSelfTest;
import org.apache.ignite.spi.communication.tcp.GridTcpCommunicationSpiTcpNoDelayOffSelfTest;
import org.apache.ignite.spi.communication.tcp.GridTcpCommunicationSpiTcpSelfTest;
+import org.apache.ignite.spi.communication.tcp.IgniteTcpCommunicationHandshakeWaitSslTest;
+import org.apache.ignite.spi.communication.tcp.IgniteTcpCommunicationHandshakeWaitTest;
import org.apache.ignite.spi.communication.tcp.IgniteTcpCommunicationRecoveryAckClosureSelfTest;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpiDropNodesTest;
import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpiFaultyClientTest;
@@ -87,6 +89,9 @@ public class IgniteSpiCommunicationSelfTestSuite extends TestSuite {
suite.addTest(new TestSuite(TcpCommunicationStatisticsTest.class));
+ suite.addTest(new TestSuite(IgniteTcpCommunicationHandshakeWaitTest.class));
+ suite.addTest(new TestSuite(IgniteTcpCommunicationHandshakeWaitSslTest.class));
+
//suite.addTest(new TestSuite(GridCacheDhtLockBackupSelfTest.class));
return suite;