spring rabbit and spring transactions

Question

I have a spring rabbit consumer like:

@Override public void onMessage(Message amqpMessage, Channel channel)
            throws Exception {
//..some code goes here - I want it to be in spring transaction
}

The issue is the code which is in onMessage method is not under transaction. I checked it, I save data to 3 tables, then throw exception, then save to 4th table. And data from 3 previos tables is not being rolled back. How to do that properly in spring? I want all code in onMessage method to be within a transaction. Thanks

UPDATE My rabbit conf:

@Configuration @ComponentScan(basePackages = {"com.mycompany"})
public class TicketModeRabbit {
    @Bean TicketModeConsumer ticketModeConsumer() {
        return new TicketModeConsumer();
    }

    @Bean(name = TicketModeRabbitData.QUEUE_BEAN_NAME) Queue queue() {
        return new Queue(TicketModeRabbitData.QUEUE_BEAN_NAME);
    }


    @Bean(name = TicketModeRabbitData.QUEUE_BINDING_NAME) Binding binding(
            @Qualifier(TicketModeRabbitData.QUEUE_BEAN_NAME) Queue q, TopicExchange e) {
        return BindingBuilder.bind(q).to(e).with(TicketModeRabbitData.QUEUE_TOKEN_NAME);
    }

    @Bean(name = TicketModeRabbitData.CONTAINER_NAME)
    SimpleMessageListenerContainer container(ConnectionFactory connectionFactory,
            @Qualifier(TicketModeRabbitData.LISTENER_ADAPTED_NAME)
                    MessageListenerAdapter listenerAdapter) {
        return WorkerConfigHelper
                .rabbitConfigListenerContainer(connectionFactory, listenerAdapter,
                        TicketModeRabbitData.QUEUE_BEAN_NAME,
                        WorkerConfigHelper.GLOBAL_CONCURRENT_CONSUMERS);
    }

    @Bean(name = TicketModeRabbitData.LISTENER_ADAPTED_NAME)
    MessageListenerAdapter listenerAdapter() {
        return new MessageListenerAdapter(ticketModeConsumer());
    }
}

Answer 1

If your transaction manager is properly set up for your database, the only thing you need to do is add the @Transactional annotation on the onMessage method. Note that the consumer (MessageListener) needs to be a bean managed by the Spring container.

@Override 
@Transactional
public void onMessage(Message amqpMessage, Channel channel)
            throws Exception {
    //..some code goes here - I want it to be in spring transaction
}