Spring Integration, ActiveMQ Dead Letter Queue and Testing

Spring Integration, ActiveMQ Dead Letter Queue and Testing

recently i was working on a design whereby the application would consume messages from a JMS queue, but may encounter a transient exception (e.g. the database connection may not be available temporarily).  in Spring Integration i could use a retry pattern on the endpoint, but as the application is in a cluster, another JVM may have better luck getting to the database.  so, to use the cluster design, if a transient exception should occur, the exception would be thrown up, the transaction would roll back and the message would go back on the queue, awaiting consumption again.  this would allow another node in the cluster an opportunity to get the message and process it.

ok, so this is all very nice, but how do we confirm that this actually works?  we're using ActiveMQ in the test environment (embedded) and wanted to see the message actually got rolled back.  the default behavior of ActiveMQ, (since version 5.4?) is the following;

• no delay in retries
• 6 attempts
• if all attempt fail, put it on the Dead Letter Queue

so, here's a couple of snippets that use this to test.

first the integration setup;

***note: a DefaultMessageListenerContainer is manually configured to support the transactions, setting the transactionManager directly on the channel adapter alone doesn't work (in 2.2.4.RELEASE)

now our test resources;

our testing bean;

package de.incompleteco.spring.integration.service;

public class FailureService {

 public void process(Object payload) {
  throw new RuntimeException("simulated failure");
 }
 
}

and our junit test;

package de.incompleteco.spring.integration;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.Queue;
import javax.jms.TextMessage;

import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration({"classpath*:/META-INF/spring/*-context.xml"})
@ActiveProfiles("junit")
public class DeadLetterQueueIntegrationTest {

 private static final String PAYLOAD = "hello world";
 
 @Autowired
 private ConnectionFactory connectionFactory;
 
 @Autowired
 @Qualifier("request.queue")
 private Queue requestQueue;
 
 @Autowired
 @Qualifier("dead.letter.queue")
 private Queue dlQueue;
 
 private JmsTemplate jmsTemplate;
 
 @Before
 public void before() throws Exception {
  jmsTemplate = new JmsTemplate(connectionFactory);
  jmsTemplate.setDefaultDestination(dlQueue);
  jmsTemplate.setReceiveTimeout(2 * 1000);//about 2 seconds
 }
 
 @Test
 public void test() throws Exception {
  //send a message
  jmsTemplate.convertAndSend(requestQueue, PAYLOAD);
  //expect it to fail and, by default, be put back on the dead letter queue
  Message failedMessage = jmsTemplate.receive();
  //now check that it has the same content
  assertTrue(failedMessage instanceof TextMessage);
  assertEquals(PAYLOAD,((TextMessage) failedMessage).getText());
 }
 
}

Spring Batch - Parallel Item Writer

Spring Batch - Parallel Item Writer

here's a quick solution i put together for stackoverflow and writing to many itemwriters in parallel

here's the class

package de.incompleteco.spring.batch.item.support;

import java.util.List;

import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.CompositeItemWriter;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;

import de.incompleteco.spring.domain.SimpleEntity;

public class ParallelCompositeItemWriter extends CompositeItemWriter< SimpleEntity > {

 private List< ItemWriter< ? super SimpleEntity > > delegates;
 
 private TaskExecutor taskExecutor;
 
 @Override
 public void write(final List< ? extends SimpleEntity > item) throws Exception {
  for (final ItemWriter< ? super SimpleEntity > writer : delegates) {
   taskExecutor.execute(new Runnable()  {
    @Override
    public void run() {
     try {
      writer.write(item);
     } catch (Throwable t) {
      rethrow(t);
     } 
    }
    
    private void rethrow(Throwable t) {
     if (t instanceof RuntimeException) {
      throw (RuntimeException) t;
     }
     else if (t instanceof Error) {
      throw (Error) t;
     }
     throw new IllegalStateException(t);
    }  
   });
  }//end for
 }


 public void setTaskExecutor(TaskExecutor taskExecutor) {
  this.taskExecutor = taskExecutor;
 }
 
 @Override
 public void setDelegates(List < ItemWriter < ? super SimpleEntity > > delegates) {
  this.delegates = delegates;
  super.setDelegates(delegates);
 }

 @Override
 public void afterPropertiesSet() throws Exception {
  super.afterPropertiesSet();
  Assert.notNull(taskExecutor,"Task executor needs to be set");
 }
 
}

and here's a sample configuration to run it

 
  
   
    
   
  
 

 
  
 
 
 
  
  
 
 
 
  
   
   
   
    
   
  
  
   
   
   
    
   
    
 
 
 
  
   
  
  
   
    
  
   
    
 
 
 
 

 
     
     
     
     
     
         
             jdbc:h2:mem:a;DB_CLOSE_DELAY=-1
         
     
  
 
 
     
     
     
     
     
         
             jdbc:h2:mem:b;DB_CLOSE_DELAY=-1
         
     
   
 
 
  
 
 
 
  
 
 

 
 
 

 
     
     
 

there's a couple of things to note;
- Bitronix JTA is used to support XA transactions across multiple databases
- the jdbc stuff is pretty rubbish and needs to be done better

Spring LocalContainerEntityManagerFactoryBean, packagesToScan and JTA

Spring LocalContainerEntityManagerFactoryBean, packagesToScan and JTA

here's an interesting one, Hibernate 4, Spring JPA (using LocalContainerEntityManagerFactoryBean) and JTA.  the scenario is, testing in JUnit and need to specify (and emulate) a JTA transaction. unfortunately, it's not all straight out-of-the-box, but here it is;

first, LocalContainerEntityManagerFactoryBean treats everything as a non-JTA transaction, so there's a little extra configuration to be done.  To switch it out, Spring does provide a postProcessor for the PersistenceUnit and a mutable persistent unit to it, but we have to write our own switch.  (it's pretty straight forward, but maybe we can ask the nice chaps to include it out of the box in 3.2.x?)

here's my cut - you can roll/cut & paste your own

package org.incompletecode.spring.batch.jpa;
 
import org.springframework.orm.jpa.persistenceunit.MutablePersistenceUnitInfo;
import org.springframework.orm.jpa.persistenceunit.PersistenceUnitPostProcessor;
 
public class JTAPersistenceUnitPostProcessor implements
  PersistenceUnitPostProcessor {
 
 @Override
 public void postProcessPersistenceUnitInfo(MutablePersistenceUnitInfo pui) {
  //switch the nonJTA to JTA
  pui.setJtaDataSource(pui.getNonJtaDataSource());
  pui.setNonJtaDataSource(null);
 }
 
}

and now here's a quick example of using Bitronix as your JTA provider

 
 
 
 
     
 
  
 
 
 
     
 
 
 
     
     
 
 
 
 
 
      
      
      
      
      
      
          
              jdbc:h2:mem:a;DB_CLOSE_ON_EXIT=FALSE;DB_CLOSE_DELAY=-1