Uploaded image for project: 'Camel'
  1. Camel
  2. CAMEL-10873

camel-sjms transacted routes dead-lock when exceptions are thrown by asynchronous processors

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Resolved
    • Priority: Major
    • Resolution: Fixed
    • Affects Version/s: 2.18.2, 2.19.0
    • Fix Version/s: 2.18.3, 2.19.0
    • Component/s: camel-sjms
    • Labels:
      None
    • Estimated Complexity:
      Unknown

      Description

      SJMS Consumers attempt rollback of the JMS Session if an exception is thrown along the Camel route. The rollback is attempted where the exception occurs and not on the original synchronous pipeline.

      For example with ActiveMQ and the thread component (see attached unit test)

      AtomicInteger counter = new AtomicInteger();
      
      from(destination + "?acknowledgementMode=SESSION_TRANSACTED&transacted=true")
         .threads()
         .process(exchange -> {
            if (counter.incrementAndGet() < TRANSACTION_REDELIVERY_COUNT) {
                throw new IllegalArgumentException();
            }
          })
          .to("mock:async.exception")
      

      `Session.rollback()` must be called by the ActiveMQ Thread, but is attempted by the asynchronous Camel thread, resulting in a dead-lock:

      "Camel (camel-1) thread #0 - Threads" Id=26 BLOCKED on java.lang.Object@4b9df8a owned by "ActiveMQ Session Task-1" Id=22
          org.apache.activemq.ActiveMQMessageConsumer.rollback(ActiveMQMessageConsumer.java:1212)
          org.apache.activemq.ActiveMQMessageConsumer$5.afterRollback(ActiveMQMessageConsumer.java:1091)
          org.apache.activemq.TransactionContext.afterRollback(TransactionContext.java:162)
          org.apache.activemq.TransactionContext.rollback(TransactionContext.java:287)
          org.apache.activemq.ActiveMQSession.rollback(ActiveMQSession.java:603)
          org.apache.camel.component.sjms.tx.SessionTransactionSynchronization.onFailure(SessionTransactionSynchronization.java:54)
          org.apache.camel.util.UnitOfWorkHelper.doneSynchronizations(UnitOfWorkHelper.java:101)
          org.apache.camel.impl.DefaultUnitOfWork.done(DefaultUnitOfWork.java:234)
          org.apache.camel.util.UnitOfWorkHelper.doneUow(UnitOfWorkHelper.java:65)
          org.apache.camel.processor.CamelInternalProcessor$UnitOfWorkProcessorAdvice.after(CamelInternalProcessor.java:675)
          org.apache.camel.processor.CamelInternalProcessor$UnitOfWorkProcessorAdvice.after(CamelInternalProcessor.java:630)
          org.apache.camel.processor.CamelInternalProcessor$InternalCallback.done(CamelInternalProcessor.java:247)
          org.apache.camel.processor.CamelInternalProcessor$InternalCallback.done(CamelInternalProcessor.java:258)
          org.apache.camel.processor.RedeliveryErrorHandler$2.done(RedeliveryErrorHandler.java:553)
          org.apache.camel.processor.Pipeline$1.done(Pipeline.java:147)
          org.apache.camel.processor.ThreadsProcessor$ProcessCall.run(ThreadsProcessor.java:87)
          java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
          java.util.concurrent.FutureTask.run(FutureTask.java:266)
          java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          java.lang.Thread.run(Thread.java:745)
      
      "ActiveMQ Session Task-1" Id=22 WAITING on java.util.concurrent.CountDownLatch$Sync@6955cb39
          sun.misc.Unsafe.park(Native Method)
          java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
          java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:836)
          java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:997)
          java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1304)
          java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
          org.apache.camel.impl.DefaultAsyncProcessorAwaitManager.await(DefaultAsyncProcessorAwaitManager.java:75)
          org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:105)
          org.apache.camel.component.sjms.consumer.InOnlyMessageHandler.handleMessage(InOnlyMessageHandler.java:65)
          org.apache.camel.component.sjms.consumer.AbstractMessageHandler.onMessage(AbstractMessageHandler.java:89)
          org.apache.activemq.ActiveMQMessageConsumer.dispatch(ActiveMQMessageConsumer.java:1401)
          org.apache.activemq.ActiveMQSessionExecutor.dispatch(ActiveMQSessionExecutor.java:131)
          org.apache.activemq.ActiveMQSessionExecutor.iterate(ActiveMQSessionExecutor.java:202)
          org.apache.activemq.thread.PooledTaskRunner.runTask(PooledTaskRunner.java:133)
          org.apache.activemq.thread.PooledTaskRunner$1.run(PooledTaskRunner.java:48)
          java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
          java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
          java.lang.Thread.run(Thread.java:745)
      

        Attachments

        1. TransactedAsyncExceptionTest.java
          4 kB
          Daniele Fognini

          Activity

            People

            • Assignee:
              davsclaus Claus Ibsen
              Reporter:
              fogninid Daniele Fognini
            • Votes:
              0 Vote for this issue
              Watchers:
              2 Start watching this issue

              Dates

              • Created:
                Updated:
                Resolved: