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 <dma...@gridgain.com> Authored: Fri Aug 14 10:35:57 2015 +0300 Committer: Denis Magda <dma...@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);
