Forcefully commit transaction when shutting an integration flow with an JMS/ActiveMQ inbound-CodePudding

We have a dynamically registered (and later started by an event) integration flow that goes like this:

private IntegrationFlow createTxxIntegrationFlow(Long taskId) {
        return IntegrationFlows 
            // INBOUND
                .from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory)
                         .id("jmsInboundT"   taskId).autoStartup(false)
                         .destination(new ActiveMQQueue("T"   taskId   "?consumer.exclusive=true&consumer.prefetchSize=0")))

            // PROCESSINGS
                ...
                // several handlers and transformers... not relevant here
                
            // OUTBOUND
                .routeToRecipients(r -> r// write to ActiveMQ
                                         .recipientFlow(f -> f.handle(Jms.outboundAdapter(jmsConnectionFactory).destination(myDestination)
                                         // stop only the inbound (using the control bus)
                                         //.recipientFlow(f -> f.handle((p, h) -> "@jmsInboundT" taskId ".stop()").channel(controlChannel))
                                         // or stop the whole flow (using the control bus)
                                         .recipientFlow(f -> f.handle((p, h) -> "@T" taskId ".stop()").channel(controlChannel))
                )
               .get();
}

The purpose is to have an exclusive consumer on an ActiveMQ queue, that disconnects after reading one message, and can be restarted later. Stopping and starting the flow is done using a control bus. With the flow defined above, after the inbound/flow is stopped I get the following log message:

2022-12-16 11:19:20.601  WARN 6576 --- [erContainer#0-1] o.s.j.l.DefaultMessageListenerContainer  : Setup of JMS message listener invoker failed for destination 'queue://T65' - trying to recover. Cause: The Session is closed

I do read one message from ActiveMQ, process it and stop the flow, but the transaction is not committed, and the ActiveMQ message gets put back in the queue.

After some google searching, we came up with the following update:

// autowired
private final PlatformTransactionManager txManager;

public void handleTransaction() {
        var status = txManager.getTransaction(txDefinition);
        txManager.commit(status);
}

private IntegrationFlow createTxxIntegrationFlow(Long taskId) {
        return IntegrationFlows 
            // INBOUND
                .from(Jms.messageDrivenChannelAdapter(jmsConnectionFactory)
                         .id("jmsInboundT"   taskId).autoStartup(false)
                         .destination(new ActiveMQQueue("T"   taskId   "?consumer.exclusive=true&consumer.prefetchSize=0"))
                         .configureListenerContainer(c -> c.transactionManager(txManager)))

            // PROCESSINGS
                ...
                
            // OUTBOUND MESSAGE
                .routeToRecipients(r -> r// write to ActiveMQ
                                         .recipientFlow(f -> f.handle(Jms.outboundAdapter(jmsConnectionFactory).destination(myDestination)
                                         // commit the transaction
                                         .recipientFlow(f -> f.handle(this, "handleTransaction"))
                                         // stop the inbound (using the control bus)
                                         //.recipientFlow(f -> f.handle((p, h) -> "@jmsInboundT" taskId ".stop()").channel(controlChannel))
                                         // stop the whole flow (using the control bus)
                                         .recipientFlow(f -> f.handle((p, h) -> "@T" taskId ".stop()").channel(controlChannel))
                )
               .get();
}

It works as expected: the transaction is committed, the ActiveMQ message consumed and not put back in the queue, and the inbound/flow stopped. But now, the application keeps disconnecting from the broker:

2022-12-16 11:50:23.782  INFO 14908 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61616
2022-12-16 11:50:25.852  INFO 14908 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61616
2022-12-16 11:50:26.872  INFO 14908 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61616
2022-12-16 11:50:27.891  INFO 14908 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61616
2022-12-16 11:50:29.914  INFO 14908 --- [ActiveMQ Task-1] o.a.a.t.failover.FailoverTransport       : Successfully connected to tcp://localhost:61616

We no longer have and exclusive connection to the broker, since the connection is discontinued.

How can we effectively force-commit the transaction before stopping the flow, without "breaking" the connection first?

CodePudding user response：

I believe all the code you show is performed on the same consumer thread. Even that .recipientFlow(f -> f.handle((p, h) -> "@T" taskId ".stop()").channel(controlChannel)) - call to Control Bus, is happened on that thread as well. So, you are not done with the consumer yet to make it to commit TX, but you are already stopping it. Or consider to have that controlChannel publishing to be done in separate thread, e.g. an ExecutorChannel.

Another thought: why would one use a message-driven channel adapter (JMS consumer) just for one message to pull? Why the plain JmsTemplate.receive(String destinationName) doesn't work for you? So, you would not need to register dynamic flow and you would not need to think about stopping it afterwards.