Update to use Spring JMS Template or DefaultMessageContainerListener

[mattrpav]

Is there a reason you are coding up JMS handling vs using the Spring JMS Template or DefaultMessageContainerListener in your app?

Those components have a lot of mileage on them and help avoid common issues when working with event-driven apps

[fcurvat]

I would say that's legacy stuff - mostly component is used in services but also in osgi container. Still i could try to use JMS Template. I can do a branch for that.

[fcurvat]

I also checked quickly on Spring and used this method https://github.com/spring-projects/spring-framework/blob/main/spring-jms/src/main/java/org/springframework/jms/support/JmsUtils.java#L146-L168 but that's not helping (and mostly hidding exceptions...)

[mattrpav]

I'm a big fan of OSGi!

Spring is essentially a different service wiring model. You can mix the two, but I'd recommend picking one-- Spring or OSGi for wiring all the services together.

Once you decide that, the service organization can be better defined.

Suggestion:

1. Think of the Connection Factory as a service component. i. In Spring init-method, destroy-method ii. In OSGi scr activate/decativate

2. PooledConnectionFactory as a second service component that takes the Connection Factory as an argument i. NOTE: start()/stop() must be called or pooling doesn't start

3. Separate Producer and Consumer Connection Factories into separate service instances

4. Event Producer Service takes the object -> convert to string (or bytes) and then takes the producer pooled connection factory component as an injected dependency.

5. Event Consumer Service takes the consumer pooled connection factory as an injected dependency. Start single threaded-- most cases it is fast enough since ActiveMQ uses prefetch to send 1,000 messages to the consumer at a time.

You can make #5 multi-threaded as a follow-on. Getting the service domain separation and start/stop lifecycle organized will help make this easier to test / troubleshoot and then scale!

[tlnd-fcurvat]

Sure, thanks for advices ! Do you mean you would have one ConnectionFactory for producer and another one for consumer ? (it is common on our side now). To give you more context :

• In production we use prefetch 100 - i think that's good for us, some events are long to read - and we don't want to block other events too much or get to much disorder.
• It's already working good either in osgi or springboot services. Some queues read up to 200msg/s, overall up to 750msg/s on persisted queues (posgresql)
• we use one connection, several consumers (5 to 10), HA on spring boot services (up to 10 pods)

BUT we read the payload, process the business, and then after we acknowledge the message (transacted sessions). We might consider do the other way around but there would be some cons. That's why troubles of shutdown is bothering - for some of the messages we do not want them to be replayed.

I did play with JmsTemplate (from a baeldung example spring-jms), but by default ack is done before processing the business logic. I don't see how i can change that but i will continue to play with that example (i can push it somewhere if you want too).

[tlnd-fcurvat]

Hello ! Last news are with Spring JmsTemplate, i can finally/hopefully get the ack after reading the message. Shutdown is fast and consumers/sessions are stopped quickly. So checking spring code to see how they do (trying to spot the difference...)

[tlnd-fcurvat]

Finally nailed the issue. On our side, we do two thing on the connection. stop() then close() https://github.com/fcurvat/amqpoc/blob/main/src/main/java/sample/rs/service/amq/AMQEventSourceEngine.java#L353-L355 If i do only stop everything is good. For me there is a problem with connection.close() that is somewhat buggy since some messages (usually one) are not ack.

In the end i get good result with

• just calling stop() on the connection
• call close() on consumers then call stop() on connection; according to javadoc, close() on connection is enough since it is closing the consumers too.

Any comment or recommendation is welcome. If you want i can fill a jira ticket for the bug. On my side i will probably do changes on our codebase to just use connection.stop()

[mattrpav]

Calling .stop() tells the connection to stop serving messages to consumers-- but does not close the connection and return it to the connection pool.

You should not need to use connection.stop() in most normal application processing scenarios. Simply use the .close() so the connection can be returned to the connection pool.

[tlnd-fcurvat]

Well on spring side, only connection.stop() is used. And close() indeed is not working so good for me (regular lost of ack on one message)