This is an automated email from the ASF dual-hosted git repository. gnodet pushed a commit to branch backport/27637-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 237cade7ee61ad1f3c769c88983662f964347e97 Author: Guillaume Nodet <[email protected]> AuthorDate: Sat Oct 10 07:42:34 2026 +0000 [backport camel-4.22.x] CAMEL-25504: camel-jta, camel-spring - Do not roll back transactions that complete during a graceful shutdown --- components/camel-jta/pom.xml | 6 + .../apache/camel/jta/TransactionErrorHandler.java | 7 +- ...ansactionErrorHandlerGracePeriodShutdownIT.java | 4 +- ...ransactionErrorHandlerGracefulShutdownTest.java | 187 +++++++++++++++++++++ ...tionalClientDataSourceGracefulShutdownTest.java | 166 ++++++++++++++++++ .../camel/spring/spi/TransactionErrorHandler.java | 19 ++- .../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 7 + 7 files changed, 392 insertions(+), 4 deletions(-) diff --git a/components/camel-jta/pom.xml b/components/camel-jta/pom.xml index 36493b401006..eccbbbf2ce4a 100644 --- a/components/camel-jta/pom.xml +++ b/components/camel-jta/pom.xml @@ -116,5 +116,11 @@ <artifactId>junit-jupiter</artifactId> <scope>test</scope> </dependency> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <version>${awaitility-version}</version> + <scope>test</scope> + </dependency> </dependencies> </project> diff --git a/components/camel-jta/src/main/java/org/apache/camel/jta/TransactionErrorHandler.java b/components/camel-jta/src/main/java/org/apache/camel/jta/TransactionErrorHandler.java index 99acf2ad241d..30a928a6e58d 100644 --- a/components/camel-jta/src/main/java/org/apache/camel/jta/TransactionErrorHandler.java +++ b/components/camel-jta/src/main/java/org/apache/camel/jta/TransactionErrorHandler.java @@ -57,6 +57,7 @@ public class TransactionErrorHandler extends ErrorHandlerSupport protected final CamelContext camelContext; protected final Processor output; protected volatile boolean preparingShutdown; + protected volatile boolean forcedShutdown; private JtaTransactionPolicy transactionPolicy; private final String transactionKey; private final LoggingLevel rollbackLoggingLevel; @@ -202,7 +203,8 @@ public class TransactionErrorHandler extends ErrorHandlerSupport processByErrorHandler(exchange); // if forced shutdown is in progress, mark the exchange for rollback - if (preparingShutdown) { + // (a graceful shutdown lets the in-flight exchange complete and commit) + if (forcedShutdown) { LOG.debug("Forced shutdown in progress, marking exchange for rollback: {}", exchange.getExchangeId()); exchange.setRollbackOnly(true); @@ -319,6 +321,7 @@ public class TransactionErrorHandler extends ErrorHandlerSupport protected void doStart() throws Exception { ServiceHelper.startService(output); preparingShutdown = false; + forcedShutdown = false; } @Override @@ -326,6 +329,7 @@ public class TransactionErrorHandler extends ErrorHandlerSupport super.doResume(); // reset flag when resuming preparingShutdown = false; + forcedShutdown = false; } @Override @@ -369,6 +373,7 @@ public class TransactionErrorHandler extends ErrorHandlerSupport LOG.trace("Prepare shutdown on error handler {}", this); preparingShutdown = true; if (forced) { + forcedShutdown = true; // mark all in-flight transacted exchanges for rollback so the transaction // is rolled back before the connection pool is destroyed during shutdown for (Exchange exchange : inflightTransactedExchanges) { diff --git a/components/camel-jta/src/test/java/org/apache/camel/jta/TransactionErrorHandlerGracePeriodShutdownIT.java b/components/camel-jta/src/test/java/org/apache/camel/jta/TransactionErrorHandlerGracePeriodShutdownIT.java index bc33769b69bf..741ee7c70052 100644 --- a/components/camel-jta/src/test/java/org/apache/camel/jta/TransactionErrorHandlerGracePeriodShutdownIT.java +++ b/components/camel-jta/src/test/java/org/apache/camel/jta/TransactionErrorHandlerGracePeriodShutdownIT.java @@ -194,7 +194,7 @@ public class TransactionErrorHandlerGracePeriodShutdownIT { * <li>TransactionErrorHandler.prepareShutdown(forced=true) marks exchange rollbackOnly</li> * <li>context.stop() returns (timeout expired, in-flight exchange still running)</li> * <li>Delay released, exchange completes normally (no exception)</li> - * <li>TransactionErrorHandler checks preparingShutdown -> sets rollbackOnly -> throws</li> + * <li>TransactionErrorHandler checks forcedShutdown -> sets rollbackOnly -> throws</li> * <li>JtaTransactionPolicy.run() catches -> Narayana rolls back the JTA transaction</li> * <li>Assert: the INSERT is NOT in the database (rolled back)</li> * </ol> @@ -235,7 +235,7 @@ public class TransactionErrorHandlerGracePeriodShutdownIT { // Now release the delay. The exchange wakes up and finishes processing. // The route has no more steps, so processByErrorHandler() returns normally. - // Back in doInTransactionTemplate(), the preparingShutdown check (the fix) + // Back in doInTransactionTemplate(), the forcedShutdown check (the fix) // sets rollbackOnly, causing the transaction to be rolled back. releaseLatch.countDown(); diff --git a/components/camel-jta/src/test/java/org/apache/camel/jta/TransactionErrorHandlerGracefulShutdownTest.java b/components/camel-jta/src/test/java/org/apache/camel/jta/TransactionErrorHandlerGracefulShutdownTest.java new file mode 100644 index 000000000000..27650c93f52a --- /dev/null +++ b/components/camel-jta/src/test/java/org/apache/camel/jta/TransactionErrorHandlerGracefulShutdownTest.java @@ -0,0 +1,187 @@ +/* + * 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.jta; + +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.CamelContext; +import org.apache.camel.Exchange; +import org.apache.camel.LoggingLevel; +import org.apache.camel.Navigate; +import org.apache.camel.Processor; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.impl.engine.DefaultUnitOfWork; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A graceful shutdown lets in-flight exchanges complete: an exchange that finishes within the shutdown timeout must be + * committed. Only a forced shutdown (timeout expired) must mark in-flight transacted exchanges for rollback. + * <p> + * DefaultShutdownStrategy calls {@code prepareShutdown(false, false)} on the route services before it waits for the + * in-flight exchanges, and {@code prepareShutdown(false, true)} only once the timeout expired. + */ +class TransactionErrorHandlerGracefulShutdownTest { + + private CamelContext camelContext; + private final AtomicInteger commits = new AtomicInteger(); + private final AtomicInteger rollbacks = new AtomicInteger(); + private final CountDownLatch processingStarted = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + private final JtaTransactionPolicy policy = new JtaTransactionPolicy() { + @Override + public void run(Runnable runnable) throws Throwable { + try { + runnable.run(); + } catch (Throwable t) { + rollbacks.incrementAndGet(); + throw t; + } + commits.incrementAndGet(); + } + }; + + private final Processor blocking = exchange -> { + processingStarted.countDown(); + assertTrue(release.await(20, TimeUnit.SECONDS), "test did not release the exchange"); + }; + + @BeforeEach + void setUp() { + camelContext = new DefaultCamelContext(); + camelContext.getRegistry().bind("PROPAGATION_REQUIRED", policy); + // the in-flight exchange completes well within the graceful shutdown timeout + camelContext.getShutdownStrategy().setTimeout(20); + camelContext.getShutdownStrategy().setTimeUnit(TimeUnit.SECONDS); + } + + @AfterEach + void tearDown() { + release.countDown(); + if (camelContext != null) { + camelContext.stop(); + } + } + + @Test + void inflightExchangeCompletingDuringGracefulShutdownIsCommitted() throws Exception { + camelContext.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:start").routeId("transactedRoute") + .transacted() + .process(blocking); + } + }); + camelContext.start(); + + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future<Exchange> sent = executor.submit( + () -> camelContext.createProducerTemplate().send("direct:start", e -> e.getIn().setBody("order"))); + assertTrue(processingStarted.await(10, TimeUnit.SECONDS), "exchange should be in flight"); + + TransactionErrorHandler teh = findTransactionErrorHandler(camelContext.getRoute("transactedRoute").navigate()); + assertNotNull(teh, "transacted route should use the JTA TransactionErrorHandler"); + + // graceful shutdown: the strategy notifies the error handler, then waits for the in-flight exchange + Future<?> stopped = executor.submit(() -> camelContext.stop()); + await().atMost(10, TimeUnit.SECONDS).until(() -> teh.preparingShutdown); + + // the exchange finishes inside the grace period (no timeout, no forced shutdown) + release.countDown(); + Exchange done = sent.get(10, TimeUnit.SECONDS); + stopped.get(10, TimeUnit.SECONDS); + + assertFalse(done.isRollbackOnly(), + "an exchange that completed during a graceful shutdown must not be marked rollback only"); + assertEquals(0, rollbacks.get(), + "an exchange that completed during a graceful shutdown must not be rolled back"); + assertEquals(1, commits.get(), "an exchange that completed during a graceful shutdown must be committed"); + } finally { + executor.shutdownNow(); + } + } + + @Test + void gracefulPrepareShutdownDoesNotRollBack() throws Exception { + TransactionErrorHandler teh = new TransactionErrorHandler(camelContext, blocking, policy, LoggingLevel.WARN); + camelContext.start(); + teh.start(); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Exchange exchange = camelContext.getEndpoint("direct:test").createExchange(); + exchange.getExchangeExtension().setUnitOfWork(new DefaultUnitOfWork(exchange)); + Future<?> processed = executor.submit(() -> { + teh.process(exchange); + return null; + }); + assertTrue(processingStarted.await(10, TimeUnit.SECONDS), "exchange should be in flight"); + + // what DefaultShutdownStrategy does before it waits for in-flight exchanges + teh.prepareShutdown(false, false); + release.countDown(); + processed.get(10, TimeUnit.SECONDS); + + assertFalse(exchange.isRollbackOnly(), "graceful shutdown must not mark the exchange rollback only"); + assertEquals(1, commits.get(), "graceful shutdown must let the transaction commit"); + assertEquals(0, rollbacks.get(), "graceful shutdown must not roll back the transaction"); + } finally { + executor.shutdownNow(); + teh.stop(); + } + } + + private static TransactionErrorHandler findTransactionErrorHandler(Navigate<Processor> nav) { + if (nav == null || !nav.hasNext()) { + return null; + } + List<Processor> next = nav.next(); + if (next == null) { + return null; + } + for (Processor p : next) { + if (p instanceof TransactionErrorHandler teh) { + return teh; + } + if (p instanceof Navigate<?>) { + @SuppressWarnings("unchecked") + TransactionErrorHandler found = findTransactionErrorHandler((Navigate<Processor>) p); + if (found != null) { + return found; + } + } + } + return null; + } +} diff --git a/components/camel-spring-parent/camel-spring-xml/src/test/java/org/apache/camel/spring/interceptor/TransactionalClientDataSourceGracefulShutdownTest.java b/components/camel-spring-parent/camel-spring-xml/src/test/java/org/apache/camel/spring/interceptor/TransactionalClientDataSourceGracefulShutdownTest.java new file mode 100644 index 000000000000..4576f52ecc15 --- /dev/null +++ b/components/camel-spring-parent/camel-spring-xml/src/test/java/org/apache/camel/spring/interceptor/TransactionalClientDataSourceGracefulShutdownTest.java @@ -0,0 +1,166 @@ +/* + * 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.spring.interceptor; + +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.Route; +import org.apache.camel.Service; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.spi.ShutdownPrepared; +import org.apache.camel.spring.spi.TransactionErrorHandler; +import org.apache.camel.support.service.ServiceHelper; +import org.apache.camel.support.service.ServiceSupport; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * A graceful shutdown lets in-flight exchanges complete: a transacted exchange that finishes within the shutdown + * timeout must be committed. Only a forced shutdown (timeout expired) marks in-flight transacted exchanges for + * rollback, see {@link TransactionalClientDataSourceForcedShutdownTest}. + * <p> + * DefaultShutdownStrategy calls {@code prepareShutdown(false, false)} on the route services before it waits for the + * in-flight exchanges, and {@code prepareShutdown(false, true)} only once the timeout expired. + */ +class TransactionalClientDataSourceGracefulShutdownTest extends TransactionClientDataSourceSupport { + + private final CountDownLatch inFlight = new CountDownLatch(1); + private final CountDownLatch release = new CountDownLatch(1); + + @Test + void gracefulPrepareShutdownCommitsInFlightTransaction() throws Exception { + assertEquals(1, countBooks(), "Initial number of books"); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future<Exchange> sent = executor.submit(() -> template.send("direct:graceful", e -> e.getIn().setBody("x"))); + assertTrue(inFlight.await(10, TimeUnit.SECONDS), "exchange should be in flight"); + + TransactionErrorHandler teh = findTransactionErrorHandler(context.getRoute("gracefulRoute")); + assertNotNull(teh, "transacted route should use the Spring TransactionErrorHandler"); + + // what DefaultShutdownStrategy does before it waits for the in-flight exchanges + teh.prepareShutdown(false, false); + release.countDown(); + Exchange done = sent.get(10, TimeUnit.SECONDS); + + assertNull(done.getException()); + assertFalse(done.isRollbackOnly(), "graceful shutdown must not mark the exchange rollback only"); + assertEquals(2, countBooks(), "graceful shutdown must let the transaction commit"); + } finally { + release.countDown(); + executor.shutdownNow(); + } + } + + @Test + void inFlightTransactionCompletingDuringContextStopIsCommitted() throws Exception { + // the in-flight exchange completes well within the graceful shutdown timeout + context.getShutdownStrategy().setTimeout(30); + assertEquals(1, countBooks(), "Initial number of books"); + + ExecutorService executor = Executors.newSingleThreadExecutor(); + try { + Future<Exchange> sent = executor.submit(() -> template.send("direct:graceful", e -> e.getIn().setBody("x"))); + assertTrue(inFlight.await(10, TimeUnit.SECONDS), "exchange should be in flight"); + + // the shutdown strategy releases the exchange from its graceful prepareShutdown call (see + // ReleaseOnPrepareShutdown), then waits for it to complete + context.stop(); + Exchange done = sent.get(10, TimeUnit.SECONDS); + + assertFalse(context.getShutdownStrategy().isTimeoutOccurred(), "shutdown should not have timed out"); + assertNull(done.getException()); + assertFalse(done.isRollbackOnly(), + "an exchange that completed during a graceful shutdown must not be marked rollback only"); + assertEquals(2, countBooks(), "an exchange that completed during a graceful shutdown must be committed"); + } finally { + release.countDown(); + executor.shutdownNow(); + } + } + + private int countBooks() { + return jdbc.queryForObject("select count(*) from books", Integer.class); + } + + private static TransactionErrorHandler findTransactionErrorHandler(Route route) { + if (route.getProcessor() instanceof Service service) { + Set<Service> children = ServiceHelper.getChildServices(service, true); + for (Service child : children) { + if (child instanceof TransactionErrorHandler teh) { + return teh; + } + } + } + return null; + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:graceful").routeId("gracefulRoute") + .transacted() + .setBody(constant("Tiger in Action")).bean("bookService") + .process(exchange -> { + inFlight.countDown(); + assertTrue(release.await(10, TimeUnit.SECONDS), "test did not release the exchange"); + }) + .process(new ReleaseOnPrepareShutdown(release)); + } + }; + } + + /** + * Releases the in-flight exchange when the shutdown strategy prepares the route for a graceful shutdown. It is + * inside the transacted block, so the strategy prepares the TransactionErrorHandler before this service. + */ + private static final class ReleaseOnPrepareShutdown extends ServiceSupport implements Processor, ShutdownPrepared { + + private final CountDownLatch release; + + private ReleaseOnPrepareShutdown(CountDownLatch release) { + this.release = release; + } + + @Override + public void prepareShutdown(boolean suspendOnly, boolean forced) { + if (!suspendOnly && !forced) { + release.countDown(); + } + } + + @Override + public void process(Exchange exchange) { + // noop + } + } +} diff --git a/components/camel-spring-parent/camel-spring/src/main/java/org/apache/camel/spring/spi/TransactionErrorHandler.java b/components/camel-spring-parent/camel-spring/src/main/java/org/apache/camel/spring/spi/TransactionErrorHandler.java index 87986c098365..f0b9d1ac8f86 100644 --- a/components/camel-spring-parent/camel-spring/src/main/java/org/apache/camel/spring/spi/TransactionErrorHandler.java +++ b/components/camel-spring-parent/camel-spring/src/main/java/org/apache/camel/spring/spi/TransactionErrorHandler.java @@ -52,6 +52,7 @@ public class TransactionErrorHandler extends RedeliveryErrorHandler { private final String transactionKey; private final LoggingLevel rollbackLoggingLevel; private final Set<Exchange> inflightTransactedExchanges = ConcurrentHashMap.newKeySet(); + private volatile boolean forcedShutdown; /** * Creates the transaction error handler. @@ -212,7 +213,8 @@ public class TransactionErrorHandler extends RedeliveryErrorHandler { processByErrorHandler(exchange); // if forced shutdown is in progress, mark the exchange for rollback - if (preparingShutdown) { + // (a graceful shutdown lets the in-flight exchange complete and commit) + if (forcedShutdown) { LOG.debug("Forced shutdown in progress, marking exchange for rollback: {}", exchange.getExchangeId()); exchange.setRollbackOnly(true); @@ -338,6 +340,20 @@ public class TransactionErrorHandler extends RedeliveryErrorHandler { } } + @Override + protected void doStart() throws Exception { + super.doStart(); + // reset flag when starting + forcedShutdown = false; + } + + @Override + protected void doResume() throws Exception { + super.doResume(); + // reset flag when resuming + forcedShutdown = false; + } + @Override public void prepareShutdown(boolean suspendOnly, boolean forced) { if (suspendOnly) { @@ -345,6 +361,7 @@ public class TransactionErrorHandler extends RedeliveryErrorHandler { } super.prepareShutdown(suspendOnly, forced); if (forced) { + forcedShutdown = true; // mark all in-flight transacted exchanges for rollback so the transaction // is rolled back before the connection pool is destroyed during shutdown for (Exchange exchange : inflightTransactedExchanges) { diff --git a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc index 5cfe518c90c8..096158878345 100644 --- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc +++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_22.adoc @@ -2019,3 +2019,10 @@ so a downstream reader that expected that type received an incompatible object. A route that reads the full embeddings or audio response from `CamelOpenAIResponse` must switch to the matching property. The chat-completion and responses operations are unchanged. + +=== camel-jta, camel-spring - a transacted exchange that completes during a graceful shutdown is committed + +Since Camel 4.19, a transacted exchange (JTA or Spring transacted routes) that was in flight when a graceful shutdown +(stopping `CamelContext` or a route) started was rolled back even when it completed within the shutdown timeout. It is +now committed again. Only the forced shutdown after the timeout marks the in-flight transacted exchanges for rollback; +with `shutdownNowOnTimeout=false` there is no forced shutdown, so an exchange completing after the timeout is committed.
