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

Fix logic in the PollingTransactionTests

Browse Source 
The `transactionWithCommitAndAdvices()` uses `SimpleRepeatAdvice`
which essentially performs `doPoll()` twice.
In this case we don't have enough messages in the `goodInputWithAdvice`
queue

* Add one more message to the `goodInputWithAdvice` queue during testing
* Increase all the timeouts
* Increase capacity in queues for the `transactionWithCommitAndAdvices()`

**Cherry-pick to 5.0.x and 4.3.x**
This commit is contained in:
Artem Bilan
2018-04-26 09:22:00 -04:00
parent 0ef659b8a7
commit 000f2972f4
2 changed files with 72 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/PollingTransactionTests.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"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@@ -34,6 +34,7 @@ import org.junit.Test;
import org.springframework.aop.Advisor; import org.springframework.aop.Advisor;
import org.springframework.aop.ProxyMethodInvocation; import org.springframework.aop.ProxyMethodInvocation;
import org.springframework.aop.framework.Advised; import org.springframework.aop.framework.Advised;
import org.springframework.context.Lifecycle;
import org.springframework.context.support.ClassPathXmlApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.endpoint.PollingConsumer; import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.test.util.TestUtils;
@@ -68,8 +69,9 @@ public class PollingTransactionTests {
PollableChannel output = (PollableChannel) context.getBean("output"); PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
assertEquals(0, txManager.getRollbackCount()); assertEquals(0, txManager.getRollbackCount());
input.send(new GenericMessage<String>("test")); context.getBean("goodService", Lifecycle.class).start();
txManager.waitForCompletion(1000); input.send(new GenericMessage<>("test"));
txManager.waitForCompletion(10000);
Message<?> message = output.receive(0); Message<?> message = output.receive(0);
assertNotNull(message); assertNotNull(message);
assertEquals(1, txManager.getCommitCount()); assertEquals(1, txManager.getCommitCount());
@@ -82,11 +84,12 @@ public class PollingTransactionTests {
public void transactionWithCommitAndAdvices() throws InterruptedException { public void transactionWithCommitAndAdvices() throws InterruptedException {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"transactionTests.xml", this.getClass()); "transactionTests.xml", this.getClass());
PollingConsumer advicedPoller = context.getBean("advicedSa", PollingConsumer.class); PollingConsumer advisedPoller = context.getBean("advisedSa", PollingConsumer.class);
List<Advice> adviceChain = TestUtils.getPropertyValue(advicedPoller, "adviceChain", List.class); List<Advice> adviceChain = TestUtils.getPropertyValue(advisedPoller, "adviceChain", List.class);
assertEquals(4, adviceChain.size()); assertEquals(4, adviceChain.size());
Runnable poller = TestUtils.getPropertyValue(advicedPoller, "poller", Runnable.class); advisedPoller.start();
Runnable poller = TestUtils.getPropertyValue(advisedPoller, "poller", Runnable.class);
Callable<?> pollingTask = TestUtils.getPropertyValue(poller, "pollingTask", Callable.class); Callable<?> pollingTask = TestUtils.getPropertyValue(poller, "pollingTask", Callable.class);
assertTrue("Poller is not Advised", pollingTask instanceof Advised); assertTrue("Poller is not Advised", pollingTask instanceof Advised);
Advisor[] advisors = ((Advised) pollingTask).getAdvisors(); Advisor[] advisors = ((Advised) pollingTask).getAdvisors();
@@ -98,10 +101,14 @@ public class PollingTransactionTests {
PollableChannel output = (PollableChannel) context.getBean("output"); PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
assertEquals(0, txManager.getRollbackCount()); assertEquals(0, txManager.getRollbackCount());
input.send(new GenericMessage<>("test")); input.send(new GenericMessage<>("test"));
input.send(new GenericMessage<>("test2"));
txManager.waitForCompletion(10000); txManager.waitForCompletion(10000);
Message<?> message = output.receive(0); Message<?> message = output.receive(0);
assertNotNull(message); assertNotNull(message);
message = output.receive(0);
assertNotNull(message);
assertEquals(0, txManager.getRollbackCount()); assertEquals(0, txManager.getRollbackCount());
context.close(); context.close();
} }
@@ -115,8 +122,11 @@ public class PollingTransactionTests {
PollableChannel output = (PollableChannel) context.getBean("output"); PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
assertEquals(0, txManager.getRollbackCount()); assertEquals(0, txManager.getRollbackCount());
input.send(new GenericMessage<String>("test"));
txManager.waitForCompletion(1000); context.getBean("badService", Lifecycle.class).start();
input.send(new GenericMessage<>("test"));
txManager.waitForCompletion(10000);
Message<?> message = output.receive(0); Message<?> message = output.receive(0);
assertNull(message); assertNull(message);
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
@@ -132,10 +142,10 @@ public class PollingTransactionTests {
PollableChannel input = (PollableChannel) context.getBean("input"); PollableChannel input = (PollableChannel) context.getBean("input");
PollableChannel output = (PollableChannel) context.getBean("output"); PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test")); input.send(new GenericMessage<>("test"));
Message<?> reply = output.receive(3000); Message<?> reply = output.receive(10000);
assertNotNull(reply); assertNotNull(reply);
txManager.waitForCompletion(3000); txManager.waitForCompletion(10000);
assertEquals(1, txManager.getCommitCount()); assertEquals(1, txManager.getCommitCount());
assertEquals(Propagation.REQUIRED.value(), txManager.getLastDefinition().getPropagationBehavior()); assertEquals(Propagation.REQUIRED.value(), txManager.getLastDefinition().getPropagationBehavior());
context.close(); context.close();
@@ -149,25 +159,25 @@ public class PollingTransactionTests {
PollableChannel input = (PollableChannel) context.getBean("input"); PollableChannel input = (PollableChannel) context.getBean("input");
PollableChannel output = (PollableChannel) context.getBean("output"); PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test")); input.send(new GenericMessage<>("test"));
Message<?> reply = output.receive(3000); Message<?> reply = output.receive(10000);
assertNotNull(reply); assertNotNull(reply);
txManager.waitForCompletion(3000); txManager.waitForCompletion(10000);
assertEquals(1, txManager.getCommitCount()); assertEquals(1, txManager.getCommitCount());
assertEquals(Propagation.REQUIRES_NEW.value(), txManager.getLastDefinition().getPropagationBehavior()); assertEquals(Propagation.REQUIRES_NEW.value(), txManager.getLastDefinition().getPropagationBehavior());
context.close(); context.close();
} }
@Test @Test
public void propagationSupports() throws InterruptedException { public void propagationSupports() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"propagationSupportsTests.xml", this.getClass()); "propagationSupportsTests.xml", this.getClass());
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
PollableChannel input = (PollableChannel) context.getBean("input"); PollableChannel input = (PollableChannel) context.getBean("input");
PollableChannel output = (PollableChannel) context.getBean("output"); PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test")); input.send(new GenericMessage<>("test"));
Message<?> reply = output.receive(3000); Message<?> reply = output.receive(10000);
assertNotNull(reply); assertNotNull(reply);
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
assertNull(txManager.getLastDefinition()); assertNull(txManager.getLastDefinition());
@@ -175,15 +185,15 @@ public class PollingTransactionTests {
} }
@Test @Test
public void propagationNotSupported() throws InterruptedException { public void propagationNotSupported() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"propagationNotSupportedTests.xml", this.getClass()); "propagationNotSupportedTests.xml", this.getClass());
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
PollableChannel input = (PollableChannel) context.getBean("input"); PollableChannel input = (PollableChannel) context.getBean("input");
PollableChannel output = (PollableChannel) context.getBean("output"); PollableChannel output = (PollableChannel) context.getBean("output");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test")); input.send(new GenericMessage<>("test"));
Message<?> reply = output.receive(3000); Message<?> reply = output.receive(10000);
assertNotNull(reply); assertNotNull(reply);
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
assertNull(txManager.getLastDefinition()); assertNull(txManager.getLastDefinition());
@@ -191,7 +201,7 @@ public class PollingTransactionTests {
} }
@Test @Test
public void propagationMandatory() throws Throwable { public void propagationMandatory() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"propagationMandatoryTests.xml", this.getClass()); "propagationMandatoryTests.xml", this.getClass());
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager"); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManager");
@@ -199,8 +209,8 @@ public class PollingTransactionTests {
PollableChannel output = (PollableChannel) context.getBean("output"); PollableChannel output = (PollableChannel) context.getBean("output");
PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel"); PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
input.send(new GenericMessage<String>("test")); input.send(new GenericMessage<>("test"));
Message<?> errorMessage = errorChannel.receive(3000); Message<?> errorMessage = errorChannel.receive(10000);
assertNotNull(errorMessage); assertNotNull(errorMessage);
Object payload = errorMessage.getPayload(); Object payload = errorMessage.getPayload();
assertEquals(MessagingException.class, payload.getClass()); assertEquals(MessagingException.class, payload.getClass());
@@ -212,7 +222,7 @@ public class PollingTransactionTests {
} }
@Test @Test
public void commitFailureAndHandlerFailureTest() throws Throwable { public void commitFailureAndHandlerFailureTest() {
ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext( ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(
"transactionFailureTests.xml", this.getClass()); "transactionFailureTests.xml", this.getClass());
TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManagerBad"); TestTransactionManager txManager = (TestTransactionManager) context.getBean("txManagerBad");
@@ -222,7 +232,7 @@ public class PollingTransactionTests {
PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel"); PollableChannel errorChannel = (PollableChannel) context.getBean("errorChannel");
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
inputTxFail.send(new GenericMessage<>("commitFailureTest")); inputTxFail.send(new GenericMessage<>("commitFailureTest"));
Message<?> errorMessage = errorChannel.receive(10000); Message<?> errorMessage = errorChannel.receive(20000);
assertNotNull(errorMessage); assertNotNull(errorMessage);
Object payload = errorMessage.getPayload(); Object payload = errorMessage.getPayload();
assertEquals(MessagingException.class, payload.getClass()); assertEquals(MessagingException.class, payload.getClass());
@@ -232,7 +242,7 @@ public class PollingTransactionTests {
assertNotNull(output.receive(0)); assertNotNull(output.receive(0));
assertEquals(0, txManager.getCommitCount()); assertEquals(0, txManager.getCommitCount());
inputHandlerFail.send(new GenericMessage<>("handlerFalilureTest")); inputHandlerFail.send(new GenericMessage<>("handlerFailureTest"));
errorMessage = errorChannel.receive(10000); errorMessage = errorChannel.receive(10000);
assertNotNull(errorMessage); assertNotNull(errorMessage);
payload = errorMessage.getPayload(); payload = errorMessage.getPayload();

62
spring-integration-core/src/test/java/org/springframework/integration/dispatcher/transactionTests.xml
View File

@@ -1,53 +1,64 @@
<?xml version="1.0" encoding="UTF-8"?> <?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration" <beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd"> http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="badInput"> <channel id="badInput">
<queue capacity="1" /> <queue capacity="1"/>
</channel> </channel>
<channel id="goodInput"> <channel id="goodInput">
<queue capacity="1" /> <queue capacity="1"/>
</channel> </channel>
<channel id="goodInputWithAdvice"> <channel id="goodInputWithAdvice">
<queue capacity="1" /> <queue capacity="2"/>
</channel> </channel>
<channel id="output"> <channel id="output">
<queue capacity="1" /> <queue capacity="2"/>
</channel> </channel>
<service-activator input-channel="badInput" ref="testBean" <service-activator id="badService"
method="bad" output-channel="output"> auto-startup="false"
input-channel="badInput"
ref="testBean" method="bad"
output-channel="output">
<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1"> <poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
<transactional transaction-manager="txManager" /> <transactional transaction-manager="txManager"/>
</poller> </poller>
</service-activator> </service-activator>
<service-activator input-channel="goodInput" ref="testBean" <service-activator id="goodService"
method="good" output-channel="output"> auto-startup="false"
input-channel="goodInput"
ref="testBean" method="good"
output-channel="output">
<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1"> <poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
<transactional transaction-manager="txManager" /> <transactional transaction-manager="txManager"/>
</poller> </poller>
</service-activator> </service-activator>
<service-activator id="advicedSa" input-channel="goodInputWithAdvice" ref="testBean" <service-activator id="advisedSa"
method="good" output-channel="output"> auto-startup="false"
input-channel="goodInputWithAdvice"
ref="testBean" method="good"
output-channel="output">
<poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1"> <poller max-messages-per-poll="1" fixed-delay="1" receive-timeout="-1">
<advice-chain> <advice-chain>
<ref bean="txAdvise"/> <ref bean="txAdvise"/>
<ref bean="adviceA" /> <ref bean="adviceA"/>
<beans:bean class="org.springframework.integration.dispatcher.PollingTransactionTests.SampleAdvice"/> <beans:bean class="org.springframework.integration.dispatcher.PollingTransactionTests$SampleAdvice"/>
<beans:bean class="org.springframework.integration.dispatcher.PollingTransactionTests.SimpleRepeatAdvice"/> <beans:bean
class="org.springframework.integration.dispatcher.PollingTransactionTests$SimpleRepeatAdvice"/>
</advice-chain> </advice-chain>
</poller> </poller>
</service-activator> </service-activator>
<beans:bean id="txAdvise" class="org.springframework.transaction.interceptor.TransactionInterceptor"> <beans:bean id="txAdvise" class="org.springframework.transaction.interceptor.TransactionInterceptor">
<beans:property name="transactionManager" ref="txManager"/> <beans:property name="transactionManager" ref="txManager"/>
<beans:property name="transactionAttributeSource"> <beans:property name="transactionAttributeSource">
@@ -58,12 +69,11 @@
</beans:bean> </beans:bean>
</beans:property> </beans:property>
</beans:bean> </beans:bean>
<beans:bean id="adviceA" class="org.springframework.integration.dispatcher.PollingTransactionTests.SampleAdvice"/>
<beans:bean id="testBean" <beans:bean id="adviceA" class="org.springframework.integration.dispatcher.PollingTransactionTests$SampleAdvice"/>
class="org.springframework.integration.dispatcher.TestBean" />
<beans:bean id="testBean" class="org.springframework.integration.dispatcher.TestBean"/>
<beans:bean id="txManager" class="org.springframework.integration.util.TestTransactionManager"/>
<beans:bean id="txManager"
class="org.springframework.integration.util.TestTransactionManager" />
</beans:beans> </beans:beans>