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

Fix JDBC tests

Browse Source 
https://build.spring.io/browse/INT-MJATS41-1279

Looks like there is a race condition when our polling rate around
data base is too fast and we have access to the DB files even during
application context close.

* Stop polling channel adapter in the tests explicitly after test methods
* Increase some tests performance decreasing timeouts to wait
* Increase timeouts and performance in the `StreamTransformerParserTests`

# Conflicts:
#	build.gradle
#	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml
#	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests.java
#	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml
#	spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java
#	spring-integration-stomp/src/test/java/org/springframework/integration/stomp/client/StompServerIntegrationTests.java

Cherry-picked from cb0d43db6b
This commit is contained in:
Artem Bilan
2018-02-27 11:33:42 -05:00
parent 0f47fd25d1
commit e83e2cafca
6 changed files with 100 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,7 +20,7 @@
<stream-transformer input-channel="directInput" output-channel="output"/>
<stream-transformer input-channel="queueInput" output-channel="output">
<poller fixed-delay="10000"/>
<poller fixed-delay="1"/>
</stream-transformer>
<chain input-channel="charsetChannel" output-channel="output">

9
spring-integration-core/src/test/java/org/springframework/integration/config/xml/StreamTransformerParserTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2018 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.
@@ -39,6 +39,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
* @author Gary Russell
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@@ -64,7 +65,7 @@ public class StreamTransformerParserTests {
@Test
public void directChannelWithStringMessage() {
this.directInput.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
Message<?> result = output.receive(0);
Message<?> result = output.receive(10000);
assertNotNull(result);
assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload());
}
@@ -72,7 +73,7 @@ public class StreamTransformerParserTests {
@Test
public void queueChannelWithStringMessage() {
this.queueInput.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
Message<?> result = output.receive(3000);
Message<?> result = output.receive(10000);
assertNotNull(result);
assertArrayEquals("foo".getBytes(), (byte[]) result.getPayload());
}
@@ -80,7 +81,7 @@ public class StreamTransformerParserTests {
@Test
public void charset() {
this.charsetChannel.send(new GenericMessage<InputStream>(new ByteArrayInputStream("foo".getBytes())));
Message<?> result = output.receive(0);
Message<?> result = output.receive(10000);
assertNotNull(result);
assertEquals("foo", result.getPayload());
}

1
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelIntegrationTests-context.xml
View File

@@ -65,6 +65,7 @@
<bean id="transactionManager"
class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource"/>
<property name="defaultTimeout" value="10"/>
</bean>
<int:header-enricher input-channel="routingSlip" output-channel="input">

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2017 the original author or authors.
* Copyright 2002-2018 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.
@@ -31,6 +31,7 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.hamcrest.Matchers;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -39,6 +40,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.serializer.support.SerializationFailedException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.store.MessageGroup;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
@@ -83,17 +85,28 @@ public class JdbcMessageStoreChannelIntegrationTests {
@Autowired
private MessageChannel routingSlip;
@Autowired
@Qualifier("service-activator")
private AbstractEndpoint serviceActivator;
@Before
public void clear() {
Service.reset(1);
for (MessageGroup group : messageStore) {
messageStore.removeMessageGroup(group.getGroupId());
}
this.serviceActivator.start();
}
@After
public void tearDown() {
this.serviceActivator.stop();
}
@Test
public void testSendAndActivate() throws Exception {
input.send(new GenericMessage<String>("foo"));
input.send(new GenericMessage<>("foo"));
Service.await(10000);
assertEquals(1, Service.messages.size());
}
@@ -101,7 +114,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
@Test
public void testSendAndActivateWithRollback() throws Exception {
Service.fail = true;
input.send(new GenericMessage<String>("foo"));
input.send(new GenericMessage<>("foo"));
Service.await(10000);
assertThat(Service.messages.size(), Matchers.greaterThanOrEqualTo(1));
// After a rollback in the poller the message is still waiting to be delivered
@@ -128,28 +141,24 @@ public class JdbcMessageStoreChannelIntegrationTests {
@Repeat(2)
public void testTransactionalSendAndReceive() throws Exception {
boolean result = new TransactionTemplate(transactionManager).execute(new TransactionCallback<Boolean>() {
boolean result = new TransactionTemplate(transactionManager).execute(status -> {
@Override
public Boolean doInTransaction(TransactionStatus status) {
synchronized (storeLock) {
boolean result = input.send(new GenericMessage<String>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(3000);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
return result;
synchronized (storeLock) {
boolean result1 = input.send(new GenericMessage<>("foo"), 100L);
// This will time out because the transaction has not committed yet
try {
Service.await(100);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
return result1;
}
});
assertTrue("Could not send message", result);
@@ -184,7 +193,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
}
@Test
public void testSameTransactionSendAndReceive() throws Exception {
public void testSameTransactionSendAndReceive() {
final StopWatch stopWatch = new StopWatch();
DefaultTransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
@@ -239,7 +248,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
@Test
public void testWithRoutingSlip() {
try {
this.routingSlip.send(new GenericMessage<String>("foo"));
this.routingSlip.send(new GenericMessage<>("foo"));
fail("MessageDeliveryException expected");
}
catch (Exception e) {
@@ -255,7 +264,7 @@ public class JdbcMessageStoreChannelIntegrationTests {
private static boolean fail = false;
private static List<String> messages = new CopyOnWriteArrayList<String>();
private static List<String> messages = new CopyOnWriteArrayList<>();
private static CountDownLatch latch;

1
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests-context.xml
View File

@@ -59,6 +59,7 @@
<bean id="transactionManager"
class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
<property name="dataSource" ref="dataSource" />
<property name="defaultTimeout" value="10"/>
</bean>
</beans>

111
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreChannelOnePollerIntegrationTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2016 the original author or authors.
* Copyright 2002-2018 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.
@@ -26,6 +26,7 @@ import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -33,6 +34,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.store.MessageGroup;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
@@ -44,6 +46,11 @@ import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StopWatch;
/**
* @author Dave Syer
* @author Gary Russell
* @author Artem Bilan
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext // close at the end after class
@@ -65,54 +72,58 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
@Autowired
private PlatformTransactionManager transactionManager;
@Autowired
@Qualifier("service-relay")
private AbstractEndpoint serviceRelay;
@Before
public void clear() {
for (MessageGroup group : messageStore) {
messageStore.removeMessageGroup(group.getGroupId());
for (MessageGroup group : this.messageStore) {
this.messageStore.removeMessageGroup(group.getGroupId());
}
}
@Test
// @Repeat(50)
public void testSameTransactionDifferentChannelSendAndReceive() throws Exception {
@After
public void tearDown() {
this.serviceRelay.stop();
}
@Test
public void testSameTransactionDifferentChannelSendAndReceive() throws Exception {
Service.reset(1);
assertNull(durable.receive(100L));
assertNull(relay.receive(100L));
assertNull(this.durable.receive(100L));
assertNull(this.relay.receive(100L));
final StopWatch stopWatch = new StopWatch();
boolean result = new TransactionTemplate(transactionManager).execute(new TransactionCallback<Boolean>() {
boolean result =
new TransactionTemplate(this.transactionManager)
.execute(status -> {
@Override
public Boolean doInTransaction(TransactionStatus status) {
synchronized (this.storeLock) {
synchronized (storeLock) {
boolean result1 = this.relay.send(new GenericMessage<>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(100);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
boolean result = relay.send(new GenericMessage<String>("foo"), 500L);
// This will time out because the transaction has not committed yet
try {
Service.await(1000);
fail("Expected timeout");
}
catch (Exception e) {
// expected
}
try {
stopWatch.start();
// It hasn't arrive yet because we are still in the sending transaction
assertNull(this.durable.receive(100L));
}
finally {
stopWatch.stop();
}
try {
stopWatch.start();
// It hasn't arrive yet because we are still in the sending transaction
assertNull(durable.receive(100L));
}
finally {
stopWatch.stop();
}
return result1;
return result;
}
}
});
}
});
assertTrue("Could not send message", result);
// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout
@@ -130,25 +141,21 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
*
* With the storeLock: It doesn't deadlock as long as the lock is injected into the poller as well.
*/
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
new TransactionTemplate(this.transactionManager)
.execute(status -> {
synchronized (this.storeLock) {
@Override
public Void doInTransaction(TransactionStatus status) {
synchronized (storeLock) {
try {
stopWatch.start();
this.durable.receive(100L);
return null;
}
finally {
stopWatch.stop();
}
try {
stopWatch.start();
durable.receive(100L);
return null;
}
finally {
stopWatch.stop();
}
}
}
});
});
// If the poll blocks in the RDBMS there is no way for the queue to respect the timeout
assertTrue("Timed out waiting for receive", stopWatch.getTotalTimeMillis() < 10000);
@@ -156,6 +163,7 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
}
public static class Service {
private static boolean fail = false;
private static List<String> messages = new CopyOnWriteArrayList<String>();
@@ -182,6 +190,7 @@ public class JdbcMessageStoreChannelOnePollerIntegrationTests {
}
return input;
}
}
}