This is an automated email from the ASF dual-hosted git repository.
Croway 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 13dc237f18b5 CAMEL-24044: camel-sql - Fix Postgres aggregation
repositories binding version without a placeholder
13dc237f18b5 is described below
commit 13dc237f18b5ed99b03803a4417a9afe08ef2528
Author: croway <[email protected]>
AuthorDate: Mon Jul 13 11:38:56 2026 +0200
CAMEL-24044: camel-sql - Fix Postgres aggregation repositories binding
version without a placeholder
---
components/camel-sql/pom.xml | 13 ++
.../camel-sql/src/main/docs/sql-component.adoc | 2 +-
.../ClusteredPostgresAggregationRepository.java | 14 +-
.../jdbc/PostgresAggregationRepository.java | 9 +-
.../ClusteredPostgresAggregationRepositoryIT.java | 135 ++++++++++++++++++
.../jdbc/PostgresAggregationRepositoryIT.java | 156 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_22.adoc | 14 ++
7 files changed, 334 insertions(+), 9 deletions(-)
diff --git a/components/camel-sql/pom.xml b/components/camel-sql/pom.xml
index 2779857598e9..c64112851391 100644
--- a/components/camel-sql/pom.xml
+++ b/components/camel-sql/pom.xml
@@ -94,6 +94,19 @@
<version>${mariadb-version}</version>
<scope>test</scope>
</dependency>
+ <!-- PostgreSQL only used for the aggregation repository integration
tests -->
+ <dependency>
+ <groupId>org.postgresql</groupId>
+ <artifactId>postgresql</artifactId>
+ <version>${pgjdbc-driver-version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-postgres</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git a/components/camel-sql/src/main/docs/sql-component.adoc
b/components/camel-sql/src/main/docs/sql-component.adoc
index 5e61e31b8923..aa5d01253367 100644
--- a/components/camel-sql/src/main/docs/sql-component.adoc
+++ b/components/camel-sql/src/main/docs/sql-component.adoc
@@ -1040,7 +1040,7 @@ uses special `INSERT .. ON CONFLICT ..` statement to
provide optimistic locking
This statement is (with default aggregation table definition):
[source,sql]
----
-INSERT INTO aggregation (id, exchange) values (?, ?) ON CONFLICT DO NOTHING
+INSERT INTO aggregation (exchange, id, version) VALUES (?, ?, ?) ON CONFLICT
DO NOTHING
----
Details can be found https://www.postgresql.org/docs/9.5/sql-insert.html[in
PostgreSQL documentation].
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredPostgresAggregationRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredPostgresAggregationRepository.java
index 1ee8289b4205..71e01e071c19 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredPostgresAggregationRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/ClusteredPostgresAggregationRepository.java
@@ -56,13 +56,14 @@ public class ClusteredPostgresAggregationRepository extends
ClusteredJdbcAggrega
final CamelContext camelContext, final String correlationId, final
Exchange exchange, String repositoryName,
final Long version, final boolean completed)
throws Exception {
- // The default totalParameterIndex is 2 for ID and Exchange. Depending
on logic this will be increased
- int totalParameterIndex = 2;
+ // The default totalParameterIndex is 3 for ID, Exchange and version.
Depending on logic this will be increased
+ int totalParameterIndex = 3;
StringBuilder queryBuilder = new StringBuilder(256)
.append("INSERT INTO ").append(repositoryName)
.append('(')
.append(EXCHANGE).append(", ")
- .append(ID);
+ .append(ID).append(", ")
+ .append(VERSION);
if (isStoreBodyAsText()) {
queryBuilder.append(", ").append(BODY);
@@ -76,6 +77,11 @@ public class ClusteredPostgresAggregationRepository extends
ClusteredJdbcAggrega
}
}
+ if (completed && isRecoveryByInstance()) {
+ queryBuilder.append(", ").append("instance_id");
+ totalParameterIndex++;
+ }
+
queryBuilder.append(") VALUES (");
queryBuilder.append("?, ".repeat(Math.max(0, totalParameterIndex -
1)));
@@ -85,7 +91,7 @@ public class ClusteredPostgresAggregationRepository extends
ClusteredJdbcAggrega
String sql = queryBuilder.toString();
- int updateCount = insertHelper(camelContext, correlationId, exchange,
sql, 1L, completed);
+ int updateCount = insertHelper(camelContext, correlationId, exchange,
sql, version, completed);
if (updateCount == 0 && getRepositoryName().equals(repositoryName)) {
throw new DataIntegrityViolationException("No row was inserted due
to data violation");
}
diff --git
a/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/PostgresAggregationRepository.java
b/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/PostgresAggregationRepository.java
index 4970877f0846..1f876e45f035 100644
---
a/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/PostgresAggregationRepository.java
+++
b/components/camel-sql/src/main/java/org/apache/camel/processor/aggregate/jdbc/PostgresAggregationRepository.java
@@ -56,13 +56,14 @@ public class PostgresAggregationRepository extends
JdbcAggregationRepository {
final CamelContext camelContext, final String correlationId, final
Exchange exchange, String repositoryName,
final Long version)
throws Exception {
- // The default totalParameterIndex is 2 for ID and Exchange. Depending
on logic this will be increased
- int totalParameterIndex = 2;
+ // The default totalParameterIndex is 3 for ID, Exchange and version.
Depending on logic this will be increased
+ int totalParameterIndex = 3;
StringBuilder queryBuilder = new StringBuilder(256)
.append("INSERT INTO ").append(repositoryName)
.append('(')
.append(EXCHANGE).append(", ")
- .append(ID);
+ .append(ID).append(", ")
+ .append(VERSION);
if (isStoreBodyAsText()) {
queryBuilder.append(", ").append(BODY);
@@ -85,7 +86,7 @@ public class PostgresAggregationRepository extends
JdbcAggregationRepository {
String sql = queryBuilder.toString();
- int updateCount = insertHelper(camelContext, correlationId, exchange,
sql, 1L);
+ int updateCount = insertHelper(camelContext, correlationId, exchange,
sql, version);
if (updateCount == 0 && getRepositoryName().equals(repositoryName)) {
throw new DataIntegrityViolationException("No row was inserted due
to data violation");
}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/ClusteredPostgresAggregationRepositoryIT.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/ClusteredPostgresAggregationRepositoryIT.java
new file mode 100644
index 000000000000..12520c98bce9
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/ClusteredPostgresAggregationRepositoryIT.java
@@ -0,0 +1,135 @@
+/*
+ * 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.processor.aggregate.jdbc;
+
+import java.util.Set;
+
+import javax.sql.DataSource;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.test.infra.postgres.services.PostgresService;
+import org.apache.camel.test.infra.postgres.services.PostgresServiceFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.postgresql.ds.PGSimpleDataSource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies {@link ClusteredPostgresAggregationRepository} against a real
PostgreSQL database, including instance-scoped
+ * recovery where each cluster node must only recover the exchanges it
completed itself.
+ */
+public class ClusteredPostgresAggregationRepositoryIT {
+
+ private static final String REPO_NAME = "camel_clu_agg_it";
+
+ @RegisterExtension
+ public static PostgresService service =
PostgresServiceFactory.createService();
+
+ private CamelContext context;
+ private DataSource dataSource;
+ private JdbcTemplate jdbcTemplate;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ PGSimpleDataSource ds = new PGSimpleDataSource();
+ ds.setUrl(service.jdbcUrl());
+ ds.setUser(service.userName());
+ ds.setPassword(service.password());
+ dataSource = ds;
+
+ jdbcTemplate = new JdbcTemplate(dataSource);
+ jdbcTemplate.execute("CREATE TABLE IF NOT EXISTS " + REPO_NAME
+ + " (id VARCHAR(255) NOT NULL PRIMARY KEY,
exchange BYTEA NOT NULL, version BIGINT NOT NULL)");
+ jdbcTemplate.execute("CREATE TABLE IF NOT EXISTS " + REPO_NAME +
"_completed"
+ + " (id VARCHAR(255) NOT NULL PRIMARY KEY,
exchange BYTEA NOT NULL, version BIGINT NOT NULL,"
+ + " instance_id VARCHAR(255))");
+ jdbcTemplate.execute("TRUNCATE TABLE " + REPO_NAME);
+ jdbcTemplate.execute("TRUNCATE TABLE " + REPO_NAME + "_completed");
+
+ context = new DefaultCamelContext();
+ context.start();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (context != null) {
+ context.stop();
+ }
+ }
+
+ private ClusteredPostgresAggregationRepository createRepository(String
instanceId) {
+ ClusteredPostgresAggregationRepository repo = new
ClusteredPostgresAggregationRepository(
+ new DataSourceTransactionManager(dataSource), REPO_NAME,
dataSource);
+ repo.setInstanceId(instanceId);
+ repo.setRecoveryByInstance(true);
+ return repo;
+ }
+
+ @Test
+ public void testInstanceScopedRecovery() throws Exception {
+ ClusteredPostgresAggregationRepository repoA =
createRepository("nodeA");
+ ClusteredPostgresAggregationRepository repoB =
createRepository("nodeB");
+ ServiceHelper.startService(repoA, repoB);
+ try {
+ // nodeA aggregates an exchange
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody("payload");
+ assertNull(repoA.add(context, "foo", exchange));
+
+ Exchange stored = repoA.get(context, "foo");
+ assertNotNull(stored);
+ assertEquals("payload", stored.getIn().getBody());
+
+ // completing moves it to the completed table, tagged with nodeA's
instance id
+ repoA.remove(context, "foo", stored);
+ assertNull(repoA.get(context, "foo"));
+
+ String exchangeId = stored.getExchangeId();
+ String instanceId = jdbcTemplate.queryForObject(
+ "SELECT instance_id FROM " + REPO_NAME + "_completed WHERE
id = ?", String.class, exchangeId);
+ assertEquals("nodeA", instanceId,
+ "completed exchange must be tagged with the instance id of
the node that completed it");
+
+ // only nodeA may recover it
+ Set<String> scannedByA = repoA.scan(context);
+ assertEquals(Set.of(exchangeId), scannedByA);
+ assertTrue(repoB.scan(context).isEmpty(), "nodeB must not recover
exchanges completed by nodeA");
+
+ Exchange recovered = repoA.recover(context, exchangeId);
+ assertNotNull(recovered);
+ assertEquals("payload", recovered.getIn().getBody());
+
+ // confirm deletes it from the completed table
+ repoA.confirm(context, exchangeId);
+ assertTrue(repoA.scan(context).isEmpty());
+ } finally {
+ ServiceHelper.stopService(repoA, repoB);
+ }
+ }
+}
diff --git
a/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/PostgresAggregationRepositoryIT.java
b/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/PostgresAggregationRepositoryIT.java
new file mode 100644
index 000000000000..93401144b3b2
--- /dev/null
+++
b/components/camel-sql/src/test/java/org/apache/camel/processor/aggregate/jdbc/PostgresAggregationRepositoryIT.java
@@ -0,0 +1,156 @@
+/*
+ * 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.processor.aggregate.jdbc;
+
+import java.util.Set;
+
+import javax.sql.DataSource;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.apache.camel.support.service.ServiceHelper;
+import org.apache.camel.test.infra.postgres.services.PostgresService;
+import org.apache.camel.test.infra.postgres.services.PostgresServiceFactory;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+import org.postgresql.ds.PGSimpleDataSource;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.DataSourceTransactionManager;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Verifies {@link PostgresAggregationRepository} against a real PostgreSQL
database.
+ */
+public class PostgresAggregationRepositoryIT {
+
+ private static final String REPO_NAME = "camel_agg_it";
+ private static final String REPO_NAME_TEXT = "camel_agg_text_it";
+
+ @RegisterExtension
+ public static PostgresService service =
PostgresServiceFactory.createService();
+
+ private CamelContext context;
+ private DataSource dataSource;
+ private JdbcTemplate jdbcTemplate;
+
+ @BeforeEach
+ public void setUp() throws Exception {
+ PGSimpleDataSource ds = new PGSimpleDataSource();
+ ds.setUrl(service.jdbcUrl());
+ ds.setUser(service.userName());
+ ds.setPassword(service.password());
+ dataSource = ds;
+
+ jdbcTemplate = new JdbcTemplate(dataSource);
+ for (String table : new String[] { REPO_NAME, REPO_NAME + "_completed"
}) {
+ jdbcTemplate.execute("CREATE TABLE IF NOT EXISTS " + table
+ + " (id VARCHAR(255) NOT NULL PRIMARY KEY,
exchange BYTEA NOT NULL, version BIGINT NOT NULL)");
+ jdbcTemplate.execute("TRUNCATE TABLE " + table);
+ }
+ for (String table : new String[] { REPO_NAME_TEXT, REPO_NAME_TEXT +
"_completed" }) {
+ jdbcTemplate.execute("CREATE TABLE IF NOT EXISTS " + table
+ + " (id VARCHAR(255) NOT NULL PRIMARY KEY,
exchange BYTEA NOT NULL, version BIGINT NOT NULL, body TEXT)");
+ jdbcTemplate.execute("TRUNCATE TABLE " + table);
+ }
+
+ context = new DefaultCamelContext();
+ context.start();
+ }
+
+ @AfterEach
+ public void tearDown() {
+ if (context != null) {
+ context.stop();
+ }
+ }
+
+ private PostgresAggregationRepository createRepository(String
repositoryName) {
+ return new PostgresAggregationRepository(
+ new DataSourceTransactionManager(dataSource), repositoryName,
dataSource);
+ }
+
+ @Test
+ public void testAddGetUpdateRemoveRecoverConfirm() throws Exception {
+ PostgresAggregationRepository repo = createRepository(REPO_NAME);
+ ServiceHelper.startService(repo);
+ try {
+ // insert a new aggregation
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody("counter:1");
+ assertNull(repo.add(context, "foo", exchange));
+
+ // read it back
+ Exchange actual = repo.get(context, "foo");
+ assertNotNull(actual);
+ assertEquals("counter:1", actual.getIn().getBody());
+
+ // update it using the version loaded by get()
+ actual.getIn().setBody("counter:2");
+ repo.add(context, "foo", actual);
+ actual = repo.get(context, "foo");
+ assertEquals("counter:2", actual.getIn().getBody());
+
+ // complete the aggregation: moves it to the completed table
+ repo.remove(context, "foo", actual);
+ assertNull(repo.get(context, "foo"));
+
+ Set<String> completed = repo.scan(context);
+ assertEquals(1, completed.size());
+
+ String exchangeId = completed.iterator().next();
+ Exchange recovered = repo.recover(context, exchangeId);
+ assertNotNull(recovered);
+ assertEquals("counter:2", recovered.getIn().getBody());
+
+ // confirm deletes it from the completed table
+ repo.confirm(context, exchangeId);
+ assertTrue(repo.scan(context).isEmpty());
+ } finally {
+ ServiceHelper.stopService(repo);
+ }
+ }
+
+ @Test
+ public void testStoreBodyAsText() throws Exception {
+ PostgresAggregationRepository repo = createRepository(REPO_NAME_TEXT);
+ repo.setStoreBodyAsText(true);
+ ServiceHelper.startService(repo);
+ try {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setBody("myBody");
+ repo.add(context, "foo", exchange);
+
+ // the body column must contain the message body as text
+ String body = jdbcTemplate.queryForObject(
+ "SELECT body FROM " + REPO_NAME_TEXT + " WHERE id = ?",
String.class, "foo");
+ assertEquals("myBody", body);
+
+ Exchange actual = repo.get(context, "foo");
+ assertEquals("myBody", actual.getIn().getBody());
+ } finally {
+ ServiceHelper.stopService(repo);
+ }
+ }
+}
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 c74ef56633b7..44014f2466d3 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
@@ -333,3 +333,17 @@ getters and setters of
`SupervisingRouteControllerConfiguration` and
The clustered route controller property
`camel.clustered.controller.initial-delay` is unchanged:
it keeps its `String` type because it supports Camel time patterns (for
example `10 seconds`)
that `Duration` binding does not.
+
+=== camel-sql - PostgreSQL aggregation repositories fixed
+
+`PostgresAggregationRepository` and `ClusteredPostgresAggregationRepository`
were broken since
+Camel 3.4: every insert failed with a parameter-index error because the
`version` column was
+bound but not included in the generated `INSERT ... ON CONFLICT DO NOTHING`
statement. Both
+repositories now include the `version` column in the insert, and the clustered
variant also
+writes the `instance_id` column to the completed table when
`recoveryByInstance` is enabled
+(previously instance-scoped recovery could never match any row).
+
+The aggregation tables used with these repositories must have the `version
BIGINT` column, as
+already required by `JdbcAggregationRepository` for reading and updating
aggregates. When
+`recoveryByInstance` is enabled, the completed table must have the
`instance_id VARCHAR(255)`
+column as documented.
