JPA Binding - Transaction per Message

[sdirbach]

Hi,

I have a simple service with a (transacted) JPA Binding, that fetches Entities from the database and processes them.

As the entities are processed one by one, I would assume that each Service Invocation would take place in it's own transaction. Contrary to that, an exception in one processing rolls all changes on every entity back.

Is this the desired behaviour?

I also tried to suspend the incoming transaction and start a new one by corresponding transactionPolicies, but that also shows the same behaviour.

Kind regards,

Sascha

[igarashitm]

Hi Sascha,

You would need to declare distinct xa-datesource for each JPA binding if you want to have transaction boundary. That's the limitation on camel jms/sql/jpa binding.

hth,

Tomo

[sdirbach]

Hi Tomo,

I currently only have only one JPA Binding, so I don't see how a XA-Datasource would help here.

The service has this interface

public interface ServiceInterface {
    public void invoke(TestEntity te);
}

The JPA Binding looks like this:

<jpa:binding.jpa name="jpa1">
    <jpa:entityClassName>com.example.switchyard.switchyard_database_binding.TestEntity</jpa:entityClassName>
    <jpa:persistenceUnit>switchyard-database-binding</jpa:persistenceUnit>
    <jpa:transactionManager>#jtaTransactionManager</jpa:transactionManager>
    <jpa:consume>
       <jpa:delay>10000</jpa:delay>
       <jpa:initialDelay>500</jpa:initialDelay>
       <jpa:consumeDelete>false</jpa:consumeDelete>
       <jpa:consumer.namedQuery>TestEntity.fetchEntityNotProcessed</jpa:consumer.namedQuery>
       <jpa:consumer.transacted>true</jpa:consumer.transacted>
    </jpa:consume>
</jpa:binding.jpa>

The Named Query has this definition:

SELECT t FROM TestEntity t WHERE t.status <> com.example.switchyard.switchyard_database_binding.TestEntityStatus.PROCESSED AND t.status <> com.example.switchyard.switchyard_database_binding.TestEntityStatus.FAILED

The Service implementation is a simple Bean:

    public void invoke(TestEntity te) {
        System.out.println(te);
        te.setStatus(TestEntityStatus.PROCESSED);

        // Do business work
        if(te.getPayLoad().equals("fail")) {
             throw new RuntimeException("Cannot proceed");
        }
    }

Now when I have two entries in the Database, one with the payload "test" and one with the payload "fail", I would expect, that the first entry would be written back with the PROCESSED status value and that the second entry would be rolled back to the NEW status as a RuntimeException is thrown.

What I am seeing is that both entries are rolled back to the NEW status.

Kind regards,

Sascha

[igarashitm]

Hmm that sounds strange then. Could you attach your reproducer application ?

[sdirbach]

Hi,

yes of course

Sascha

[igarashitm]

Thanks, I'll take a look at later. But before that, I think I can guess what is the cause now - You may need to send the entity object to the JPA reference binding to persist it. Just changing the field doesn't reflect to the database.

[sdirbach]

Hi,

thanks for your help.

The thing is, that it works when I don't throw an exception (i.e. don't put a fail entry in the database), so the entities still seem to be bound to the PersistenceContext when they reach the bean implementation.

Sascha

[igarashitm]

Oops, I didn't notice the entity is still attached - Could you try adding "<jpa:maximumResults>1</jpa:maximumResults>" ? Apparently camel JPA consumer is doing batch processing on the result set retrieved with the query.

http://camel.apache.org/jpa.html

[sdirbach]

Hi,

it doesn't seem to accept the maximumResults attribute, as the value is ignored. I also tried the maxMessagesPerPoll attribute which seems to have the same intention.

The problem with this option is, that

1. it now takes x polls to process x messages, which is not really what I want and

2. it blocks at the failed message, as the message is rolled back to the NEW status. (Which is ok, if regular exceptions are caught and only real runtime exceptions like unreachable systems occur)

Anyway I would still prefer the processing of all entries returned by the query in seperate transactions.

Sascha

[igarashitm]

1. it now takes x polls to process x messages, which is not really what I want and

I think this is the only one way to have distinct transaction for each record. The query retrieving entity must belong to same transaction, otherwise you'd have to disable transaction (i.e. transacted=false).

2. it blocks at the failed message, as the message is rolled back to the NEW status. (Which is ok, if regular exceptions are caught and only real runtime exceptions like unreachable systems occur)

then how about just setting the status to FAILED and not throwing Exception?

