How enable a camel transaction from File consumer to JMS producer?

Question

I have a route which picks up a file, splits it into individual records, and posts these records to a JMS topic.

from("ftp:..."
    .unmarshal(bindy)
    .split(body())
    .marshal(jaxb)
    .to("activemq:topic:myTopic");

This has been working fine, until recently something went bizarrely wrong with ActiveMQ, and it successfully processed the first few messages in the file and then gave some bizarre exceptions. This resulted in the consumer assuming that the exchange had failed, and thus not archiving the file, and then reprocessing it. This time it succeeded, but that meant that the records that had already been processed had been repeated.

I've been reading about using transactions with JMS and Camel, but I can't understand how I get it to include all the records generated after the split into a transaction, such that if something goes wrong, the file is left, but any previously "successful" posts to the topic are ignored.

Does this require that shared units of work are enabled for the splitter?

As far as I could see, the .transacted() instruction applies to the consumer rather than the producer, and as I understand it, the file based components aren't transacted anyway.

How do I ensure that I get an "all or nothing" gets posted to the topic for each file consumed?

Thanks for looking!

johan johan · Accepted Answer · 2019-11-19T21:22:55

Using .transacted() should solve your issue, as the messages sent to ActiveMQ queue will be committed or rolled back together.

And yes, to ensure that the messages to ActiveMQ is handled in the same transaction, you will need to enable shareUnitOfWork() for the split.

I would personally also indent and end the split for readability.

Maybe something like:

from("ftp:..."
    .transacted()
    .unmarshal(bindy)
    .split(body()).shareUnitOfWork()
        .marshal(jaxb)
        .to("activemq:topic:myTopic")
    .end();

Also have a look at "indempotence" if you want to make sure that messages are only received once. https://camel.apache.org/manual/latest/idempotentConsumer-eip.html

How enable a camel transaction from File consumer to JMS producer?

1 Answers