Re: Camel CDI with standalone JTA in Weld SE

[Gary Hodgson <ga...@gmail.com>]

Hi,

I just wanted to update this thread that I made some progress and believe I
have a somewhat working version with JavaSE, narayana, camel and activemq.
Using the example in [0] I tried using a JcaPooledConnectionFactory to
handle the XA and this seems to work. The pertinent commit is [1].

There's still work to be done as there are warnings about the jms session
already being closed, but I suspect this is a configuration problem.  Also
I of course want to further test the XA features to confirm all runs as
expected.

Regards,
Gary

[0]
https://github.com/welshstew/spring-boot-narayana-jms-xa-example/blob/master/src/main/java/bootwildfly/MySpringBootRouter.java
[1]
https://github.com/garyhodgson/camel-example-cdi-standalone-jta/commit/1cf374c63762b28c4a98e78d90a933d0edfb879d#diff-db968ea7d57eb130360c9c6563c562c2

On 30 November 2017 at 23:31, Gary Hodgson <ga...@gmail.com> wrote:

> Hi Antonin
>
> So I created a branch which attempts to use Narayana.  https://github.com/
> garyhodgson/camel-example-cdi-standalone-jta/tree/narayana
>
> I see the arjuna transaction manager starts up but the logs seem to
> indicate too little interaction with the activemq connection factory I
> think, and the results are not as I expect (the logs end with "middle
> QueueSize = 2" where it should only be one).  I've posted the output of the
> camel:run command in the off-chance anyone wants to take a look:
> https://gist.github.com/garyhodgson/c96bc6c90656c5f1a2af21f95caacf9f
>
> I'll look forward to any examples you get around to makeing - hopefully
> they will help show me where I'm going wrong.
>
> Cheers,
> Gary
>
> On 28 November 2017 at 18:00, Antonin Stefanutti <an...@stefanutti.fr>
> wrote:
>
>> Hi Gary,
>>
>> Would that be possible for you to test with anther transaction
>> implementation?
>>
>> IIRC I add some issues with Atomikos and then switch to JBoss JTA /
>> Narayana and it worked.
>>
>> I plan to add a Camel CDI / JMS / JTA example ASAP which should cover
>> that.
>>
>> Antonin
>>
>> > On 28 Nov 2017, at 00:25, Gary Hodgson <ga...@gmail.com>
>> wrote:
>> >
>> > Scratch that - setting
>> > the activeMQConfiguration.setUseSingleConnection(true) causes
>> everything to
>> > run in the same thread and the problem still occurs.
>> >
>> > On 27 November 2017 at 23:59, Gary Hodgson <ga...@gmail.com>
>> wrote:
>> >
>> >> Hi
>> >>
>> >> So after a hiatus I got time to look at this problem again.  From
>> delving
>> >> into the atomikos code and the trace logs I believe the issue lies in
>> >> threading.  As shown in the logs below the atomikos jms factory bean
>> >> appears to correctly create a composite transaction on the camel
>> >> JmsConsumer thread. However the jms session requesting the transaction
>> is
>> >> running on the main thread and cannot find it.  (The composite
>> transaction
>> >> managers appear to be stored in a map keyed by thread).
>> >>
>> >> I realise this is quite a specific problem which the majority of people
>> >> avoid by (sensibly) running camel-cdi and jta in wildfly, but if
>> anyone has
>> >> experience or ideas on how to make camel and atomikos find each other
>> here
>> >> it would be most appreciated.  I'll also have a go with Narayana
>> whether
>> >> that yields more success.
>> >>
>> >> Transaction created on this thread...
>> >>
>> >>  [Camel (camel-1) thread #1 - JmsConsumer[start]] DEBUG
>> >> com.atomikos.icatch.imp.CompositeTransactionManagerImp -
>> >> createCompositeTransaction ( 10000 ): created new ROOT transaction
>> with id
>> >> 10.0.75.1.tm151182109511700001
>> >>  [Camel (camel-1) thread #1 - JmsConsumer[start]] DEBUG
>> com.atomikos.jms.AtomikosConnectionFactoryBean
>> >> - AtomikosConnectionFactoryBean 'xamq': createConnection()...
>> >>  [Camel (camel-1) thread #1 - JmsConsumer[start]] INFO
>> com.atomikos.jms.AtomikosConnectionFactoryBean
>> >> - AtomikosConnectionFactoryBean 'xamq': init...
>> >>  [Camel (camel-1) thread #1 - JmsConsumer[start]] TRACE
>> >> com.atomikos.icatch.imp.CompositeTransactionManagerImp -
>> >> getCompositeTransaction()  returning instance with id
>> >> 10.0.75.1.tm151182109511700001
>> >>  [Camel (camel-1) thread #1 - JmsConsumer[start]] TRACE
>> >> com.atomikos.datasource.pool.ConnectionPool - atomikos connection pool
>> >> 'xamq': about to wait for connection during 30000ms...
>> >>  [main] TRACE com.atomikos.datasource.xa.session.SessionHandleState -
>> a
>> >> SessionHandleState with 0 context(s): notifySessionBorrowed
>> >>  [main] TRACE com.atomikos.datasource.xa.session.TransactionContext -
>> a
>> >> TransactionContext: changing to state com.atomikos.datasource.xa.ses
>> sion.
>> >> NotInBranchStateHandler@a826ff8
>> >>  [main] DEBUG com.atomikos.jms.AtomikosJmsXaSessionProxy - atomikos xa
>> >> session proxy for resource xamq: calling toString on JMS driver
>> session...
>> >>  [main] TRACE com.atomikos.jms.AtomikosJmsXaSessionProxy - atomikos xa
>> >> session proxy for resource xamq: toString returning ActiveMQSession
>> >> {id=ID:garyhodgson-PC-57091-1511821093820-1:1:2,started=false}
>> >> java.lang.Object@63187d63
>> >>  [main] TRACE com.atomikos.jms.AtomikosJmsConnectionProxy - atomikos
>> >> connection proxy for resource xamq: returning ActiveMQSession
>> >> {id=ID:garyhodgson-PC-57091-1511821093820-1:1:2,started=false}
>> >> java.lang.Object@63187d63
>> >>  [main] DEBUG com.atomikos.jms.AtomikosJmsXaSessionProxy - atomikos xa
>> >> session proxy for resource xamq: calling createQueue on JMS driver
>> >> session...
>> >>  [main] TRACE com.atomikos.jms.AtomikosJmsXaSessionProxy - atomikos xa
>> >> session proxy for resource xamq: createQueue returning queue://start
>> >>  [main] DEBUG com.atomikos.jms.AtomikosJmsXaSessionProxy - atomikos xa
>> >> session proxy for resource xamq: calling createProducer on JMS driver
>> >> session ActiveMQSession {id=ID:garyhodgson-PC-57091-
>> >> 1511821093820-1:1:2,started=false} java.lang.Object@63187d63
>> >>  [main] TRACE com.atomikos.jms.AtomikosJmsXaSessionProxy - atomikos xa
>> >> session proxy for resource xamq: createProducer returning atomikos
>> >> MessageProducer proxy for ActiveMQMessageProducer {
>> >> value=ID:garyhodgson-PC-57091-1511821093820-1:1:2:1 }
>> >>  [main] DEBUG com.atomikos.jms.AtomikosJmsXaSessionProxy - atomikos xa
>> >> session proxy for resource xamq: calling createTextMessage on JMS
>> driver
>> >> session...
>> >>  [main] TRACE com.atomikos.jms.AtomikosJmsXaSessionProxy - atomikos xa
>> >> session proxy for resource xamq: createTextMessage returning
>> >> ActiveMQTextMessage {commandId = 0, responseRequired = false,
>> messageId =
>> >> null, originalDestination = null, originalTransactionId = null,
>> producerId
>> >> = null, destination = null, transactionId = null, expiration = 0,
>> timestamp
>> >> = 0, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId =
>> >> null, replyTo = null, persistent = false, type = null, priority = 0,
>> >> groupID = null, groupSequence = 0, targetConsumerId = null, compressed
>> =
>> >> false, userID = null, content = null, marshalledProperties = null,
>> >> dataStructure = null, redeliveryCounter = 0, size = 0, properties =
>> null,
>> >> readOnlyProperties = false, readOnlyBody = false, droppable = false,
>> >> jmsXGroupFirstForConsumer = false, text = null}
>> >>  [main] DEBUG com.atomikos.jms.AtomikosJmsMessageProducerProxy -
>> >> atomikos MessageProducer proxy for ActiveMQMessageProducer {
>> >> value=ID:garyhodgson-PC-57091-1511821093820-1:1:2:1 }: send ( message
>> )...
>> >>
>> >> Attempt to get transaction here...
>> >>
>> >>  [main] TRACE com.atomikos.icatch.imp.CompositeTransactionManagerImp -
>> >> getCompositeTransaction() returning NULL!
>> >>  [main] WARN com.atomikos.jms.ConsumerProducerSupport - atomikos
>> >> MessageProducer proxy for ActiveMQMessageProducer {
>> >> value=ID:garyhodgson-PC-57091-1511821093820-1:1:2:1 }: The JMS session
>> >> you are using requires a JTA transaction context for the calling
>> thread and
>> >> none was found.
>> >>  Please correct your code to do one of the following:
>> >>  1. start a JTA transaction if you want your JMS operations to be
>> subject
>> >> to JTA commit/rollback, or
>> >>  2. increase the maxPoolSize of the AtomikosConnectionFactoryBean to
>> >> avoid transaction timeout while waiting for a connection, or
>> >>  3. create a non-transacted session and do session acknowledgment
>> >> yourself, or
>> >>  4. set localTransactionMode to true so connection-level
>> commit/rollback
>> >> are enabled.
>> >>
>> >>
>> >> Cheers,
>> >> Gary
>> >>
>> >>
>> >> On 6 November 2017 at 18:16, Gary Hodgson <ga...@gmail.com>
>> >> wrote:
>> >>
>> >>> Hi Antonin.
>> >>>
>> >>> Thanks for the response.
>> >>>
>> >>> I attempted a naive implementation of TransactionServices and it
>> appears
>> >>> to load via the serviceloader correctly, and produce UserTransactions:
>> >>> https://github.com/garyhodgson/camel-example-cdi-standalone-
>> >>> jta/blob/TransactionServices/src/main/java/org/apache/
>> >>> camel/example/cdi/util/TransactionServices.java
>> >>>
>> >>> Sadly the AtomikosTransactionRequiredJMSException is still thrown.
>> >>>
>> >>>
>> >>> On 6 November 2017 at 11:52, Antonin Stefanutti <
>> antonin@stefanutti.fr>
>> >>> wrote:
>> >>>
>> >>>> Hi Gary,
>> >>>>
>> >>>> Your CDI producers for the ActiveMQ component and the transaction
>> >>>> manager look OK (aside that you would have to destroy the connection
>> >>>> factory in a dispose method).
>> >>>>
>> >>>> However The way you produce the UserTransaction may be problematic.
>> With
>> >>>> Weld SE, you would typically implement org.jboss.weld.transaction.spi
>> .TransactionServices
>> >>>> and add it as a Weld service before starting the CDI container.
>> >>>>
>> >>>> Let me know if that helps.
>> >>>>
>> >>>> Antonin
>> >>>>
>> >>>>> On 5 Nov 2017, at 21:54, Gary Hodgson <ga...@gmail.com>
>> >>>> wrote:
>> >>>>>
>> >>>>> Hi,
>> >>>>>
>> >>>>> I am trying to get camel-cdi to work with a standalone JTA provider
>> in
>> >>>> a
>> >>>>> non-JEE environment.  I've made some progress but have a couple of
>> >>>> issues
>> >>>>> which I am not sure is due to my understanding or the code.
>> >>>>>
>> >>>>> I created a standalone demo project here to show the issues:
>> >>>>> https://github.com/garyhodgson/camel-example-cdi-standalone-jta
>> >>>>>
>> >>>>> The demo uses Atomikos as the JTA provider and embedded ActiveMQ for
>> >>>> JMS.
>> >>>>> Running the project with camel:run results in three messages being
>> >>>> sent to
>> >>>>> the route. The first two are called explicitly in a userTransaction
>> and
>> >>>>> behave as expected: the first is processed normally, the second
>> >>>> triggers a
>> >>>>> rollback (as expected).
>> >>>>>
>> >>>>> The third message is outside a userTransaction and throws the
>> following
>> >>>>> exception from Atomikos:
>> >>>>>
>> >>>>>   Caused by: com.atomikos.jms.AtomikosTrans
>> >>>> actionRequiredJMSException:
>> >>>>> The JMS session you are using requires a JTA transaction context for
>> >>>> the
>> >>>>> calling thread and none was found.
>> >>>>>   Please correct your code to do one of the following:
>> >>>>>   1. start a JTA transaction if you want your JMS operations to be
>> >>>>> subject to JTA commit/rollback, or
>> >>>>>   2. increase the maxPoolSize of the AtomikosConnectionFactoryBean
>> to
>> >>>>> avoid transaction timeout while waiting for a connection, or
>> >>>>>   3. create a non-transacted session and do session acknowledgment
>> >>>>> yourself, or
>> >>>>>   4. set localTransactionMode to true so connection-level
>> >>>> commit/rollback
>> >>>>> are enabled.
>> >>>>>     at
>> >>>>> com.atomikos.jms.AtomikosTransactionRequiredJMSException.thr
>> >>>> owAtomikosTransactionRequiredJMSException(AtomikosTransactio
>> >>>> nRequiredJMSException.java:23)
>> >>>>> ~[transactions-jms-4.0.4.jar:?]
>> >>>>>     at
>> >>>>> com.atomikos.jms.ConsumerProducerSupport.enlist(ConsumerProd
>> >>>> ucerSupport.java:90)
>> >>>>> ~[transactions-jms-4.0.4.jar:?]
>> >>>>>     at
>> >>>>> com.atomikos.jms.AtomikosJmsMessageProducerProxy.send(Atomik
>> >>>> osJmsMessageProducerProxy.java:34)
>> >>>>> ~[transactions-jms-4.0.4.jar:?]
>> >>>>>     at
>> >>>>> org.springframework.jms.core.JmsTemplate.doSend(JmsTemplate.
>> java:626)
>> >>>>> ~[spring-jms-4.3.11.RELEASE.jar:4.3.11.RELEASE]
>> >>>>>     at
>> >>>>> org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemp
>> >>>> late.doSend(JmsConfiguration.java:624)
>> >>>>> ~[camel-jms-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemp
>> >>>> late.doSendToDestination(JmsConfiguration.java:563)
>> >>>>> ~[camel-jms-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemp
>> >>>> late.access$100(JmsConfiguration.java:505)
>> >>>>> ~[camel-jms-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemp
>> >>>> late$1.doInJms(JmsConfiguration.java:519)
>> >>>>> ~[camel-jms-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.springframework.jms.core.JmsTemplate.execute(JmsTemplate
>> .java:484)
>> >>>>> ~[spring-jms-4.3.11.RELEASE.jar:4.3.11.RELEASE]
>> >>>>>     at
>> >>>>> org.apache.camel.component.jms.JmsConfiguration$CamelJmsTemp
>> >>>> late.send(JmsConfiguration.java:516)
>> >>>>> ~[camel-jms-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.component.jms.JmsProducer.doSend(JmsProduce
>> >>>> r.java:440)
>> >>>>> ~[camel-jms-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.component.jms.JmsProducer.processInOnly(Jms
>> >>>> Producer.java:394)
>> >>>>> ~[camel-jms-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.component.jms.JmsProducer.process(JmsProduc
>> >>>> er.java:157)
>> >>>>> ~[camel-jms-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.processor.SharedCamelInternalProcessor.proc
>> >>>> ess(SharedCamelInternalProcessor.java:186)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.processor.SharedCamelInternalProcessor.proc
>> >>>> ess(SharedCamelInternalProcessor.java:86)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.impl.ProducerCache$1.doInProducer(ProducerC
>> >>>> ache.java:541)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.impl.ProducerCache$1.doInProducer(ProducerC
>> >>>> ache.java:506)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.impl.ProducerCache.doInProducer(ProducerCac
>> >>>> he.java:369)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.impl.ProducerCache.sendExchange(ProducerCac
>> >>>> he.java:506)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     at org.apache.camel.impl.ProducerCache.send(ProducerCache.java:
>> >>>> 229)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.impl.DefaultProducerTemplate.send(DefaultPr
>> >>>> oducerTemplate.java:144)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     at
>> >>>>> org.apache.camel.impl.DefaultProducerTemplate.sendBody(Defau
>> >>>> ltProducerTemplate.java:161)
>> >>>>> ~[camel-core-2.20.0.jar:2.20.0]
>> >>>>>     ... 56 more
>> >>>>>
>> >>>>>
>> >>>>> I expected that "transacted()" on the route to have caused a
>> >>>> transaction to
>> >>>>> be created, and I am not sure why it has not been - other than it
>> >>>> having
>> >>>>> something to do with this example running outside of a JEE
>> Application
>> >>>>> Server.  Is my expectation here incorrect?
>> >>>>>
>> >>>>> I believe the various required beans have been correctly configured
>> as
>> >>>>> Producer methods under org.apache.camel.example.cdi.u
>> til.CdiProducers
>> >>>>>
>> >>>>> As an aside: in order to get this working in a Weld SE environment I
>> >>>> use a
>> >>>>> CDI extension
>> >>>>> (org.apache.camel.example.cdi.util.ResourceInjectReplacement
>> Extension)
>> >>>> to
>> >>>>> add an @Inject annotation to the transactionManager attribute in
>> >>>>> org.apache.camel.cdi.transaction.JtaTransactionPolicy.  Perhaps it
>> >>>> would be
>> >>>>> worth considering adding the annotation in camel-cdi to enable it
>> to be
>> >>>>> used in a SE environment?  (though I don't know if this would
>> adversely
>> >>>>> affect the JEE world.)
>> >>>>>
>> >>>>>
>> >>>>> Any insights would be appreciated.
>> >>>>>
>> >>>>> Cheers,
>> >>>>> Gary
>> >>>>
>> >>>>
>> >>>
>> >>
>>
>>
>