Spring Integration - JdbcPollingChannelAdapter фиксирует вместо отката обработанных исключений

Я использую Spring 4.1.x APIs, Spring Integration 4.1.x APIs и Spring Integration Java DSL 1.0.x APIs для потока EIP, где мы получаем сообщения из таблицы базы данных Oracle, используя JdbcPollingChannelAdpater в качестве точки входа в поток.

Несмотря на то, что у нас есть ErrorHandler, настроенный на JdbcPollingChannelAdapter в Poller, мы видим, что сеанс Transaction все еще откатывается и не фиксируется, когда RuntimeException генерируется и правильно обрабатывается ErrorHandler.

После прочтения этого потока: Spring Transactions - Предотвращение отката после непроверенных исключений (RuntimeException ), у меня такое ощущение, что невозможно предотвратить откат и вместо этого принудительно выполнить фиксацию. Это правильно? И, если есть способ, какой самый чистый способ принудительно выполнить фиксацию вместо отката, когда ошибка безопасно обрабатывается?

Текущая конфигурация:

ИнтеграцияConfig.java:

@Bean
public MessageSource<Object> jdbcMessageSource() {

    JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
            dataSource,
            "select * from SERVICE_TABLE where rownum <= 10 for update skip locked");
    adapter.setUpdateSql("delete from SERVICE_TABLE where SERVICE_MESSAGE_ID in (:id)");
    adapter.setRowMapper(serviceMessageRowMapper);
    adapter.setMaxRowsPerPoll(1);
    adapter.setUpdatePerRow(true);
    return adapter;
}

@SuppressWarnings("unchecked")
@Bean
public IntegrationFlow inFlow() {

    return IntegrationFlows
            .from(jdbcMessageSource(),
                    c -> {
                        c.poller(Pollers.fixedRate(100)
                                .maxMessagesPerPoll(10)
                                .transactional(transactionManager)
                                .errorHandler(errorHandler));
                    })
                .channel(inProcessCh()).get();
}

ErrorHandler.java

@Component
public class ErrorHandler implements org.springframework.util.ErrorHandler {

    @Autowired
    private PlatformTransactionManager transactionManager;

    private static final Logger logger = LogManager.getLogger();

    @Override
    public void handleError(Throwable t) {

        logger.trace("handling error:{}", t.getMessage(), t);

        // handle error code here...

        // we want to force commit the transaction here?
        TransactionStatus txStatus = transactionManager.getTransaction(null);
        transactionManager.commit(txStatus);
    }
}

--- ИЗМЕНЕНО для включения компонента ExpressionEvaluatingRequestHandlerAdvice ---

@Bean
public Advice expressionEvaluatingRequestHandlerAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice expressionEvaluatingRequestHandlerAdvice = new ExpressionEvaluatingRequestHandlerAdvice();
    expressionEvaluatingRequestHandlerAdvice.setTrapException(true);
    expressionEvaluatingRequestHandlerAdvice.setOnSuccessExpression("payload");
    expressionEvaluatingRequestHandlerAdvice
            .setOnFailureExpression("payload");
    expressionEvaluatingRequestHandlerAdvice.setFailureChannel(errorCh());
    return expressionEvaluatingRequestHandlerAdvice;
}

--- ОТРЕДАКТИРОВАНО, чтобы показать обработчик фиктивного тестового сообщения ---

    .handle(Message.class,
                    (m, h) -> {

                        boolean forceTestError = m.getHeaders().get("forceTestError");
                        if (forceTestError) {
                            logger.trace("simulated forced TestException");
                            TestException testException = new TestException(
                                    "forced test exception");
                            throw testException;
                        }

                        logger.trace("simulated successful process");

                        return m;
                    }, e-> e.advice(expressionEvaluatingRequestHandlerAdvice())

--- ИЗМЕНЕНО, чтобы показать метод ExecutorChannelInterceptor ---

@Bean
public IntegrationFlow inFlow() {

    return IntegrationFlows
            .from(jdbcMessageSource(), c -> {
                c.poller(Pollers.fixedRate(100).maxMessagesPerPoll(10)
                        .transactional(transactionManager));
            })
            .enrichHeaders(h -> h.header("errorChannel", errorCh(), true))
            .channel(
                    MessageChannels.executor("testSyncTaskExecutor",
                            syncTaskExecutor()).interceptor(
                            testExecutorChannelInterceptor()))
            .handle(Message.class, (m, h) -> {
                    boolean forceTestError = m.getHeaders().get("forceTestError");
                    if (forceTestError) {
                        logger.trace("simulated forced TestException");
                        TestException testException = new TestException(
                                "forced test exception");
                        throw testException;
                    }

                    logger.trace("simulated successful process");
            }).channel("nullChannel").get();
}