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

INT-3212: Fix Failed Tests From CI Builds

Browse Source 
JIRA: https://jira.springsource.org/browse/INT-3212

INT-3212 Fix Failing TCP Test

TcpNioConnections can deadlock when a fixed thread pool is being used.
The deadlock is detected after 60 seconds, but the test case failed
after 20 seconds - before the deadlock was detected.

Change the test to use a cached thread pool instead.

Also, increase concurrency by using threading in the test server
so each socket is handled on a separate thread.

Enchance connectionId to include both local and remote ports to
aid debugging.

Some minor formatting corrections.
This commit is contained in:
Artem Bilan
2013-11-20 18:58:45 +02:00
committed by Gary Russell
parent 6b50c49efe
commit 405fce672b
8 changed files with 61 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests-context.xml
View File

@@ -40,7 +40,7 @@
<chain input-channel="pollableInput1" output-channel="output">
<filter ref="typeSelector"/>
<poller fixed-delay="10000"/>
<poller fixed-delay="1000"/>
<service-activator ref="testHandler"/>
</chain>
@@ -49,7 +49,7 @@
<poller ref="topLevelPoller"/>
</chain>
<poller id="topLevelPoller" fixed-delay="5000"/>
<poller id="topLevelPoller" fixed-delay="1000"/>
<chain input-channel="beanInput" output-channel="output">
<beans:bean
@@ -67,7 +67,7 @@
</chain>
</chain>
<chain id="aggregatorChain2" input-channel="aggregatorInput" output-channel="output">
<chain id="aggregatorChain2" input-channel="aggregator2Input" output-channel="output">
<aggregator id="aggregatorWithinChain" ref="aggregatorBean" method="aggregate"/>
<chain id="nestedChain">
<filter id="filterWithinNestedChain" ref="typeSelector"/>

12
spring-integration-core/src/test/java/org/springframework/integration/config/ChainParserTests.java
View File

