How to make @JmsListener transactional (with Spring Boot and Atomikos)?

Viewed 69

I have a Spring Boot application which consumes messages from queue (ActiveMQ) and writes them to the database (DB2) and I need it to be fully transactional. I got to a point where I understood that transaction manager (using spring-boot-starter-jta-atomikos) is a best solution for distributed transactions and I'm trying to correctly implement it.

JMS configuration class:

@EnableJms
@Configuration
public class MQConfig {

  @Bean
  public ConnectionFactory connectionFactory() {
    RedeliveryPolicy rp = new RedeliveryPolicy();
    rp.setMaximumRedeliveries(3);
    rp.setRedeliveryDelay(1000L);

    ActiveMQConnectionFactory cf = new ActiveMQConnectionFactory();
    cf.setBrokerURL("tcp://localhost:61616");
    cf.setRedeliveryPolicy(rp);
    return cf;
  }

  @Bean
  public JmsTemplate jmsTemplate() {
    JmsTemplate template = new JmsTemplate(connectionFactory());
    template.setConnectionFactory(connectionFactory());
    return template;
  }

  @Bean
  public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {
    DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory());
    factory.setCacheLevelName("CACHE_CONSUMER");
    factory.setReceiveTimeout(1000L);
    factory.setSessionTransacted(true);
    return factory;
  }
}

JMS listener class:

@Component
public class MQListener {

  @Autowired
  private ImportRecordsService importRecordsService;

  @JmsListener(
        containerFactory = "jmsListenerContainerFactory",
        destination = "test.queue"
        // concurrency = "4-10"
  )
  public void receiveMessage(TextMessage message) throws JMSException {
    importRecordsService.createRecord();
  }
}

Service class that writes to DB:

@Service
public class ImportRecordsService {

  @Autowired
  private ImportRecordsDAO dao;

  @Transactional
  public void createRecord() {
    ImportRecord record = new ImportRecord();
    record.setDateCreated(LocalDateTime.now());
    record.setName("test-001");
    dao.save(record);
  }
}

If exception is thrown inside createRecord() after save, rollback works as it should. When an exception is thrown inside JMS listener in receiveMessage() after save, message is returned to queue but database record stays.

Any help greatly appreciated.

0 Answers
Related