From a0845b1caf892afd53c4419167539d1b007383e2 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 --- .../connection/TcpNioConnectionReadTests.java | 104 +++++++++++++++--- 1 file changed, 90 insertions(+), 14 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 f702e964b8..2c4f61f5be 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; @@ -26,6 +28,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; @@ -38,9 +41,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 { @@ -187,8 +192,15 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); + + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { - semaphore.release(); + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); + } return false; }, new TcpSender() { @@ -211,6 +223,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(); @@ -224,8 +242,15 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); + + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { - semaphore.release(); + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); + } return false; }, new TcpSender() { @@ -248,6 +273,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(); @@ -261,8 +292,15 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); + + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { - semaphore.release(); + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); + } return false; }, new TcpSender() { @@ -285,6 +323,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(); @@ -293,7 +337,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when no data received. - * * @throws Exception */ @Test @@ -303,8 +346,15 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); + + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { - semaphore.release(); + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); + } return false; }, new TcpSender() { @@ -325,6 +375,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(); @@ -332,7 +388,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when no data received. - * * @throws Exception */ @Test @@ -342,8 +397,15 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); + + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { - semaphore.release(); + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); + } return false; }, new TcpSender() { @@ -365,6 +427,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(); @@ -372,7 +440,6 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ @Test @@ -383,10 +450,8 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ - @Test public void testCloseCleanupStxEtx() throws Exception { ByteArrayCrLfSerializer serializer = new ByteArrayCrLfSerializer(); @@ -395,10 +460,8 @@ public class TcpNioConnectionReadTests { /** * Tests socket closure when mid-message - * * @throws Exception */ - @Test public void testCloseCleanupLengthHeader() throws Exception { ByteArrayLengthHeaderSerializer serializer = new ByteArrayLengthHeaderSerializer(); @@ -411,8 +474,15 @@ public class TcpNioConnectionReadTests { final Semaphore semaphore = new Semaphore(0); final List added = new ArrayList(); final List removed = new ArrayList(); + + final CountDownLatch errorMessageLetch = new CountDownLatch(1); + final AtomicReference errorMessageRef = new AtomicReference(); + AbstractServerConnectionFactory scf = getConnectionFactory(serializer, message -> { - responses.add(message); + if (message instanceof ErrorMessage) { + errorMessageRef.set(((ErrorMessage) message).getPayload()); + errorMessageLetch.countDown(); + } return false; }, new TcpSender() { @@ -433,6 +503,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();