This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26850 in repository https://gitbox.apache.org/repos/asf/camel.git
commit 424050afd2273294afc1242da7e5f960e8307241 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Nov 11 10:50:07 2026 +0100 CAMEL-26850: camel-jbang - camel run: a route with transacted processes no messages camel run binds a placeholder transaction policy for transacted, so that a route that uses it can start without a transaction manager. The placeholder's wrap() returned null: the route then had no processor, and every message completed without any step running and without an error. The placeholder was also bound over a policy of the user's own under the same name. The placeholder now runs the steps without a transaction, and camel run logs a WARN for the route. It is bound only when the transacted has no policy or reference of its own, and the registry has no transaction policy and no Spring PlatformTransactionManager, of which Camel makes the policy. That class is loaded with the application class loader, so camel run does not download camel-spring for the check. Co-Authored-By: Claude Opus 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_016Y4dDi5iQNgtHrDdz2NpwF --- .../camel/main/download/TransactedDownloader.java | 66 +++++++++++-- .../main/download/TransactedDownloaderTest.java | 110 +++++++++++++++++++++ 2 files changed, 168 insertions(+), 8 deletions(-) diff --git a/dsl/camel-kamelet-main/src/main/java/org/apache/camel/main/download/TransactedDownloader.java b/dsl/camel-kamelet-main/src/main/java/org/apache/camel/main/download/TransactedDownloader.java index 9f8f14474dfc..a455a36c1359 100644 --- a/dsl/camel-kamelet-main/src/main/java/org/apache/camel/main/download/TransactedDownloader.java +++ b/dsl/camel-kamelet-main/src/main/java/org/apache/camel/main/download/TransactedDownloader.java @@ -16,8 +16,7 @@ */ package org.apache.camel.main.download; -import java.util.Arrays; - +import org.apache.camel.CamelContext; import org.apache.camel.NamedNode; import org.apache.camel.Processor; import org.apache.camel.Route; @@ -26,12 +25,17 @@ import org.apache.camel.reifier.ProcessReifier; import org.apache.camel.reifier.ProcessorReifier; import org.apache.camel.spi.Registry; import org.apache.camel.spi.TransactedPolicy; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * When using transacted then we need to download the runtime implementation */ public class TransactedDownloader { + private static final Logger LOG = LoggerFactory.getLogger(TransactedDownloader.class); + private static final String[] TRANSACTED_POLICIES = new String[] { "PROPAGATION_REQUIRED", "PROPAGATION_REQUIRES_NEW", @@ -50,21 +54,67 @@ public class TransactedDownloader { // KameletMain that registers them, which would otherwise stay referenced when it is done ProcessorReifier.registerReifier(TransactedDefinition.class, (route, processorDefinition) -> { - if (processorDefinition instanceof TransactedDefinition) { + if (processorDefinition instanceof TransactedDefinition transacted) { DependencyDownloader downloader = route.getCamelContext().hasService(DependencyDownloader.class); if (downloader != null) { downloader.downloadDependency("org.apache.camel", "camel-jta", route.getCamelContext().getVersion()); - TransactedPolicy policy = new DummyTransactedPolicy(); - Registry registry = route.getCamelContext().getRegistry(); - Arrays.stream(TRANSACTED_POLICIES).forEach(p -> registry.bind(p, policy)); + if (bindPassThroughPolicies(route.getCamelContext(), transacted)) { + LOG.warn("Route {} uses transacted, but there is no transaction manager: the route runs" + + " without a transaction. Declare a transaction manager bean, or export to" + + " Spring Boot or Quarkus, for transactions.", + route.getRouteId()); + } } } return ProcessReifier.coreReifier(route, processorDefinition); }); } - private static class DummyTransactedPolicy implements TransactedPolicy { + /** + * Binds a policy that runs the steps without a transaction under the PROPAGATION_* names that have no bean, when + * the transacted has no policy or reference of its own and the registry has no transaction manager. + * + * @return whether the steps run without a transaction + */ + static boolean bindPassThroughPolicies(CamelContext context, TransactedDefinition transacted) { + if (transacted.getPolicy() != null || transacted.getRef() != null || hasTransactionManager(context)) { + return false; + } + Registry registry = context.getRegistry(); + TransactedPolicy policy = new PassThroughTransactedPolicy(); + for (String name : TRANSACTED_POLICIES) { + // a policy of the user's own under that name is used as it is + if (registry.lookupByName(name) == null) { + registry.bind(name, policy); + } + } + return true; + } + + /** + * Whether the registry has a transacted policy of the user's own, or a Spring PlatformTransactionManager, of which + * Camel makes the policy. + */ + private static boolean hasTransactionManager(CamelContext context) { + Registry registry = context.getRegistry(); + for (TransactedPolicy policy : registry.findByType(TransactedPolicy.class)) { + if (!(policy instanceof PassThroughTransactedPolicy)) { + return true; + } + } + // not with the class resolver of camel run, which downloads camel-spring to load the class (and camel export + // would then add it): a transaction manager bean of the user's own has loaded the class already + Class<?> tm = ObjectHelper.loadClass("org.springframework.transaction.PlatformTransactionManager", + context.getApplicationContextClassLoader()); + return tm != null && !registry.findByType(tm).isEmpty(); + } + + /** + * Runs the steps without a transaction. CAMEL-26850: it returned null before, and a route with transacted then had + * no processor, so every message was gone without a trace. + */ + private static final class PassThroughTransactedPolicy implements TransactedPolicy { @Override public void beforeWrap(final Route route, final NamedNode definition) { @@ -72,7 +122,7 @@ public class TransactedDownloader { @Override public Processor wrap(final Route route, final Processor processor) { - return null; + return processor; } } } diff --git a/dsl/camel-kamelet-main/src/test/java/org/apache/camel/main/download/TransactedDownloaderTest.java b/dsl/camel-kamelet-main/src/test/java/org/apache/camel/main/download/TransactedDownloaderTest.java new file mode 100644 index 000000000000..bd4e3e007516 --- /dev/null +++ b/dsl/camel-kamelet-main/src/test/java/org/apache/camel/main/download/TransactedDownloaderTest.java @@ -0,0 +1,110 @@ +/* + * 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.main.download; + +import org.apache.camel.CamelContext; +import org.apache.camel.NamedNode; +import org.apache.camel.Processor; +import org.apache.camel.Route; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.model.TransactedDefinition; +import org.apache.camel.spi.TransactedPolicy; +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.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * CAMEL-26850: camel run has no transaction manager unless one is declared, so a route with transacted runs its steps + * without a transaction. The policy bound for it before returned null, and the route dropped every message. + */ +public class TransactedDownloaderTest { + + @Test + public void aTransactedRouteWithoutATransactionManagerRunsItsSteps() throws Exception { + try (CamelContext context = new DefaultCamelContext()) { + assertTrue(TransactedDownloader.bindPassThroughPolicies(context, new TransactedDefinition())); + context.addRoutes(orders()); + context.start(); + + MockEndpoint mock = context.getEndpoint("mock:stored", MockEndpoint.class); + mock.expectedBodiesReceived("ORD-1001"); + context.createProducerTemplate().sendBody("direct:orders", "ORD-1001"); + mock.assertIsSatisfied(); + } + } + + @Test + public void aPolicyOfTheUsersOwnIsKept() throws Exception { + try (CamelContext context = new DefaultCamelContext()) { + CountingPolicy own = new CountingPolicy(); + context.getRegistry().bind("PROPAGATION_REQUIRED", own); + assertFalse(TransactedDownloader.bindPassThroughPolicies(context, new TransactedDefinition())); + assertSame(own, context.getRegistry().lookupByName("PROPAGATION_REQUIRED")); + assertNull(context.getRegistry().lookupByName("PROPAGATION_REQUIRES_NEW")); + context.addRoutes(orders()); + context.start(); + + MockEndpoint mock = context.getEndpoint("mock:stored", MockEndpoint.class); + mock.expectedBodiesReceived("ORD-1001"); + context.createProducerTemplate().sendBody("direct:orders", "ORD-1001"); + mock.assertIsSatisfied(); + assertEquals(1, own.wrapped); + } + } + + @Test + public void aTransactedWithAReferenceIsLeftAsItIs() throws Exception { + try (CamelContext context = new DefaultCamelContext()) { + TransactedDefinition transacted = new TransactedDefinition(); + transacted.setRef("myPolicy"); + assertFalse(TransactedDownloader.bindPassThroughPolicies(context, transacted)); + assertNull(context.getRegistry().lookupByName("PROPAGATION_REQUIRED")); + } + } + + private static RouteBuilder orders() { + return new RouteBuilder() { + @Override + public void configure() { + from("direct:orders").routeId("orders") + .transacted() + .to("mock:stored"); + } + }; + } + + private static final class CountingPolicy implements TransactedPolicy { + + private int wrapped; + + @Override + public void beforeWrap(Route route, NamedNode definition) { + } + + @Override + public Processor wrap(Route route, Processor processor) { + wrapped++; + return processor; + } + } +}
