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

Change prefetch on jms tests to 0 and add @DirtiesContext to tests

Browse Source 
With the upgrade to ActiveMQ 5.9.0 as part of Spring IO, ActiveMQ's
prefetch behavior changed.  With each call to
JmsTemplate#receiveAndConvert, the template creates a new consumer.  In
these tests, the first call created a consumer which prefetched all the
test messages on the queue, leaving them unavailable for the second
consumer the JmsTemplate created.  By setting prefetch to 0, the
messages are now available for the subsequent
JmsTemplate#receiveAndConvert calls.  This addresses BATCH-2248.
This commit is contained in:
Michael Minella
2014-07-07 09:29:59 -05:00
committed by Michael Minella
parent 090e6f2039
commit a066647149
5 changed files with 83 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/container/jms/BatchMessageListenerContainerIntegrationTests.java
View File

@@ -15,11 +15,26 @@
*/
package org.springframework.batch.container.jms;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.retry.RecoveryCallback;
@@ -28,28 +43,17 @@ import org.springframework.retry.RetryContext;
import org.springframework.retry.policy.NeverRetryPolicy;
import org.springframework.retry.support.DefaultRetryState;
import org.springframework.retry.support.RetryTemplate;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.TextMessage;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
/**
* @author Dave Syer
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
@DirtiesContext
public class BatchMessageListenerContainerIntegrationTests {
@Autowired

27
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/repeat/jms/AsynchronousTests.java
View File

@@ -16,7 +16,9 @@
package org.springframework.batch.repeat.jms;
import static org.junit.Assert.*;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
@@ -27,22 +29,25 @@ import javax.jms.Session;
import javax.jms.TextMessage;
import javax.sql.DataSource;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.container.jms.BatchMessageListenerContainer;
import org.springframework.batch.jms.ExternalRetryInBatchTests;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.listener.SessionAwareMessageListener;
import org.springframework.util.ClassUtils;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.beans.factory.annotation.Autowired;
import org.junit.runner.RunWith;
import org.junit.Before;
import org.junit.After;
import org.junit.Test;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ClassUtils;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
@DirtiesContext
public class AsynchronousTests {
protected String[] getConfigLocations() {
@@ -117,7 +122,7 @@ public class AsynchronousTests {
});
container.initializeProxy();
container.start();
// Need to sleep for at least a second here...
@@ -139,7 +144,7 @@ public class AsynchronousTests {
public void testRollback() throws Exception {
assertInitialState();
// Prevent us from being overwhelmed after rollback
container.setRecoveryInterval(500);
@@ -155,7 +160,7 @@ public class AsynchronousTests {
}
}
});
container.initializeProxy();
container.start();

71
spring-batch-infrastructure-tests/src/test/java/org/springframework/batch/repeat/jms/SynchronousTests.java
View File

@@ -16,8 +16,20 @@
package org.springframework.batch.repeat.jms;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.sql.DataSource;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.batch.repeat.RepeatCallback;
import org.springframework.batch.repeat.RepeatContext;
import org.springframework.batch.repeat.RepeatStatus;
@@ -27,7 +39,10 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jms.connection.SessionProxy;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.SessionCallback;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.test.context.transaction.BeforeTransaction;
@@ -36,16 +51,9 @@ import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionCallback;
import org.springframework.transaction.support.TransactionTemplate;
import javax.jms.ConnectionFactory;
import javax.sql.DataSource;
import java.util.ArrayList;
import java.util.List;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "/org/springframework/batch/jms/jms-context.xml")
@DirtiesContext
public class SynchronousTests implements ApplicationContextAware {
@Autowired
@@ -101,18 +109,18 @@ public class SynchronousTests implements ApplicationContextAware {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) jmsTemplate.receiveAndConvert("queue");
System.out.println("text = " + text);
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
return RepeatStatus.continueIf(text != null);
}
});
System.err.println(jdbcTemplate.queryForList("select * from T_BARS"));
int count = jdbcTemplate.queryForObject("select count(*) from T_BARS", Integer.class);
assertEquals(2, count);
assertTrue(list.contains("foo"));
assertTrue(list.contains("bar"));
String text = (String) jmsTemplate.receiveAndConvert("queue");
assertEquals(null, text);
@@ -121,6 +129,8 @@ public class SynchronousTests implements ApplicationContextAware {
@Test
public void testFullRollback() throws Exception {
onSetUpBeforeTransaction();
assertInitialState();
new TransactionTemplate(transactionManager).execute(new TransactionCallback<Void>() {
@@ -178,7 +188,6 @@ public class SynchronousTests implements ApplicationContextAware {
@Override
public RepeatStatus doInIteration(RepeatContext context) throws Exception {
String text = (String) txJmsTemplate.receiveAndConvert("queue");
System.out.println("Receiving in transaction: " + text);
list.add(text);
jdbcTemplate.update("INSERT into T_BARS (id,name,foo_date) values (?,?,null)", list.size(), text);
return RepeatStatus.continueIf(text != null);
@@ -187,36 +196,32 @@ public class SynchronousTests implements ApplicationContextAware {
// Simulate a message system failure before the main transaction
// commits...
// txJmsTemplate.execute(new SessionCallback<Void>() {
// @Override
// public Void doInJms(Session session) throws JMSException {
// try {
// System.out.println("Session = " + session + " pass test? " + (session instanceof SessionProxy));
// assertTrue("Not a SessionProxy - wrong spring version?", session instanceof SessionProxy);
// ((SessionProxy) session).getTargetSession().rollback();
// }
// catch (JMSException e) {
// throw e;
// }
// catch (Exception e) {
// // swallow it
// e.printStackTrace();
// }
// return null;
// }
// });
txJmsTemplate.execute(new SessionCallback<Void>() {
@Override
public Void doInJms(Session session) throws JMSException {
try {
assertTrue("Not a SessionProxy - wrong spring version?", session instanceof SessionProxy);
((SessionProxy) session).getTargetSession().rollback();
}
catch (JMSException e) {
throw e;
}
catch (Exception e) {
// swallow it
e.printStackTrace();
}
return null;
}
});
return null;
}
});
System.err.println(jdbcTemplate.queryForList("select * from T_BARS"));
String text = "";
List<String> msgs = new ArrayList<String>();
while (text != null) {
text = (String) txJmsTemplate.receiveAndConvert("queue");
System.out.println("text = " + text);
msgs.add(text);
}

2
spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/jms/init.sql
View File

@@ -1,3 +1,5 @@
DROP TABLE T_BARS;
create table T_BARS (
id integer not null primary key,
name varchar(80),

19
spring-batch-infrastructure-tests/src/test/resources/org/springframework/batch/jms/jms-context.xml
View File

@@ -17,10 +17,13 @@
<property name="dataSource" ref="dataSource" />
</bean>
<bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" depends-on="brokerService">
<property name="brokerURL">
<value>vm://localhost</value>
</property>
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<property name="targetConnectionFactory" ref="targetConnectionFactory"/>
<property name="sessionCacheSize" value="5"/>
</bean>
<bean id="targetConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory" depends-on="brokerService">
<property name="brokerURL" value="vm://localhost?jms.prefetchPolicy.all=0"/>
</bean>
<bean id="txAwareConnectionFactory"
@@ -70,11 +73,9 @@
<bean id="transactionalBatchTemplate"
class="org.springframework.transaction.interceptor.TransactionProxyFactoryBean">
<property name="transactionManager" ref="transactionManager" />
<property name="proxyInterfaces">
<value>org.springframework.batch.repeat.RepeatOperations</value>
</property>
<property name="proxyInterfaces" value="org.springframework.batch.repeat.RepeatOperations"/>
<property name="proxyTargetClass" value="false" />
<property name="transactionAttributes" value="*=PROPAGATION_REQUIRED"></property>
<property name="transactionAttributes" value="*=PROPAGATION_REQUIRED"/>
<property name="target">
<bean class="org.springframework.batch.repeat.support.RepeatTemplate">
<property name="completionPolicy">
@@ -92,7 +93,7 @@
<property name="useJmx" value="false"/>
<property name="transportConnectorURIs">
<list>
<value>vm://localhost</value>
<value>vm://localhost?jms.prefetchPolicy.all=0</value>
</list>
</property>
<property name="persistenceAdapter">