Spring Integration: Persistent and transactional QueueChannel

user_8932531 0
Viewed 2556

In Spring Integration we have a Setup that looks something like this:

                                                     ---> 
                                                     ---> 
(dispatcher) Messages --> Gateway ----> QueueChannel ---> MessageHandler (worker)
                                                     ---> 
                                                     --->

So we have one Dispatcher Thread that takes Messages from a MQTT-Broker and forwards them into the Queue. The Poller for the Queue is provided with a TaskExecuter, so the Consumer is multithreaded. We managed to implement all the functionalities. So the just described setup is already implemented.

Now to guarantee no data loss we want to make two things:

1.: We want our queue to persist the data, so when the Programm shuts down ungracefully, all the data in the queue will still be there. This also worked for us, we are using MongoDB as a database because we read somewhere in your docs that this is the recommended way to do it.

2.: The second thing we want to assure is that the worker threads are working transactional. So only if the worker threads return correctly the messages will permanently be deleted from the queue (and therefore the persistent MessageStore). If the program shuts down during the processing of a message (by the worker thread) the message will still be in the queue at the next startup. Also if the worker, for example, thows an exception during the processing of the message, it will be put back into the queue.

Our implementation:

As explained before, the basic setup of the program is already implemented. We then extended the basic implementation with a message store implementation for the queue.

QueueChannel:

@Bean
public PollableChannel inputChannel(BasicMessageGroupStore mongoDbChannelMessageStore) {
    return new QueueChannel(new MessageGroupQueue(mongoDbChannelMessageStore, "inputChannel"));
}

backed by a Messagestore:

@Bean
public BasicMessageGroupStore mongoDbChannelMessageStore(MongoDbFactory mongoDbFactory) {
    MongoDbChannelMessageStore store = new MongoDbChannelMessageStore(mongoDbFactory);
    store.setPriorityEnabled(true);
    return store;
}

the matching Poller:

@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
    PollerMetadata poll = Pollers.fixedDelay(10).get(); 
    poll.setTaskExecutor(consumer);
    return poll;
}

Executor:

private Executor consumer = Executors.newFixedThreadPool(5);

What we have tried? As explained now we want to extend this implementation with a transactional functionality. We tried using the setTransactionSynchronizationFactory like explained here but it wasn't working (didn't get errors or anything but the behavior was still as it was before we added the TransactionSynchronizer):

@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() {
    PollerMetadata poll = Pollers.fixedDelay(10).get(); 
    poll.setTaskExecutor(consumer);

    BeanFactory factory = mock(BeanFactory.class);
    ExpressionEvaluatingTransactionSynchronizationProcessor etsp = new ExpressionEvaluatingTransactionSynchronizationProcessor();
    etsp.setBeanFactory(factory);
    etsp.setAfterRollbackChannel(inputChannel());
    etsp.setAfterRollbackExpression(new SpelExpressionParser().parseExpression("#bix"));
    etsp.setAfterCommitChannel(inputChannel());
    etsp.setAfterCommitExpression(new SpelExpressionParser().parseExpression("#bix"));
    DefaultTransactionSynchronizationFactory dtsf = new DefaultTransactionSynchronizationFactory(etsp);

    poll.setTransactionSynchronizationFactory(dtsf);

    return poll;
}

What would be the best way to realize our requirements in spring integration?

EDIT: As recommended in the answer I chose to do this with the JdbcChannelMessageStore. So I tried converting the XML Implementation described here (18.4.2) into Java. I wasn't quite sure on how to do it, this is what I have tried so far:

I created H2 database and run the script shown here on it.

Created JDBCChannelMessageStore Bean:

@Bean
public JdbcChannelMessageStore store() {
    JdbcChannelMessageStore ms =  new JdbcChannelMessageStore();
    ms.setChannelMessageStoreQueryProvider(queryProvider());
    ms.setUsingIdCache(true);
    ms.setDataSource(dataSource);
    return ms;
}

Created H2ChannelMessageStoreQueryProvider

    @Bean
    public ChannelMessageStoreQueryProvider queryProvider() {
        return new H2ChannelMessageStoreQueryProvider();
    }

Adapted the poller:

@Bean(name = PollerMetadata.DEFAULT_POLLER)
public PollerMetadata poller() throws Exception {
    PollerMetadata poll = Pollers.fixedDelay(10).get();
    poll.setTaskExecutor(consumer);
    poll.setAdviceChain(Collections.singletonList(transactionInterceptor()));

    return poll;
}

Autowired my PlaatformTransactionManager:

@Autowired
PlatformTransactionManager transactionManager;

And created TransactionInterceptor from the TransactonManager:

@Bean
public TransactionInterceptor transactionInterceptor() {
    return new TransactionInterceptorBuilder(true)
                .transactionManager(transactionManager)
                .isolation(Isolation.READ_COMMITTED)
                .propagation(Propagation.REQUIRED)
                .build();
}
spring-integration spring java
2 Answers