[rcernich]

This is just a shot in the dark, but you might try setting transacted to false.  Based on the Camel documentation, the default is to commit all the successfully processed elements, rolling back the last failed element.  Something to try, anyway.

Rob

[sdirbach]

Hi,

igarashit

I could get it to work, that each processing happens in one seperate Transaction by applying the transaction policies suspendTransaction & managedTransaction.Global on my implementation bean. Furthermore I had to inject the Persistence Context in my Bean and call EntitiyManager.merge() on my Entity. Otherwise the entity still seems to be bound by the global (JPA-Binding) transaction. Anyway, the processing of all entites stops if one fails. Thats means if I have 3 entries in my database with these payloads:

1 - test

2 - fail

3 - anotherTest

The first entry would by processed and commited (as it is now bound to the internal transaction), but 2 and 3 won't be processed. This is due to the fact that the second processing fails and it stops the current polling. On the second polling it would again stop at this entry, which blocks the whole processing.

By the way: Yes, I thought about setting the FAILED Status, but I can't always catch any exception, thats why I hoped I could continue processing even if one entry fails.

rcernich

Interestingly this property does not change the behaviour, it still runs everything in one transaction.

[igarashitm]

I could get it to work, that each processing happens in one seperate Transaction by applying the transaction policies suspendTransaction & managedTransaction.Global on my implementation bean. Furthermore I had to inject the Persistence Context in my Bean and call EntitiyManager.merge() on my Entity. Otherwise the entity still seems to be bound by the global (JPA-Binding) transaction.

It sounds like same as transaction is disabled.

Anyway, the processing of all entites stops if one fails. Thats means if I have 3 entries in my database with these payloads:

1 - test

2 - fail

3 - anotherTest

The first entry would by processed and commited (as it is now bound to the internal transaction), but 2 and 3 won't be processed. This is due to the fact that the second processing fails and it stops the current polling. On the second polling it would again stop at this entry, which blocks the whole processing.

By the way: Yes, I thought about setting the FAILED Status, but I can't always catch any exception, thats why I hoped I could continue processing even if one entry fails.

I don't try it yet, but if you change the Bean service to IN_OUT (return a value but not void) and declare a Exception (say DeclaredFault) on its operation, then throwing DeclaredFault may not trigger a rollback.

Interestingly this property does not change the behaviour, it still runs everything in one transaction.

Did you remove transactionManager parameter as well?

[sdirbach]

I could get it to work, that each processing happens in one seperate Transaction by applying the transaction policies suspendTransaction & managedTransaction.Global on my implementation bean. Furthermore I had to inject the Persistence Context in my Bean and call EntitiyManager.merge() on my Entity. Otherwise the entity still seems to be bound by the global (JPA-Binding) transaction.

It sounds like same as transaction is disabled.

Yes, at least, what it should be like

Anyway, the processing of all entites stops if one fails. Thats means if I have 3 entries in my database with these payloads:

1 - test

2 - fail

3 - anotherTest

The first entry would by processed and commited (as it is now bound to the internal transaction), but 2 and 3 won't be processed. This is due to the fact that the second processing fails and it stops the current polling. On the second polling it would again stop at this entry, which blocks the whole processing.

By the way: Yes, I thought about setting the FAILED Status, but I can't always catch any exception, thats why I hoped I could continue processing even if one entry fails.

I don't try it yet, but if you change the Bean service to IN_OUT (return a value but not void) and declare a Exception (say DeclaredFault) on its operation, then throwing DeclaredFault may not trigger a rollback.

Ok, I tried that, but even though the log tells me, that it will commit the processed entries, it still fails due to a NoTransactionException:

Error processing last message due: org.apache.camel.RuntimeCamelException: java.lang.RuntimeException: Cannot proceed. Will commit all previous successful processed message, and ignore this last failure.: javax.persistence.PersistenceException: org.apache.camel.RuntimeCamelException: java.lang.RuntimeException: Cannot proceed

Caused by: javax.persistence.TransactionRequiredException: no transaction is in progress

Interestingly this property does not change the behaviour, it still runs everything in one transaction.

Did you remove transactionManager parameter as well?

No I didn't. I tried it and now it fails always and directly on the consumer (independent from the database row):

Caused by: java.lang.IllegalStateException: A JTA EntityManager cannot use getTransaction()

[igarashitm]

Looks like it's still trying to use JTA... did you set transaction-type="RESOURCE_LOCAL" on persistence.xml?