This is an automated email from the ASF dual-hosted git repository. jamesnetherton pushed a commit to branch 3.33.x in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
commit 1b69958733a4275a26c5b7d1c0548551dc5b3114 Author: Andrea Cosentino <[email protected]> AuthorDate: Tue Sep 1 15:40:49 2026 +0200 Fixes #9056. Stop swallowing JTA rollback and resume failures * Fixes #9056. Stop swallowing JTA rollback and resume failures TransactionalJtaTransactionPolicy, the base class of all six PROPAGATION_* policies, logged a warning and carried on when rollback(), setRollbackOnly() or resume() failed. A route that handled the original exception could then continue as though the transaction had been marked for rollback when it had not, and work after a failed resume ran outside the transaction the policy was expected to restore. Raise those failures instead. Neither call site lets one replace the exception that made the rollback necessary: runWithTransaction and commit attach a rollback failure to the in-flight exception as a suppressed exception, and the two policies that suspend pass their in-flight exception into resumeTransaction so a failure in the finally block is attached rather than masking it. No signatures change. resumeTransaction(Transaction) keeps its shape and wraps a failure in a RuntimeCamelException; a resumeTransaction(Transaction, Throwable) overload carries the in-flight exception for the policies that need it. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> * Fixes #9056. Use the non-deprecated QuarkusExtensionTest QuarkusUnitTest is deprecated in favour of QuarkusExtensionTest. Both extend AbstractQuarkusExtensionTest and setArchiveProducer lives on that shared base, so this is a straight swap. Co-Authored-By: Claude Opus 5 (1M context) <[email protected]> --------- Co-authored-by: Claude Opus 5 (1M context) <[email protected]> (cherry picked from commit ae0256a9cc1bda67c8ada0513594b2292aa90aac) --- .../modules/ROOT/pages/migration-guide/3.40.0.adoc | 15 +++ .../jta/JtaTransactionFailurePropagationTest.java | 137 +++++++++++++++++++++ .../jta/NotSupportedJtaTransactionPolicy.java | 7 +- .../jta/RequiresNewJtaTransactionPolicy.java | 7 +- .../jta/TransactionalJtaTransactionPolicy.java | 56 +++++++-- 5 files changed, 212 insertions(+), 10 deletions(-) diff --git a/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc b/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc index 69cb3e0074..630aa2255e 100644 --- a/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc +++ b/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc @@ -18,3 +18,18 @@ To keep the previous behaviour for a context that relied on the default, set it ---- quarkus.camel.ldap.dir-contexts."my-context".security-authentication=none ---- + +== JTA extension changes + +=== Rollback and resume failures are no longer swallowed + +`TransactionalJtaTransactionPolicy`, the base class of all six `PROPAGATION_*` policies, logged a warning and carried on when `rollback()`, `setRollbackOnly()` or `resume()` failed. A route that handled the original exception could therefore continue as though the transaction had been marked for rollback when it had not, and work after a failed resume ran outside the transaction the policy was expected to restore. + +These failures are now raised: + +* A failure to roll back, or to mark an outer transaction for rollback, is attached to the exception that made the rollback necessary as a suppressed exception. That exception still propagates unchanged, so the failure is visible without hiding its cause. +* A failure to resume a suspended transaction, in `PROPAGATION_REQUIRES_NEW` and `PROPAGATION_NOT_SUPPORTED`, is attached to the in-flight failure in the same way, or raised on its own when the body succeeded. + +An application that inspected only the top-level exception sees no difference. One that handles exceptions and continues may now see a failure it previously never learned about. Check `Throwable.getSuppressed()` when diagnosing. + +No method signatures changed. `resumeTransaction(Transaction)` is unchanged and now wraps a resume failure in a `RuntimeCamelException`; a new `resumeTransaction(Transaction, Throwable)` overload is what the policies use so that a resume failure cannot replace a failure already on its way out. diff --git a/extensions/jta/deployment/src/test/java/org/apache/camel/quarkus/component/jta/JtaTransactionFailurePropagationTest.java b/extensions/jta/deployment/src/test/java/org/apache/camel/quarkus/component/jta/JtaTransactionFailurePropagationTest.java new file mode 100644 index 0000000000..5d22fa3d00 --- /dev/null +++ b/extensions/jta/deployment/src/test/java/org/apache/camel/quarkus/component/jta/JtaTransactionFailurePropagationTest.java @@ -0,0 +1,137 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.jta; + +import java.util.Arrays; + +import io.quarkus.test.QuarkusExtensionTest; +import jakarta.inject.Inject; +import jakarta.inject.Named; +import jakarta.transaction.Status; +import jakarta.transaction.SystemException; +import jakarta.transaction.Transaction; +import jakarta.transaction.TransactionManager; +import org.jboss.shrinkwrap.api.ShrinkWrap; +import org.jboss.shrinkwrap.api.spec.JavaArchive; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.when; + +/** + * A transaction whose rollback, rollback marking or resumption fails must not let that failure disappear, and must not + * let it replace the exception that made the rollback necessary. + */ +public class JtaTransactionFailurePropagationTest { + + @RegisterExtension + static final QuarkusExtensionTest CONFIG = new QuarkusExtensionTest() + .setArchiveProducer(() -> ShrinkWrap.create(JavaArchive.class) + .addClasses(MockTransactionManagerProducer.class, MockTransaction.class)); + + @Inject + TransactionManager transactionManager; + + @Inject + @Named("PROPAGATION_REQUIRED") + RequiredJtaTransactionPolicy requiredPolicy; + + @Inject + @Named("PROPAGATION_REQUIRES_NEW") + RequiresNewJtaTransactionPolicy requiresNewPolicy; + + @AfterEach + public void afterEach() { + reset(transactionManager); + } + + @Test + public void failedRollbackMarkingIsReportedAndKeepsTheOriginalCause() throws Exception { + // Participating in an outer transaction, so the policy marks it rather than rolling it back + when(transactionManager.getStatus()).thenReturn(Status.STATUS_ACTIVE); + doThrow(new SystemException("mark failed")).when(transactionManager).setRollbackOnly(); + + Exception routeFailure = new Exception("route failed"); + Throwable thrown = assertThrows(Throwable.class, () -> requiredPolicy.run(() -> { + throw routeFailure; + })); + + // The original failure still surfaces + assertEquals(routeFailure, thrown); + // and the marking failure rides along rather than vanishing into a log line + assertTrue(Arrays.stream(thrown.getSuppressed()) + .anyMatch(s -> s.getMessage().contains("Unable to mark the transaction for rollback")), + "expected the failed setRollbackOnly to be attached, got " + + Arrays.toString(thrown.getSuppressed())); + } + + @Test + public void failedRollbackIsReportedAndKeepsTheOriginalCause() throws Exception { + when(transactionManager.getStatus()).thenReturn(Status.STATUS_NO_TRANSACTION); + doThrow(new SystemException("rollback failed")).when(transactionManager).rollback(); + + Exception routeFailure = new Exception("route failed"); + Throwable thrown = assertThrows(Throwable.class, () -> requiredPolicy.run(() -> { + throw routeFailure; + })); + + assertEquals(routeFailure, thrown); + assertTrue(Arrays.stream(thrown.getSuppressed()) + .anyMatch(s -> s.getMessage().contains("Unable to rollback transaction")), + "expected the failed rollback to be attached, got " + Arrays.toString(thrown.getSuppressed())); + } + + @Test + public void failedResumeIsReportedWhenTheBodySucceeded() throws Exception { + Transaction suspended = new MockTransaction(); + when(transactionManager.getStatus()).thenReturn(Status.STATUS_NO_TRANSACTION); + when(transactionManager.suspend()).thenReturn(suspended); + doThrow(new SystemException("resume failed")).when(transactionManager).resume(suspended); + + // Nothing else is in flight, so the resume failure is the failure + Throwable thrown = assertThrows(Throwable.class, () -> requiresNewPolicy.run(() -> { + })); + + assertTrue(thrown.getMessage().contains("Unable to resume transaction"), + "expected the resume failure to surface, got " + thrown); + } + + @Test + public void failedResumeDoesNotReplaceTheBodyFailure() throws Exception { + Transaction suspended = new MockTransaction(); + when(transactionManager.getStatus()).thenReturn(Status.STATUS_NO_TRANSACTION); + when(transactionManager.suspend()).thenReturn(suspended); + doThrow(new SystemException("resume failed")).when(transactionManager).resume(suspended); + + Exception routeFailure = new Exception("route failed"); + Throwable thrown = assertThrows(Throwable.class, () -> requiresNewPolicy.run(() -> { + throw routeFailure; + })); + + // The body's failure wins; the resume failure is attached to it + assertEquals(routeFailure, thrown); + assertTrue(Arrays.stream(thrown.getSuppressed()) + .anyMatch(s -> s.getMessage().contains("Unable to resume transaction")), + "expected the failed resume to be attached, got " + Arrays.toString(thrown.getSuppressed())); + } +} diff --git a/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/NotSupportedJtaTransactionPolicy.java b/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/NotSupportedJtaTransactionPolicy.java index 7dec059d92..02d6a5f910 100644 --- a/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/NotSupportedJtaTransactionPolicy.java +++ b/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/NotSupportedJtaTransactionPolicy.java @@ -25,11 +25,16 @@ public final class NotSupportedJtaTransactionPolicy extends TransactionalJtaTran @Override public void run(final Runnable runnable) throws Throwable { Transaction suspendedTransaction = null; + Throwable primary = null; try { suspendedTransaction = suspendTransaction(); runnable.run(); + } catch (Throwable e) { + primary = e; + throw e; } finally { - resumeTransaction(suspendedTransaction); + // Passing the failure already on its way out keeps a resume failure from replacing it + resumeTransaction(suspendedTransaction, primary); } } } diff --git a/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/RequiresNewJtaTransactionPolicy.java b/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/RequiresNewJtaTransactionPolicy.java index 60156dc221..2bc769f0f9 100644 --- a/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/RequiresNewJtaTransactionPolicy.java +++ b/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/RequiresNewJtaTransactionPolicy.java @@ -25,11 +25,16 @@ public final class RequiresNewJtaTransactionPolicy extends TransactionalJtaTrans @Override public void run(final Runnable runnable) throws Throwable { Transaction suspendedTransaction = null; + Throwable primary = null; try { suspendedTransaction = suspendTransaction(); runWithTransaction(runnable, true); + } catch (Throwable e) { + primary = e; + throw e; } finally { - resumeTransaction(suspendedTransaction); + // Passing the failure already on its way out keeps a resume failure from replacing it + resumeTransaction(suspendedTransaction, primary); } } } diff --git a/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/TransactionalJtaTransactionPolicy.java b/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/TransactionalJtaTransactionPolicy.java index 00a2fc2e00..6bd69b9b94 100644 --- a/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/TransactionalJtaTransactionPolicy.java +++ b/extensions/jta/runtime/src/main/java/org/apache/camel/quarkus/component/jta/TransactionalJtaTransactionPolicy.java @@ -25,17 +25,14 @@ import jakarta.transaction.SystemException; import jakarta.transaction.Transaction; import jakarta.transaction.TransactionManager; import org.apache.camel.CamelException; +import org.apache.camel.RuntimeCamelException; import org.apache.camel.jta.JtaTransactionPolicy; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; /** * Helper methods for transaction handling */ public abstract class TransactionalJtaTransactionPolicy extends JtaTransactionPolicy { - private static final Logger LOG = LoggerFactory.getLogger(TransactionalJtaTransactionPolicy.class); - @Inject TransactionManager transactionManager; @@ -46,7 +43,7 @@ public abstract class TransactionalJtaTransactionPolicy extends JtaTransactionPo try { runnable.run(); } catch (Throwable e) { - rollback(isNew); + rollbackSuppressing(e, isNew); throw e; } if (isNew) { @@ -64,11 +61,18 @@ public abstract class TransactionalJtaTransactionPolicy extends JtaTransactionPo } catch (HeuristicMixedException | HeuristicRollbackException | RollbackException | SystemException e) { throw new CamelException("Unable to commit transaction", e); } catch (Exception | Error e) { - rollback(true); + rollbackSuppressing(e, true); throw e; } } + /** + * Rolls the transaction back, or marks it for rollback when it belongs to an outer policy. + * + * A failure here is raised rather than logged and discarded. Callers that already have an exception on its way + * out attach this one to it through {@link #rollbackSuppressing(Throwable, boolean)}, so the original cause is + * never replaced. + */ final protected void rollback(boolean isNew) throws Exception { try { if (isNew) { @@ -77,7 +81,20 @@ public abstract class TransactionalJtaTransactionPolicy extends JtaTransactionPo transactionManager.setRollbackOnly(); } } catch (Throwable e) { - LOG.warn("Could not rollback transaction!", e); + throw new CamelException( + isNew ? "Unable to rollback transaction" : "Unable to mark the transaction for rollback", e); + } + } + + /** + * Rolls back while an exception is already on its way out, attaching a rollback failure to it as a suppressed + * exception. Replacing the original would hide why the rollback was needed in the first place. + */ + private void rollbackSuppressing(Throwable primary, boolean isNew) { + try { + rollback(isNew); + } catch (Throwable rollbackFailure) { + primary.addSuppressed(rollbackFailure); } } @@ -85,7 +102,25 @@ public abstract class TransactionalJtaTransactionPolicy extends JtaTransactionPo return transactionManager.suspend(); } + /** + * Resumes the suspended transaction, raising a failure rather than logging and discarding it, so that work after + * this point does not silently continue outside the transaction the caller expects to be restored. + */ final protected void resumeTransaction(final Transaction suspendedTransaction) { + try { + resumeTransaction(suspendedTransaction, null); + } catch (Exception e) { + throw RuntimeCamelException.wrapRuntimeCamelException(e); + } + } + + /** + * Resumes the suspended transaction while {@code primary} may already be on its way out, which is the case in the + * `finally` block of a policy that suspends. A resume failure is attached to {@code primary} when there is one and + * raised on its own otherwise, so it is neither lost nor allowed to replace the original failure. + */ + final protected void resumeTransaction(final Transaction suspendedTransaction, final Throwable primary) + throws Exception { if (suspendedTransaction == null) { return; } @@ -93,7 +128,12 @@ public abstract class TransactionalJtaTransactionPolicy extends JtaTransactionPo try { transactionManager.resume(suspendedTransaction); } catch (Throwable e) { - LOG.warn("Could not resume transaction!", e); + CamelException failure = new CamelException("Unable to resume transaction", e); + if (primary != null) { + primary.addSuppressed(failure); + } else { + throw failure; + } } }
