You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2015/08/14 09:36:38 UTC
[2/2] incubator-ignite git commit: ignite-1229: added tests that
check ping interruption
ignite-1229: added tests that check ping interruption
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/23dc6558
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/23dc6558
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/23dc6558
Branch: refs/heads/ignite-1229
Commit: 23dc6558e771382a244cd5d3ec7abe5ddbad1f61
Parents: 4381bf7
Author: Denis Magda <dm...@gridgain.com>
Authored: Fri Aug 14 10:35:57 2015 +0300
Committer: Denis Magda <dm...@gridgain.com>
Committed: Fri Aug 14 10:35:57 2015 +0300
----------------------------------------------------------------------
.../spi/discovery/tcp/TcpDiscoverySpi.java | 6 +-
.../spi/discovery/tcp/TcpDiscoverySelfTest.java | 151 ++++++++++++++++++-
.../TcpDiscoverySpiFailureTimeoutSelfTest.java | 5 +-
3 files changed, 156 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/23dc6558/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
----------------------------------------------------------------------
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 65ab8fd..2f3d410 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
@@ -1184,7 +1184,7 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements DiscoverySpi, T
* @throws IOException If failed.
* @throws IgniteSpiOperationTimeoutException In case of timeout.
*/
- Socket openSocket(Socket sock, InetSocketAddress remAddr, IgniteSpiOperationTimeoutHelper timeoutHelper)
+ protected Socket openSocket(Socket sock, InetSocketAddress remAddr, IgniteSpiOperationTimeoutHelper timeoutHelper)
throws IOException, IgniteSpiOperationTimeoutException {
assert remAddr != null;
@@ -1277,8 +1277,8 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter implements DiscoverySpi, T
* @throws IOException If IO failed or write timed out.
* @throws IgniteCheckedException If marshalling failed.
*/
- protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout)
- throws IOException, IgniteCheckedException {
+ protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout) throws IOException,
+ IgniteCheckedException {
writeToSocket(sock, msg, new GridByteArrayOutputStream(8 * 1024), timeout); // 8K.
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/23dc6558/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
index 9a44c24..e5118e3 100644
--- a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySelfTest.java
@@ -70,7 +70,8 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
@Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
IgniteConfiguration cfg = super.getConfiguration(gridName);
- TcpDiscoverySpi spi = new TcpDiscoverySpi();
+ TcpDiscoverySpi spi = gridName.contains("testPingInterruptedOnNodeFailedFailingNode") ?
+ new TestTcpDiscoverySpi() : new TcpDiscoverySpi();
discoMap.put(gridName, spi);
@@ -128,6 +129,8 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
if (U.isMacOs())
spi.setLocalAddress(F.first(U.allLocalIps()));
}
+ else if (gridName.contains("testPingInterruptedOnNodeFailedPingingNode"))
+ cfg.setFailureDetectionTimeout(30_000);
return cfg;
}
@@ -339,6 +342,152 @@ public class TcpDiscoverySelfTest extends GridCommonAbstractTest {
/**
* @throws Exception If any error occurs.
*/
+ public void testPingInterruptedOnNodeFailed() throws Exception {
+ try {
+ final Ignite pingingNode = startGrid("testPingInterruptedOnNodeFailedPingingNode");
+ final Ignite failedNode = startGrid("testPingInterruptedOnNodeFailedFailingNode");
+ startGrid("testPingInterruptedOnNodeFailedSimpleNode");
+
+ ((TestTcpDiscoverySpi)failedNode.configuration().getDiscoverySpi()).ignorePingResponse = true;
+
+ final CountDownLatch pingLatch = new CountDownLatch(1);
+
+ final CountDownLatch eventLatch = new CountDownLatch(1);
+
+ final AtomicBoolean pingRes = new AtomicBoolean(true);
+
+ final AtomicBoolean failRes = new AtomicBoolean(false);
+
+ long startTs = System.currentTimeMillis();
+
+ pingingNode.events().localListen(
+ new IgnitePredicate<Event>() {
+ @Override public boolean apply(Event event) {
+ if (((DiscoveryEvent)event).eventNode().id().equals(failedNode.cluster().localNode().id())) {
+ failRes.set(true);
+ eventLatch.countDown();
+ }
+
+ return true;
+ }
+ },
+ EventType.EVT_NODE_FAILED);
+
+ IgniteInternalFuture<?> pingFut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ pingLatch.countDown();
+
+ pingRes.set(pingingNode.configuration().getDiscoverySpi().pingNode(
+ failedNode.cluster().localNode().id()));
+
+ return null;
+ }
+ }, 1);
+
+ IgniteInternalFuture<?> failingFut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ pingLatch.await();
+
+ Thread.sleep(3000);
+
+ ((TestTcpDiscoverySpi)failedNode.configuration().getDiscoverySpi()).simulateNodeFailure();
+
+ return null;
+ }
+ }, 1);
+
+ failingFut.get();
+ pingFut.get();
+
+ assertFalse(pingRes.get());
+
+ assertTrue(System.currentTimeMillis() - startTs <
+ pingingNode.configuration().getFailureDetectionTimeout() / 2);
+
+ assertTrue(eventLatch.await(7, TimeUnit.SECONDS));
+ assertTrue(failRes.get());
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+
+ /**
+ * @throws Exception If any error occurs.
+ */
+ public void testPingInterruptedOnNodeLeft() throws Exception {
+ try {
+ final Ignite pingingNode = startGrid("testPingInterruptedOnNodeFailedPingingNode");
+ final Ignite leftNode = startGrid("testPingInterruptedOnNodeFailedFailingNode");
+ startGrid("testPingInterruptedOnNodeFailedSimpleNode");
+
+ ((TestTcpDiscoverySpi)leftNode.configuration().getDiscoverySpi()).ignorePingResponse = true;
+
+ final CountDownLatch pingLatch = new CountDownLatch(1);
+
+ final AtomicBoolean pingRes = new AtomicBoolean(true);
+
+ long startTs = System.currentTimeMillis();
+
+ IgniteInternalFuture<?> pingFut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ pingLatch.countDown();
+
+ pingRes.set(pingingNode.configuration().getDiscoverySpi().pingNode(
+ leftNode.cluster().localNode().id()));
+
+ return null;
+ }
+ }, 1);
+
+ IgniteInternalFuture<?> stoppingFut = multithreadedAsync(
+ new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ pingLatch.await();
+
+ Thread.sleep(3000);
+
+ stopGrid("testPingInterruptedOnNodeFailedFailingNode");
+
+ return null;
+ }
+ }, 1);
+
+ stoppingFut.get();
+ pingFut.get();
+
+ assertFalse(pingRes.get());
+
+ assertTrue(System.currentTimeMillis() - startTs <
+ pingingNode.configuration().getFailureDetectionTimeout() / 2);
+ }
+ finally {
+ stopAllGrids();
+ }
+ }
+
+ /**
+ *
+ */
+ private static class TestTcpDiscoverySpi extends TcpDiscoverySpi {
+ /** */
+ private boolean ignorePingResponse;
+
+ protected void writeToSocket(Socket sock, TcpDiscoveryAbstractMessage msg, long timeout) throws IOException,
+ IgniteCheckedException {
+ if (msg instanceof TcpDiscoveryPingResponse && ignorePingResponse)
+ return;
+ else
+ super.writeToSocket(sock, msg, timeout);
+ }
+ }
+
+ /**
+ * @throws Exception If any error occurs.
+ */
public void testNodeAdded() throws Exception {
try {
final Ignite g1 = startGrid(1);
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/23dc6558/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpiFailureTimeoutSelfTest.java
----------------------------------------------------------------------
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 fbea187..630f2fd 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
@@ -305,7 +305,8 @@ public class TcpDiscoverySpiFailureTimeoutSelfTest extends AbstractDiscoverySelf
/** {@inheritDoc} */
- @Override protected Socket openSocket(InetSocketAddress sockAddr, IgniteSpiOperationTimeoutHelper timeoutHelper)
+ @Override protected Socket openSocket(Socket sock, InetSocketAddress sockAddr,
+ IgniteSpiOperationTimeoutHelper timeoutHelper)
throws IOException, IgniteSpiOperationTimeoutException {
if (openSocketTimeout) {
@@ -330,7 +331,7 @@ public class TcpDiscoverySpiFailureTimeoutSelfTest extends AbstractDiscoverySelf
}
}
- Socket sock = super.openSocket(sockAddr, timeoutHelper);
+ super.openSocket(sock, sockAddr, timeoutHelper);
try {
Thread.sleep(1500);