From 3127f25bfed6565e5fd7c8fb53c9bb346a7f8499 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 14 Dec 2016 17:20:12 -0500 Subject: [PATCH] Fix Race Conditions in TcpNioConnectionReadTests https://build.spring.io/browse/INT-SI43X-60 After introduction the `sendErrorToListener()` function for the `TcpConnectionSupport`, all the tests in the `TcpNioConnectionReadTests` have flaw to be affected by the race condition to release semaphore in the `onMessage()` and don't get a connection as closed yet. * Fix the tests adding `CountDownLatch` for the `ErrorMessage` and `AtomicReference` to assert an Exception message afterwards. **Cherry-pick to 4.3.x** Add Travis diagnostic to determine what is the difference if that Revert Travis diagnostic Looks like would be better to wait for CI failure to determine the root of sporadic cause Conflicts: spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java --- .../connection/TcpNioConnectionReadTests.java | 135 ++++++++++++------ 1 file changed, 91 insertions(+), 44 deletions(-) diff --git a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java index e85f9142e3..c7ba6dd6fc 100644 --- a/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java +++ b/spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,7 +16,9 @@ package org.springframework.integration.ip.tcp.connection; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; import static org.mockito.Mockito.mock; @@ -27,6 +29,7 @@ import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import javax.net.SocketFactory; @@ -40,9 +43,11 @@ import org.springframework.integration.ip.tcp.serializer.ByteArrayStxEtxSerializ import org.springframework.integration.ip.util.SocketTestUtils; import org.springframework.integration.ip.util.TestingUtilities; import org.springframework.messaging.Message; +import org.springframework.messaging.support.ErrorMessage; /** * @author Gary Russell + * @author Artem Bilan * @since 2.0 */ public class TcpNioConnectionReadTests { @@ -210,14 +215,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -239,6 +246,12 @@ public class TcpNioConnectionReadTests { CountDownLatch done = SocketTestUtils.testSendLengthOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("Message length 2147483647 exceeds max message length: 2048")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -252,14 +265,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -281,6 +296,12 @@ public class TcpNioConnectionReadTests { CountDownLatch done = SocketTestUtils.testSendStxEtxOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("ETX not found before max message length: 1024")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -294,14 +315,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -323,6 +346,12 @@ public class TcpNioConnectionReadTests { CountDownLatch done = SocketTestUtils.testSendCrLfOverflow(scf.getPort()); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("CRLF not found before max message length: 1024")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -331,7 +360,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when no data received. - * * @throws Exception */ @Test @@ -341,14 +369,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -368,6 +398,12 @@ public class TcpNioConnectionReadTests { socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("Connection is closed")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -375,7 +411,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when no data received. - * * @throws Exception */ @Test @@ -385,14 +420,16 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - semaphore.release(); - return false; + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } - + return false; }, new TcpSender() { @Override @@ -413,6 +450,12 @@ public class TcpNioConnectionReadTests { socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("Connection is closed")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop(); @@ -420,7 +463,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ @Test @@ -431,10 +473,8 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ - @Test public void testCloseCleanupStxEtx() throws Exception { ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -443,10 +483,8 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ - @Test public void testCloseCleanupLengthHeader() throws Exception { ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); @@ -455,16 +493,19 @@ public class TcpNioConnectionReadTests { private void testClosureMidMessageGuts(AbstractByteArraySerializer serializer, String shortMessage) throws Exception { - final List> responses = new ArrayList>(); final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); - AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() { - @Override - public boolean onMessage(Message message) { - responses.add(message); - return false; + + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); } + return false; }, new TcpSender() { @Override public void addNewConnection(TcpConnection connection) { @@ -482,6 +523,12 @@ public class TcpNioConnectionReadTests { socket.close(); whileOpen(semaphore, added); assertEquals(1, added.size()); + + assertTrue(errorMessageLetch.await(10, TimeUnit.SECONDS)); + + assertThat(errorMessageRef.get().getMessage(), + containsString("Connection is closed")); + assertTrue(semaphore.tryAcquire(10000, TimeUnit.MILLISECONDS)); assertTrue(removed.size() > 0); scf.stop();