@@ -171,7 +171,7 @@ public class ChainParserTests {
public void chainWithAcceptingFilter() {
Message<?> message = MessageBuilder.withPayload("test").build();
this.filterInput.send(message);
Message<?> reply = this.output.receive(0);
Message<?> reply = this.output.receive(1000);
assertNotNull(reply);
assertEquals("foo", reply.getPayload());
}
@@ -188,7 +188,7 @@ public class ChainParserTests {
public void chainWithHeaderEnricher() {
Message<?> message = MessageBuilder.withPayload(123).build();
this.headerEnricherInput.send(message);
Message<?> reply = this.replyOutput.receive(0);
Message<?> reply = this.replyOutput.receive(1000);
assertNotNull(reply);
assertEquals("foo", reply.getPayload());
assertEquals("ABC", reply.getHeaders().getCorrelationId());
@@ -239,8 +239,8 @@ public class ChainParserTests {
Message<?> message2 = MessageBuilder.withPayload(123).build();
this.payloadTypeRouterInput.send(message1);
this.payloadTypeRouterInput.send(message2);
Message<?> reply1 = this.strings.receive(0);
Message<?> reply2 = this.numbers.receive(0);
Message<?> reply1 = this.strings.receive(1000);
Message<?> reply2 = this.numbers.receive(1000);
assertNotNull(reply1);
assertNotNull(reply2);
assertEquals("test", reply1.getPayload());
@@ -253,8 +253,8 @@ public class ChainParserTests {
Message<?> message2 = MessageBuilder.withPayload(123).setHeader("routingHeader", "numbers").build();
this.headerValueRouterInput.send(message1);
this.headerValueRouterInput.send(message2);
Message<?> reply1 = this.strings.receive(0);
Message<?> reply2 = this.numbers.receive(0);
Message<?> reply1 = this.strings.receive(1000);
Message<?> reply2 = this.numbers.receive(1000);
assertNotNull(reply1);
assertNotNull(reply2);
assertEquals("test", reply1.getPayload());

15
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/AbstractConnectionFactory.java
View File

@@ -125,6 +125,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
this.port = port;
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;
}
@@ -413,6 +414,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
*/
public abstract void close();
@Override
public void start() {
if (logger.isInfoEnabled()) {
logger.info("started " + this);
@@ -438,6 +440,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
/**
* Stops the server.
*/
@Override
public void stop() {
this.active = false;
this.close();
@@ -547,7 +550,6 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
else {
if (logger.isWarnEnabled()) {
logger.warn("Timing out TcpNioConnection " +
this.port + " : " +
connection.getConnectionId());
}
connection.publishConnectionExceptionEvent(new SocketTimeoutException("Timing out connection"));
@@ -581,16 +583,19 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
connection = (TcpNioConnection) key.attachment();
connection.setLastRead(System.currentTimeMillis());
this.taskExecutor.execute(new Runnable() {
@Override
public void run() {
try {
connection.readPacket();
} catch (Exception e) {
}
catch (Exception e) {
if (connection.isOpen()) {
logger.error("Exception on read " +
connection.getConnectionId() + " " +
e.getMessage());
connection.close();
} else {
}
else {
logger.debug("Connection closed");
}
}
@@ -633,6 +638,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
throw new UnsupportedOperationException("Nio server factory must override this method");
}
@Override
public int getPhase() {
return 0;
}
@@ -641,10 +647,12 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
* We are controlled by the startup options of
* the bound endpoint.
*/
@Override
public boolean isAutoStartup() {
return false;
}
@Override
public void stop(Runnable callback) {
stop();
callback.run();
@@ -688,6 +696,7 @@ public abstract class AbstractConnectionFactory extends IntegrationObjectSupport
this.removeClosedConnectionsAndReturnOpenConnectionIds();
}
@Override
public boolean isRunning() {
return this.active;
}

9
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpConnectionSupport.java
View File

@@ -118,15 +118,18 @@ public abstract class TcpConnectionSupport implements TcpConnection {
this.hostAddress = inetAddress.getHostAddress();
if (lookupHost) {
this.hostName = inetAddress.getHostName();
} else {
}
else {
this.hostName = this.hostAddress;
}
}
int port = socket.getPort();
this.connectionId = this.hostName + ":" + port + ":" + UUID.randomUUID().toString();
int localPort = socket.getLocalPort();
this.connectionId = this.hostName + ":" + port + ":" + localPort + ":" + UUID.randomUUID().toString();
try {
this.soLinger = socket.getSoLinger();
} catch (SocketException e) { }
}
catch (SocketException e) { }
this.applicationEventPublisher = applicationEventPublisher;
if (connectionFactoryName != null) {
this.connectionFactoryName = connectionFactoryName;

37
spring-integration-ip/src/test/java/org/springframework/integration/ip/tcp/TcpSendingMessageHandlerTests.java
View File

@@ -35,6 +35,7 @@ import java.util.List;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
@@ -805,21 +806,39 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
final Semaphore semaphore = new Semaphore(0);
final AtomicBoolean done = new AtomicBoolean();
final List<Socket> serverSockets = new ArrayList<Socket>();
Executors.newSingleThreadExecutor().execute(new Runnable() {
final ExecutorService exec = Executors.newCachedThreadPool();
exec.execute(new Runnable() {
@Override
public void run() {
try {
ServerSocket server = ServerSocketFactory.getDefault().createServerSocket(port, 100);
latch.countDown();
for (int i = 0; i < 100; i++) {
Socket socket = server.accept();
final Socket socket = server.accept();
serverSockets.add(socket);
semaphore.release();
byte[] b = new byte[9];
readFully(socket.getInputStream(), b);
b = ("Reply" + i + "\r\n").getBytes();
socket.getOutputStream().write(b);
socket.close();
final int j = i;
exec.execute(new Runnable() {
@Override
public void run() {
semaphore.release();
byte[] b = new byte[9];
try {
readFully(socket.getInputStream(), b);
b = ("Reply" + j + "\r\n").getBytes();
socket.getOutputStream().write(b);
}
catch (IOException e) {
e.printStackTrace();
}
finally {
try {
socket.close();
}
catch (IOException e) { }
}
}
});
}
server.close();
} catch (Exception e) {
@@ -836,7 +855,7 @@ public class TcpSendingMessageHandlerTests extends AbstractTcpChannelAdapterTest
ccf.setDeserializer(serializer);
ccf.setSoTimeout(10000);
ccf.setSingleUse(true);
ccf.setTaskExecutor(Executors.newFixedThreadPool(100));
ccf.setTaskExecutor(Executors.newCachedThreadPool());
ccf.start();
TcpSendingMessageHandler handler = new TcpSendingMessageHandler();
handler.setConnectionFactory(ccf);

2
spring-integration-ip/src/test/resources/log4j.properties
View File

@@ -2,7 +2,7 @@ log4j.rootCategory=WARN, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%c{1} [%t] : %m%n
log4j.appender.stdout.layout.ConversionPattern=%d %c{1} [%t] : %m%n
log4j.category.org.springframework.integration=WARN
log4j.category.org.springframework.integration.ip=WARN

6
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2013 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. You may obtain a copy of the License at
@@ -15,6 +15,7 @@ package org.springframework.integration.jdbc;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -23,6 +24,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.hamcrest.Matchers;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -86,7 +88,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
Service.fail = true;
input.send(new GenericMessage<String>("foo"));
Service.await(1000);
assertEquals(1, Service.messages.size());
assertThat(Service.messages.size(), Matchers.greaterThanOrEqualTo(1));
// After a rollback in the poller the message is still waiting to be delivered
// but unless we use a transaction here there is a chance that the queue will
// appear empty....

2
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/StoredProcPollingChannelAdapterWithSpringContextIntegrationTests-context.xml
View File

@@ -73,7 +73,7 @@
<int:channel id="outputChannel"/>
<!-- <int:channel id="errorChannel" /> -->
<int:service-activator id="consumerEndpoint" input-channel="outputChannel" ref="consumer" />
<int:service-activator id="consumerEndpoint" input-channel="outputChannel" ref="consumer" phase="-100"/>
<bean id="consumer" class="org.springframework.integration.jdbc.StoredProcPollingChannelAdapterWithSpringContextIntegrationTests$Consumer"/>
<int:logging-channel-adapter channel="errorChannel" log-full-message="true"/>