Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix Race Conditions in TcpNioConnectionReadTests

Browse Source 
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
This commit is contained in:
Artem Bilan
2016-12-14 17:20:12 -05:00
committed by Gary Russell
parent 6a51da65e6
commit 3127f25bfe
1 changed files with 91 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files

135
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/connection/TcpNioConnectionReadTests.java
View File

@@ -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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
semaphore.release();
return false;
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
semaphore.release();
return false;
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
semaphore.release();
return false;
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
semaphore.release();
return false;
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
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<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
AbstractServerConnectionFactory scf = getConnectionFactory(serializer, new TcpListener() {
@Override
public boolean onMessage(Message<?> message) {
semaphore.release();
return false;
final CountDownLatch errorMessageLetch = new CountDownLatch(1);
final AtomicReference<Throwable> errorMessageRef = new AtomicReference<Throwable>();
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<Message<?>> responses = new ArrayList<Message<?>>();
final Semaphore semaphore = new Semaphore(0);
final List<TcpConnection> added = new ArrayList<TcpConnection>();
final List<TcpConnection> removed = new ArrayList<TcpConnection>();
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<Throwable> errorMessageRef = new AtomicReference<Throwable>();
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();