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/08/10 11:24:10 UTC
ignite git commit: IGNITE-5103 Fixed TcpDiscoverySpi not to ignore
maxMissedHeartbeats property. Fixes #4446.
Repository: ignite
Updated Branches:
refs/heads/master 40f6be89b -> 1c840f590
IGNITE-5103 Fixed TcpDiscoverySpi not to ignore maxMissedHeartbeats property. Fixes #4446.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/1c840f59
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/1c840f59
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/1c840f59
Branch: refs/heads/master
Commit: 1c840f59016273e0e99c95345c3afde639ef9689
Parents: 40f6be8
Author: Evgenii Zhuravlev <ez...@gridgain.com>
Authored: Fri Aug 10 14:23:37 2018 +0300
Committer: Alexey Goncharuk <al...@gmail.com>
Committed: Fri Aug 10 14:23:37 2018 +0300
----------------------------------------------------------------------
.../ignite/spi/discovery/tcp/ServerImpl.java | 40 +++++-
.../tcp/TcpClientDiscoverySpiSelfTest.java | 10 +-
.../TcpDiscoveryClientSuspensionSelfTest.java | 135 +++++++++++++++++++
.../IgniteSpiDiscoverySelfTestSuite.java | 2 +
4 files changed, 184 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/1c840f59/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 4ba071c..3859e17 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
@@ -6787,6 +6787,9 @@ class ServerImpl extends TcpDiscoveryImpl {
/** Current client metrics. */
private volatile ClusterMetrics metrics;
+ /** Last metrics update message receive time. */
+ private volatile long lastMetricsUpdateMsgTime;
+
/** */
private final AtomicReference<GridFutureAdapter<Boolean>> pingFut = new AtomicReference<>();
@@ -6799,10 +6802,12 @@ class ServerImpl extends TcpDiscoveryImpl {
* @param log Logger.
*/
private ClientMessageWorker(Socket sock, UUID clientNodeId, IgniteLogger log) {
- super("tcp-disco-client-message-worker", log, 2000, null);
+ super("tcp-disco-client-message-worker", log, Math.max(spi.metricsUpdateFreq, 10), null);
this.sock = sock;
this.clientNodeId = clientNodeId;
+
+ lastMetricsUpdateMsgTime = U.currentTimeMillis();
}
/**
@@ -6823,6 +6828,8 @@ class ServerImpl extends TcpDiscoveryImpl {
* @param metrics New current client metrics.
*/
void metrics(ClusterMetrics metrics) {
+ lastMetricsUpdateMsgTime = U.currentTimeMillis();
+
this.metrics = metrics;
}
@@ -7003,6 +7010,37 @@ class ServerImpl extends TcpDiscoveryImpl {
U.closeQuiet(sock);
}
+
+ /** {@inheritDoc} */
+ @Override protected void noMessageLoop() {
+ if (U.currentTimeMillis() - lastMetricsUpdateMsgTime > spi.clientFailureDetectionTimeout()) {
+ TcpDiscoveryNode clientNode = ring.node(clientNodeId);
+
+ if (clientNode != null) {
+ boolean failedNode;
+
+ synchronized (mux) {
+ failedNode = failedNodes.containsKey(clientNode);
+ }
+
+ if (!failedNode) {
+ String msg = "Client node considered as unreachable " +
+ "and will be dropped from cluster, " +
+ "because no metrics update messages received in interval: " +
+ "TcpDiscoverySpi.clientFailureDetectionTimeout() ms. " +
+ "It may be caused by network problems or long GC pause on client node, try to increase this " +
+ "parameter. " +
+ "[nodeId=" + clientNodeId +
+ ", clientFailureDetectionTimeout=" + spi.clientFailureDetectionTimeout() +
+ ']';
+
+ failNode(clientNodeId, msg);
+
+ U.warn(log, msg);
+ }
+ }
+ }
+ }
}
/** */
http://git-wip-us.apache.org/repos/asf/ignite/blob/1c840f59/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java
index 2d130e1..c85e94e 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
@@ -148,7 +148,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
private boolean longSockTimeouts;
/** */
- protected long clientFailureDetectionTimeout = 1000;
+ protected long clientFailureDetectionTimeout = 5000;
/** */
private IgniteInClosure2X<TcpDiscoveryAbstractMessage, Socket> afterWrite;
@@ -263,7 +263,7 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
clientIpFinder = null;
joinTimeout = TcpDiscoverySpi.DFLT_JOIN_TIMEOUT;
netTimeout = TcpDiscoverySpi.DFLT_NETWORK_TIMEOUT;
- clientFailureDetectionTimeout = 1000;
+ clientFailureDetectionTimeout = 5000;
longSockTimeouts = false;
assert G.allGrids().isEmpty();
@@ -538,6 +538,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
((TestTcpDiscoverySpi)client.configuration().getDiscoverySpi()).resumeAll();
+ Thread.sleep(2000);
+
assert ((IgniteEx)srv1).context().discovery().pingNode(client.cluster().localNode().id());
assert ((IgniteEx)srv0).context().discovery().pingNode(client.cluster().localNode().id());
}
@@ -583,6 +585,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
* @throws Exception If failed.
*/
public void testClientReconnectOnRouterSuspendTopologyChange() throws Exception {
+ clientFailureDetectionTimeout = 20_000;
+
reconnectAfterSuspend(true);
}
@@ -1266,6 +1270,8 @@ public class TcpClientDiscoverySpiSelfTest extends GridCommonAbstractTest {
public void testTimeoutWaitingNodeAddedMessage() throws Exception {
longSockTimeouts = true;
+ clientFailureDetectionTimeout = 20_000;
+
startServerNodes(2);
final CountDownLatch cnt = new CountDownLatch(1);
http://git-wip-us.apache.org/repos/asf/ignite/blob/1c840f59/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryClientSuspensionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryClientSuspensionSelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryClientSuspensionSelfTest.java
new file mode 100644
index 0000000..a519d25
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryClientSuspensionSelfTest.java
@@ -0,0 +1,135 @@
+/*
+ * 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.discovery.tcp;
+
+import java.util.Timer;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteSystemProperties;
+import org.apache.ignite.Ignition;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ * Test for missed client metrics update messages.
+ */
+public class TcpDiscoveryClientSuspensionSelfTest extends GridCommonAbstractTest {
+ /** */
+ private static final TcpDiscoveryIpFinder IP_FINDER = new TcpDiscoveryVmIpFinder(true);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ TcpDiscoverySpi disco = new TcpDiscoverySpi();
+
+ disco.setIpFinder(IP_FINDER);
+
+ cfg.setDiscoverySpi(disco);
+
+ cfg.setMetricsUpdateFrequency(100);
+
+ cfg.setClientFailureDetectionTimeout(1000);
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ System.setProperty(IgniteSystemProperties.IGNITE_DISCO_FAILED_CLIENT_RECONNECT_DELAY, "10000");
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ System.clearProperty(IgniteSystemProperties.IGNITE_DISCO_FAILED_CLIENT_RECONNECT_DELAY);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testOneServer() throws Exception {
+ doTestClientSuspension(1);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testTwoServers() throws Exception {
+ doTestClientSuspension(2);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ public void testThreeServers() throws Exception {
+ doTestClientSuspension(3);
+ }
+
+ /**
+ * @param serverCnt Servers count.
+ * @throws Exception If failed.
+ */
+ private void doTestClientSuspension(int serverCnt) throws Exception {
+ startGrids(serverCnt);
+
+ Ignition.setClientMode(true);
+
+ Ignite client = startGrid("client");
+
+ for (int i = 0; i < serverCnt; i++)
+ assertEquals(1, grid(i).cluster().forClients().nodes().size());
+
+ Thread.sleep(2000);
+
+ for (int i = 0; i < serverCnt; i++)
+ assertEquals(1, grid(i).cluster().forClients().nodes().size());
+
+ suspendClientMetricsUpdate(client);
+
+ Thread.sleep(2000);
+
+ for (int i = 0; i < serverCnt; i++)
+ assertEquals(0, grid(i).cluster().forClients().nodes().size());
+ }
+
+ /**
+ * @param client Client.
+ */
+ private void suspendClientMetricsUpdate(Ignite client) {
+ assert client.cluster().localNode().isClient();
+
+ ClientImpl impl = U.field(client.configuration().getDiscoverySpi(), "impl");
+
+ Timer timer = U.field(impl, "timer");
+
+ timer.cancel();
+
+ System.out.println("Metrics update message suspended");
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/1c840f59/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java
index 9ba9d2d..cb284cf 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteSpiDiscoverySelfTestSuite.java
@@ -30,6 +30,7 @@ import org.apache.ignite.spi.discovery.tcp.TcpClientDiscoveryMarshallerCheckSelf
import org.apache.ignite.spi.discovery.tcp.TcpClientDiscoverySpiFailureTimeoutSelfTest;
import org.apache.ignite.spi.discovery.tcp.TcpClientDiscoverySpiMulticastTest;
import org.apache.ignite.spi.discovery.tcp.TcpClientDiscoverySpiSelfTest;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryClientSuspensionSelfTest;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryConcurrentStartTest;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryMarshallerCheckSelfTest;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoveryMultiThreadedTest;
@@ -123,6 +124,7 @@ public class IgniteSpiDiscoverySelfTestSuite extends TestSuite {
// Client connect.
suite.addTest(new TestSuite(IgniteClientConnectTest.class));
suite.addTest(new TestSuite(IgniteClientReconnectMassiveShutdownTest.class));
+ suite.addTest(new TestSuite(TcpDiscoveryClientSuspensionSelfTest.class));
// SSL.
suite.addTest(new TestSuite(TcpDiscoverySslSelfTest.class));