This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 9a4805438a58 CAMEL-26850: camel-jbang - camel run: a route with
transacted processes no messages (#29624)
9a4805438a58 is described below
commit 9a4805438a58b14d089096587ed008253285de12
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Nov 11 12:14:51 2026 +0100
CAMEL-26850: camel-jbang - camel run: a route with transacted processes no
messages (#29624)
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;
+ }
+ }
+}
