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 28000bbe7c86 CAMEL-26851: camel-main - transacted with a local
transaction manager when the application has one DataSource (#29629)
28000bbe7c86 is described below
commit 28000bbe7c867360560a446e88e0a406ef2b1725
Author: Claus Ibsen <[email protected]>
AuthorDate: Wed Nov 11 16:13:23 2026 +0100
CAMEL-26851: camel-main - transacted with a local transaction manager when
the application has one DataSource (#29629)
A route that uses transacted needs a transaction manager. Spring Boot and
Quarkus set one up; Camel Main and camel run do not, and with spring-tx on the
classpath the route failed to start: no PlatformTransactionManager in the
registry. When there is none, but the registry has exactly one DataSource and
spring-jdbc is on the classpath, Camel now creates a JdbcTransactionManager for
that DataSource, binds it as transactionManager for the other transacted
routes, and makes the SpringTra [...]
camel run bound a placeholder policy for transacted, which runs the steps
without a transaction (CAMEL-26850). In the same case it now downloads
camel-spring instead, and Camel makes the local transaction manager for the
DataSource. The placeholder and its WARN stay for the other cases; the WARN now
says what gives transactions.
The transactional client page explains the convention, and that the SQL and
Spring JDBC components take part in it while the JDBC component does not.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../camel/catalog/docs/transactional-client.adoc | 7 ++
.../SqlTransactedLocalTransactionManagerTest.java | 107 +++++++++++++++++++++
.../modules/eips/pages/transactional-client.adoc | 7 ++
.../camel/reifier/AbstractPolicyReifier.java | 46 ++++++++-
dsl/camel-kamelet-main/pom.xml | 6 ++
.../camel/main/download/TransactedDownloader.java | 29 +++++-
.../main/download/TransactedDownloaderTest.java | 32 ++++++
7 files changed, 230 insertions(+), 4 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/transactional-client.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/transactional-client.adoc
index 8ec329fb2c66..527f615754f1 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/transactional-client.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/transactional-client.adoc
@@ -188,6 +188,13 @@ manager configured. In more complex scenarios, where you
either use multiple
transaction managers or transaction propagation policies, you have to do
additional
configuration.
+When there is no transaction manager at all, as with Camel Main and the Camel
CLI (`camel run`), and the
+application has exactly one `DataSource`, Camel creates a local
`JdbcTransactionManager` for that `DataSource`
+and binds it as `transactionManager`. This needs `camel-spring` and
`spring-jdbc` on the classpath. The
+xref:ROOT:sql-component.adoc[SQL] and
xref:ROOT:spring-jdbc-component.adoc[Spring JDBC] components take part in
+the transaction, as they get their connections through Spring. The
xref:ROOT:jdbc-component.adoc[JDBC] component
+takes its connections from the `DataSource` directly, so it does not.
+
=== Using different transaction propagations
In some rare situations, you may need to use multiple transactions with the
same exchange.
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlTransactedLocalTransactionManagerTest.java
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlTransactedLocalTransactionManagerTest.java
new file mode 100644
index 000000000000..67dbfb0cd201
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/component/sql/SqlTransactedLocalTransactionManagerTest.java
@@ -0,0 +1,107 @@
+/*
+ * 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.component.sql;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder;
+import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
+import org.springframework.jdbc.support.JdbcTransactionManager;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * CAMEL-26851: transacted without a transaction manager or a policy in the
registry, as with Camel Main and camel run,
+ * uses a local transaction manager for the one DataSource of the application.
+ */
+public class SqlTransactedLocalTransactionManagerTest extends CamelTestSupport
{
+
+ private static final String SQL =
"sql:overriddenByTheHeader?dataSource=#testdb&allowQueryFromHeader=true";
+
+ private EmbeddedDatabase db;
+ private JdbcTemplate jdbc;
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ // only the DataSource: no transaction manager and no transacted policy
+ db = new EmbeddedDatabaseBuilder()
+ .setName(getClass().getSimpleName())
+ .setType(EmbeddedDatabaseType.H2).build();
+ registry.bind("testdb", db);
+ }
+
+ @Override
+ public void doPostSetup() {
+ jdbc = new JdbcTemplate(db);
+ jdbc.execute("CREATE TABLE CUSTOMER (ID VARCHAR(15) NOT NULL PRIMARY
KEY, NAME VARCHAR(100))");
+ }
+
+ @Override
+ public void doPostTearDown() {
+ if (db != null) {
+ db.shutdown();
+ }
+ }
+
+ private void insertTwice(String routeId, String second) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:" + routeId).routeId(routeId)
+ .transacted()
+ .to(SQL)
+ .setHeader(SqlConstants.SQL_QUERY, constant(second))
+ .to(SQL);
+ }
+ });
+ }
+
+ private Exchange send(String routeId) {
+ return template.send("direct:" + routeId,
+ e -> e.getIn().setHeader(SqlConstants.SQL_QUERY, "insert into
customer values('cust1','cmueller')"));
+ }
+
+ @Test
+ public void bothInsertsCommit() throws Exception {
+ insertTwice("commit", "insert into customer
values('cust2','muellerc')");
+
+ assertFalse(send("commit").isFailed());
+ assertEquals(2, jdbc.queryForObject("select count(*) from customer",
Long.class));
+ // the local transaction manager is bound, so other transacted routes
use the same one
+ Object tm = context.getRegistry().lookupByName("transactionManager");
+ JdbcTransactionManager local =
assertInstanceOf(JdbcTransactionManager.class, tm);
+ assertSame(db, local.getDataSource());
+ }
+
+ @Test
+ public void aFailingSecondInsertRollsBackTheFirst() throws Exception {
+ // primary key violation
+ insertTwice("rollback", "insert into customer
values('cust1','muellerc')");
+
+ assertTrue(send("rollback").isFailed());
+ assertEquals(0, jdbc.queryForObject("select count(*) from customer",
Long.class));
+ }
+}
diff --git
a/core/camel-core-engine/src/main/docs/modules/eips/pages/transactional-client.adoc
b/core/camel-core-engine/src/main/docs/modules/eips/pages/transactional-client.adoc
index 8ec329fb2c66..527f615754f1 100644
---
a/core/camel-core-engine/src/main/docs/modules/eips/pages/transactional-client.adoc
+++
b/core/camel-core-engine/src/main/docs/modules/eips/pages/transactional-client.adoc
@@ -188,6 +188,13 @@ manager configured. In more complex scenarios, where you
either use multiple
transaction managers or transaction propagation policies, you have to do
additional
configuration.
+When there is no transaction manager at all, as with Camel Main and the Camel
CLI (`camel run`), and the
+application has exactly one `DataSource`, Camel creates a local
`JdbcTransactionManager` for that `DataSource`
+and binds it as `transactionManager`. This needs `camel-spring` and
`spring-jdbc` on the classpath. The
+xref:ROOT:sql-component.adoc[SQL] and
xref:ROOT:spring-jdbc-component.adoc[Spring JDBC] components take part in
+the transaction, as they get their connections through Spring. The
xref:ROOT:jdbc-component.adoc[JDBC] component
+takes its connections from the `DataSource` directly, so it does not.
+
=== Using different transaction propagations
In some rare situations, you may need to use multiple transactions with the
same exchange.
diff --git
a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/AbstractPolicyReifier.java
b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/AbstractPolicyReifier.java
index 960d3299e4d6..6680be8a5011 100644
---
a/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/AbstractPolicyReifier.java
+++
b/core/camel-core-reifier/src/main/java/org/apache/camel/reifier/AbstractPolicyReifier.java
@@ -17,6 +17,7 @@
package org.apache.camel.reifier;
import java.lang.reflect.Method;
+import java.util.Set;
import org.apache.camel.CamelContext;
import org.apache.camel.Route;
@@ -79,7 +80,16 @@ public abstract class AbstractPolicyReifier<T extends
ProcessorDefinition<?>> ex
if (tmClazz != null) {
// see if we can find the platform transaction manager in the
// registry
- Object transactionManager = mandatoryFindSingleByType(tmClazz);
+ Object transactionManager = findSingleByType(tmClazz);
+ if (transactionManager == null) {
+ // none, but the application has one DataSource (Camel
Main, camel run): a local transaction
+ // manager for it
+ transactionManager = localTransactionManager(tmClazz);
+ }
+ if (transactionManager == null) {
+ // fails telling that there is no
PlatformTransactionManager
+ transactionManager = mandatoryFindSingleByType(tmClazz);
+ }
// only one platform manager then use it as default and
// create a transacted
// policy with it and default to required
@@ -119,4 +129,38 @@ public abstract class AbstractPolicyReifier<T extends
ProcessorDefinition<?>> ex
return answer;
}
+
+ /**
+ * A local transaction manager for the one DataSource of the registry,
when the registry has no transaction manager,
+ * as with Camel Main and camel run (CAMEL-26851). It is bound as
transactionManager, so the other transacted routes
+ * use it too. Null without exactly one DataSource, or without spring-jdbc
on the classpath.
+ */
+ private Object localTransactionManager(Class<?> tmClazz) {
+ Class<?> dsClazz =
camelContext.getClassResolver().resolveClass("javax.sql.DataSource");
+ Class<?> jtmClazz = camelContext.getClassResolver()
+
.resolveClass("org.springframework.jdbc.support.JdbcTransactionManager");
+ if (dsClazz == null || jtmClazz == null ||
!tmClazz.isAssignableFrom(jtmClazz)) {
+ return null;
+ }
+ Set<?> dataSources = camelContext.getRegistry().findByType(dsClazz);
+ if (dataSources.size() != 1) {
+ return null;
+ }
+ Object dataSource = dataSources.iterator().next();
+ Object transactionManager;
+ try {
+ transactionManager =
jtmClazz.getConstructor(dsClazz).newInstance(dataSource);
+ } catch (Exception e) {
+ throw new RuntimeCamelException("Cannot create a
JdbcTransactionManager for the DataSource: " + dataSource, e);
+ }
+ if (camelContext.getRegistry().lookupByName("transactionManager") ==
null) {
+ camelContext.getRegistry().bind("transactionManager",
transactionManager);
+ LOG.info("Using a local JdbcTransactionManager for transacted
routes, with the DataSource: {}", dataSource);
+ } else {
+ LOG.debug("Using a local JdbcTransactionManager for this
transacted route only, with the DataSource: {},"
+ + " as another bean is bound as transactionManager",
+ dataSource);
+ }
+ return transactionManager;
+ }
}
diff --git a/dsl/camel-kamelet-main/pom.xml b/dsl/camel-kamelet-main/pom.xml
index f09d2aa98791..c981972f75d2 100644
--- a/dsl/camel-kamelet-main/pom.xml
+++ b/dsl/camel-kamelet-main/pom.xml
@@ -183,6 +183,12 @@
</dependency>
<!-- test -->
+ <dependency>
+ <groupId>org.springframework</groupId>
+ <artifactId>spring-jdbc</artifactId>
+ <version>${spring-version}</version>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.junit.jupiter</groupId>
<artifactId>junit-jupiter-api</artifactId>
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 a455a36c1359..c4934de8f436 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,6 +16,8 @@
*/
package org.apache.camel.main.download;
+import javax.sql.DataSource;
+
import org.apache.camel.CamelContext;
import org.apache.camel.NamedNode;
import org.apache.camel.Processor;
@@ -59,10 +61,16 @@ public class TransactedDownloader {
if (downloader != null) {
downloader.downloadDependency("org.apache.camel",
"camel-jta",
route.getCamelContext().getVersion());
- if
(bindPassThroughPolicies(route.getCamelContext(), transacted)) {
+ if
(usesLocalTransactionManager(route.getCamelContext(), transacted)) {
+ // CAMEL-26851: Camel makes a local
transaction manager for the one DataSource, and the
+ // policy of it, which is in camel-spring
+
downloader.downloadDependency("org.apache.camel", "camel-spring",
+ route.getCamelContext().getVersion());
+ } else 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.",
+ + " without a transaction. Declare
one DataSource (spring.datasource.*) with"
+ + " spring-jdbc on the classpath, as
the sql component has, or a transaction"
+ + " manager bean, or export to Spring
Boot or Quarkus, for transactions.",
route.getRouteId());
}
}
@@ -92,6 +100,21 @@ public class TransactedDownloader {
return true;
}
+ /**
+ * Whether Camel makes a local transaction manager for the transacted
(CAMEL-26851): it has no policy or reference
+ * of its own, the registry has no transaction manager but exactly one
DataSource, and spring-jdbc is on the
+ * classpath.
+ */
+ static boolean usesLocalTransactionManager(CamelContext context,
TransactedDefinition transacted) {
+ if (transacted.getPolicy() != null || transacted.getRef() != null ||
hasTransactionManager(context)) {
+ return false;
+ }
+ // with the application class loader, as for the
PlatformTransactionManager below
+ Class<?> jtm =
ObjectHelper.loadClass("org.springframework.jdbc.support.JdbcTransactionManager",
+ context.getApplicationContextClassLoader());
+ return jtm != null &&
context.getRegistry().findByType(DataSource.class).size() == 1;
+ }
+
/**
* Whether the registry has a transacted policy of the user's own, or a
Spring PlatformTransactionManager, of which
* Camel makes the policy.
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
index bd4e3e007516..ca7da114c7b2 100644
---
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
@@ -26,6 +26,7 @@ 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 org.springframework.jdbc.datasource.DriverManagerDataSource;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -82,6 +83,37 @@ public class TransactedDownloaderTest {
}
}
+ @Test
+ public void oneDataSourceWithSpringJdbcGetsALocalTransactionManager()
throws Exception {
+ // CAMEL-26851: Camel makes a local transaction manager for it, so no
placeholder is bound
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.getRegistry().bind("orders", new
DriverManagerDataSource("jdbc:h2:mem:orders"));
+
assertTrue(TransactedDownloader.usesLocalTransactionManager(context, new
TransactedDefinition()));
+ }
+ }
+
+ @Test
+ public void noLocalTransactionManagerWithoutExactlyOneDataSource() throws
Exception {
+ try (CamelContext context = new DefaultCamelContext()) {
+
assertFalse(TransactedDownloader.usesLocalTransactionManager(context, new
TransactedDefinition()));
+ context.getRegistry().bind("orders", new
DriverManagerDataSource("jdbc:h2:mem:orders"));
+ context.getRegistry().bind("audit", new
DriverManagerDataSource("jdbc:h2:mem:audit"));
+
assertFalse(TransactedDownloader.usesLocalTransactionManager(context, new
TransactedDefinition()));
+ }
+ }
+
+ @Test
+ public void noLocalTransactionManagerForAPolicyOfItsOwn() throws Exception
{
+ try (CamelContext context = new DefaultCamelContext()) {
+ context.getRegistry().bind("orders", new
DriverManagerDataSource("jdbc:h2:mem:orders"));
+ TransactedDefinition transacted = new TransactedDefinition();
+ transacted.setRef("myPolicy");
+
assertFalse(TransactedDownloader.usesLocalTransactionManager(context,
transacted));
+ context.getRegistry().bind("PROPAGATION_REQUIRED", new
CountingPolicy());
+
assertFalse(TransactedDownloader.usesLocalTransactionManager(context, new
TransactedDefinition()));
+ }
+ }
+
private static RouteBuilder orders() {
return new RouteBuilder() {
@Override