If you need to have queue as transactional, you definitely should take a look into the transactional MessageStore. And only JDBC one is like that. Just because only JDBC support transactions. So, when we perform DELETE, it is OK only if TX is committed.

The MongoDB, nor any other NoSQL DataBases, support such a model, therefore you only can push back the failed messages to the DB on rollback using TransactionSynchronizationFactory.

UPDATE

@RunWith(SpringRunner.class)
@DirtiesContext
public class So47264688Tests {

    private static final String MESSAGE_GROUP = "transactionalQueueChannel";

    private static EmbeddedDatabase dataSource;

    @BeforeClass
    public static void init() {
        dataSource = new EmbeddedDatabaseBuilder()
                .setType(EmbeddedDatabaseType.H2)
                .addScript("classpath:/org/springframework/integration/jdbc/schema-drop-h2.sql")
                .addScript("classpath:/org/springframework/integration/jdbc/schema-h2.sql")
                .build();
    }

    @AfterClass
    public static void destroy() {
        dataSource.shutdown();
    }

    @Autowired
    private PollableChannel transactionalQueueChannel;

    @Autowired
    private JdbcChannelMessageStore jdbcChannelMessageStore;

    @Autowired
    private PollingConsumer serviceActivatorEndpoint;

    @Autowired
    private CountDownLatch exceptionLatch;

    @Test
    public void testTransactionalQueueChannel() throws InterruptedException {
        GenericMessage<String> message = new GenericMessage<>("foo");
        this.transactionalQueueChannel.send(message);

        assertTrue(this.exceptionLatch.await(10, TimeUnit.SECONDS));
        this.serviceActivatorEndpoint.stop();

        assertEquals(1, this.jdbcChannelMessageStore.messageGroupSize(MESSAGE_GROUP));
        Message<?> messageFromStore = this.jdbcChannelMessageStore.pollMessageFromGroup(MESSAGE_GROUP);

        assertNotNull(messageFromStore);
        assertEquals(message, messageFromStore);
    }

    @Configuration
    @EnableIntegration
    public static class ContextConfiguration {

        @Bean
        public PlatformTransactionManager transactionManager() {
            return new DataSourceTransactionManager(dataSource);
        }

        @Bean
        public ChannelMessageStoreQueryProvider queryProvider() {
            return new H2ChannelMessageStoreQueryProvider();
        }

        @Bean
        public JdbcChannelMessageStore jdbcChannelMessageStore() {
            JdbcChannelMessageStore jdbcChannelMessageStore = new JdbcChannelMessageStore(dataSource);
            jdbcChannelMessageStore.setChannelMessageStoreQueryProvider(queryProvider());
            return jdbcChannelMessageStore;
        }

        @Bean
        public PollableChannel transactionalQueueChannel() {
            return new QueueChannel(new MessageGroupQueue(jdbcChannelMessageStore(), MESSAGE_GROUP));
        }

        @Bean
        public TransactionInterceptor transactionInterceptor() {
            return new TransactionInterceptorBuilder()
                    .transactionManager(transactionManager())
                    .isolation(Isolation.READ_COMMITTED)
                    .propagation(Propagation.REQUIRED)
                    .build();
        }

        @Bean
        public TaskExecutor threadPoolTaskExecutor() {
            ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
            threadPoolTaskExecutor.setCorePoolSize(5);
            return threadPoolTaskExecutor;
        }

        @Bean(name = PollerMetadata.DEFAULT_POLLER)
        public PollerMetadata poller() {
            return Pollers.fixedDelay(10)
                    .advice(transactionInterceptor())
                    .taskExecutor(threadPoolTaskExecutor())
                    .get();
        }

        @Bean
        public CountDownLatch exceptionLatch() {
            return new CountDownLatch(2);
        }

        @ServiceActivator(inputChannel = "transactionalQueueChannel")
        public void handle(Message<?> message) {
            System.out.println(message);
            try {
                throw new RuntimeException("Intentional for rollback");
            }
            finally {
                exceptionLatch().countDown();
            }
        }

    }

}
user_27565470

Thanks to Artem Bilan for your great support. I finally found the solution. It seemed like there was another bean with the name transactionManager and transactionInterceptor active. This resulted in the strange behavior, that my trans-manager was never initialized, instead the other transactionmanager (null) was used for the transactioninterceptor and the PollingConsumer. Thats why my Transactionmanager in PollingConsumer was null, and why my Transactions were never working.

The solution was to rename all my beans, for some beans I also used the annotation @Primary to tell spring to always use this speciffic bean when autowired.

I also downgraded two 4.3, just to make sure this wasn't an error related to Version 5. I haven't testet if it would work with V 5 yet, but I think it should work also.

user_89325310
Related