How to configure a TransactionInterceptor equivalent to the @Transactional annotation for use with PollerMetadata advice

Question

I want to configure my custom PollerMetaData bean to enable transactions with a JdbcPollingChannelAdapter's UPDATE query. I'm using Spring boot with Spring Data JPA and all auto configuration enabled.

@Configuration
public class PollerMetaDataConfigurer {
    @Autowired
    //edit
    PlatformTransactionManager ptm;

    @Bean("CustomPoller")
    PollerMetadata getCustomPollermetaData(){
        PollerMetadata transactionConfiguredPoller=new PollerMetadata();
        TransactionInterceptor transactionInterceptor=new TransactionInterceptor();

        transactionInterceptor.setTransactionManager(ptm);
        List<Advice> pollerAdvice=new LinkedList<>();
        //edit
        pollerAdvice.add(transactionInterceptor);
        transactionConfiguredPoller.setAdviceChain(pollerAdvice);
        transactionConfiguredPoller.setTrigger(new PeriodicTrigger(2000));
        return transactionConfiguredPoller;
    }
}

This is the integration code I want transactions to work with.

@Configuration
@EnableIntegration    
public class IntegrationMain {

    @Bean
    public MessageChannel aChannel() {
        return new DirectChannel();
    }

    @Splitter(inputChannel = "inChannel", outputChannel = "outChannel")
    public List<myEnity> splitter(Message message) {

    }
    @Bean
    @InboundChannelAdapter(value = "inChannel", poller = @Poller(value="CustomPoller"))
    public MessageSource<?> pollDbForQuery(DataSource dataSource) {
        JdbcPollingChannelAdapter myAdapter = new JdbcPollingChannelAdapter(dataSource, SQL); 
        myAdapter.setRowMapper();
        myAdapter.setUpdateSql("update query");
        retrun adapter;


    }
}

Currently the transaction Interceptor is invoked after the UPDATE query has already been executed.

EDIT: I have set transaction.interceptor to TRACE and I can see the message

No need to create transaction for [org.springframework.integration.endpoint.AbstractPollingEndpoint$$Lambda$861/0x0000000800832520.call]: This method is not transactional.

Answer 1

Looks like you didn't finish to configure a TransactionInterceptor. See if this helps you:

    MatchAlwaysTransactionAttributeSource txAttributeSource = new MatchAlwaysTransactionAttributeSource();
    txAttributeSource.setTransactionAttribute(new DefaultTransactionAttribute());
    transactionInterceptor.setTransactionAttributeSource(txAttributeSource);

Since you didn't create it as a bean it afterPropertiesSet() is not called:

public void afterPropertiesSet() {
    if (getTransactionManager() == null && this.beanFactory == null) {
        throw new IllegalStateException(
                "Set the 'transactionManager' property or make sure to run within a BeanFactory " +
                "containing a TransactionManager bean!");
    }
    if (getTransactionAttributeSource() == null) {
        throw new IllegalStateException(
                "Either 'transactionAttributeSource' or 'transactionAttributes' is required: " +
                "If there are no transactional methods, then don't use a transaction aspect.");
    }
}

You can use a convenient TransactionInterceptorBuilder from Spring Integration.