[
https://issues.apache.org/jira/browse/CAMEL-12971?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17169840#comment-17169840
]
Claus Ibsen commented on CAMEL-12971:
-------------------------------------
Yeah I can see this gets complex as there dont seem to be a good neutral hook
in the JMS API for being able to commit the transaction on timeout safely,
without the session being used to receive another message.
There may be some ActiveMQ specific API but then it should be in camel-activemq
or some camel-activemq-sjms component that has special support for the old
classic ActiveMQ broker.
> SJMS Component javax.jms.JMSException: Unmatched acknowledge: MessageAck when
> transactionBatchTimeout expired
> -------------------------------------------------------------------------------------------------------------
>
> Key: CAMEL-12971
> URL: https://issues.apache.org/jira/browse/CAMEL-12971
> Project: Camel
> Issue Type: Bug
> Components: camel-sjms
> Affects Versions: 2.20.2
> Reporter: Alexey Kachalov
> Priority: Major
> Fix For: Future
>
> Attachments: BatchSynchronizationReproduceTest.java, test-result.log
>
>
> We use Apache ActiveMQ 5.15.2 and Camel 2.20.2.
> When we switched to use camel-sjms instead of camel-jms we sometimes get
> this error:
> {code:java}
> 2018-11-30 20:39:33,264 +0300 INFO 19 [TransactionContext] commit failed for
> transaction TX:ID:a-kachalov-61507-1543599563462-1:1:2
> javax.jms.JMSException: COMMIT FAILED: Transaction marked rollback only
> xaErrorCode:100
> at
> org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:54)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1399)
> ~[activemq-client-5.15.2.jar:?]
> at
> org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1428)
> ~[activemq-client-5.15.2.jar:?]
> at
> org.apache.activemq.BatchSynchronizationReproduceTest$2.answer(BatchSynchronizationReproduceTest.java:66)
> ~[classes/:?]
> at
> org.apache.activemq.TransactionContext.commit(TransactionContext.java:330)
> ~[activemq-client-5.15.2.jar:?]
> at org.apache.activemq.ActiveMQSession.commit(ActiveMQSession.java:582)
> [activemq-client-5.15.2.jar:?]
> at
> org.apache.camel.component.sjms.tx.SessionBatchTransactionSynchronization$TimeoutTask.run(SessionBatchTransactionSynchronization.java:135)
> [camel-sjms-2.20.2.jar:2.20.2]
> at java.util.TimerThread.mainLoop(Timer.java:555) [?:1.8.0_73]
> at java.util.TimerThread.run(Timer.java:505) [?:1.8.0_73]
> Caused by: javax.transaction.xa.XAException: COMMIT FAILED: Transaction
> marked rollback only xaErrorCode:100
> at
> org.apache.activemq.transaction.Transaction.newXAException(Transaction.java:213)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transaction.Transaction.prePrepare(Transaction.java:115)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transaction.LocalTransaction.commit(LocalTransaction.java:54)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransactionBroker.commitTransaction(TransactionBroker.java:252)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.BrokerFilter.commitTransaction(BrokerFilter.java:114)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection.processCommitTransactionOnePhase(TransportConnection.java:529)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.command.TransactionInfo.visit(TransactionInfo.java:100)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:330)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:194)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:125)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:301)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:235)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:216)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_73]
> Caused by: javax.jms.TransactionRolledBackException: COMMIT FAILED:
> Transaction marked rollback only xaErrorCode:100
> at
> org.apache.activemq.transaction.Transaction.prePrepare(Transaction.java:116)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transaction.LocalTransaction.commit(LocalTransaction.java:54)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransactionBroker.commitTransaction(TransactionBroker.java:252)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.BrokerFilter.commitTransaction(BrokerFilter.java:114)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection.processCommitTransactionOnePhase(TransportConnection.java:529)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.command.TransactionInfo.visit(TransactionInfo.java:100)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:330)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:194)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:125)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:301)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:235)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:216)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_73]
> Caused by: javax.jms.JMSException: Unmatched acknowledge: MessageAck
> {commandId = 9, responseRequired = false, ackType = 2, consumerId =
> ID:a-kachalov-61507-1543599563462-1:1:1:1, firstMessageId =
> ID:a-kachalov-61507-1543599563462-1:2:1:1:1, lastMessageId =
> ID:a-kachalov-61507-1543599563462-1:2:1:1:2, destination =
> queue://TEST.QUEUE, transactionId =
> TX:ID:a-kachalov-61507-1543599563462-1:1:2, messageCount = 2, poisonCause =
> null}; Could not find Message-ID ID:a-kachalov-61507-1543599563462-1:2:1:1:1
> in dispatched-list (start of ack)
> at
> org.apache.activemq.broker.region.PrefetchSubscription.assertAckMatchesDispatched(PrefetchSubscription.java:473)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.region.PrefetchSubscription.acknowledge(PrefetchSubscription.java:213)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.region.AbstractRegion.acknowledge(AbstractRegion.java:528)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.region.RegionBroker.acknowledge(RegionBroker.java:475)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:89)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransactionBroker.acknowledge(TransactionBroker.java:276)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.BrokerFilter.acknowledge(BrokerFilter.java:89)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection.processMessageAck(TransportConnection.java:581)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at org.apache.activemq.command.MessageAck.visit(MessageAck.java:245)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection.service(TransportConnection.java:330)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.broker.TransportConnection$1.onCommand(TransportConnection.java:194)
> ~[activemq-broker-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.MutexTransport.onCommand(MutexTransport.java:50)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.WireFormatNegotiator.onCommand(WireFormatNegotiator.java:125)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.AbstractInactivityMonitor.onCommand(AbstractInactivityMonitor.java:301)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.TransportSupport.doConsume(TransportSupport.java:83)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.tcp.TcpTransport.doRun(TcpTransport.java:235)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at
> org.apache.activemq.transport.tcp.TcpTransport.run(TcpTransport.java:216)
> ~[activemq-client-5.15.2.jar:5.15.2]
> at java.lang.Thread.run(Thread.java:745) ~[?:1.8.0_73]
> {code}
> This happens because Batch Transaction Timer expires and commit is done in it
> own thread.
> We use a route such as
> {code:java}
> sjms:queue:TEST.QUEUE?transacted=true&transactionBatchCount=30&transactionBatchTimeout=1000
> {code}
>
> Steps of reproducing:
> * The first message is dispatched to consumer
> * Consumer waits the next message, but Timer expired and
> SessionBatchTransactionSynchronization.TimeoutTask executes in another thread
> and invokes commit() method
> * Commit consist of 3 operations: beforeEnd(), asyncSendPacket(...) and
> afterCommit() methods
> * When beforeEnd() finishes, transactionId is set to NULL, and consumer
> sends an acknowledgement to ActiveMQ via method
> ActiveMQConnection.asyncSendPacket.
> * While asyncSendPacket() is executing, new message can be dispatched to
> consumer
> * Because transactionId = null method TransactionContext.begin() produces
> new TransactionId in TransactionContext and set synchronizations list to NULL
> (but we remember that asyncSendPacket is still executing)
> * After asyncSendPacket() method finished afterCommit() method does nothing,
> because synchronizations = NULL, that means deliveredMessages collection with
> commited message of ActiveMQMessageConsumer can not be cleared
> * when Consumer sends the next acknowledgement, it sends acknowledgement for
> this message again, but ActimeMQ removed it before.
> I wrote a test reproducing the error. Real case is more complex in production
> and now we can't use batch transactions with SJMS.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
