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 605845de1603 CAMEL-25511: camel-jooq - do not delete an entity whose
exchange failed
605845de1603 is described below
commit 605845de1603dd6dc155fb21f9d0aafa3b173a02
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 10 12:21:52 2026 +0530
CAMEL-25511: camel-jooq - do not delete an entity whose exchange failed
With consumeDelete=true (the default) the jOOQ consumer deleted every
entity of a poll after the batch, also the entities whose exchange failed
(a failed route sets the exception on the exchange instead of throwing
it), so those rows were lost. The consumer now deletes only the entities
whose exchange completed successfully. An entity whose exchange failed or
was marked rollback only, or that was not processed because a graceful
shutdown started in the middle of the batch, stays in the table and is
consumed again by the next poll. Handle the exception in the route, for
example with onException(...).handled(true), to delete it anyway.
The exchanges are no longer auto released, so the outcome is read before
the pooled exchange factory resets the exchange, and a processing error
is passed to the consumer's exception handler, as in camel-jpa.
Closes #27644
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/docs/jooq-component.adoc | 3 +
.../camel-jooq/src/main/docs/jooq-component.adoc | 3 +
.../apache/camel/component/jooq/JooqConsumer.java | 41 +++++++--
...JooqConsumerDeleteFailedPooledExchangeTest.java | 34 ++++++++
.../jooq/JooqConsumerDeleteFailedTest.java | 98 ++++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 10 +++
6 files changed, 182 insertions(+), 7 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jooq-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jooq-component.adoc
index 3e383ab121dc..a6d33c8bad3d 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jooq-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/jooq-component.adoc
@@ -93,6 +93,9 @@ See the examples below.
Consuming messages from a JOOQ consumer endpoint removes (or updates) entity
beans in the database.
This allows you to use a database table as a logical queue: consumers take
messages from the queue and then delete/update them to logically remove them
from the queue.
If you do not wish to delete the entity bean when it has been processed, you
can specify consumeDelete=false on the URI.
+An entity bean is only deleted when its exchange completed successfully: if
the exchange failed, the entity stays in
+the table and is consumed again by the next poll. An entity that always fails
is consumed again at every poll; handle the
+failure in the route (for example with `onException(...).handled(true)`) to
delete the entity regardless of the outcome.
=== Operations
diff --git a/components/camel-jooq/src/main/docs/jooq-component.adoc
b/components/camel-jooq/src/main/docs/jooq-component.adoc
index 3e383ab121dc..a6d33c8bad3d 100644
--- a/components/camel-jooq/src/main/docs/jooq-component.adoc
+++ b/components/camel-jooq/src/main/docs/jooq-component.adoc
@@ -93,6 +93,9 @@ See the examples below.
Consuming messages from a JOOQ consumer endpoint removes (or updates) entity
beans in the database.
This allows you to use a database table as a logical queue: consumers take
messages from the queue and then delete/update them to logically remove them
from the queue.
If you do not wish to delete the entity bean when it has been processed, you
can specify consumeDelete=false on the URI.
+An entity bean is only deleted when its exchange completed successfully: if
the exchange failed, the entity stays in
+the table and is consumed again by the next poll. An entity that always fails
is consumed again at every poll; handle the
+failure in the route (for example with `onException(...).handled(true)`) to
delete the entity regardless of the outcome.
=== Operations
diff --git
a/components/camel-jooq/src/main/java/org/apache/camel/component/jooq/JooqConsumer.java
b/components/camel-jooq/src/main/java/org/apache/camel/component/jooq/JooqConsumer.java
index d1f8e8cb4294..bf1a46576f04 100644
---
a/components/camel-jooq/src/main/java/org/apache/camel/component/jooq/JooqConsumer.java
+++
b/components/camel-jooq/src/main/java/org/apache/camel/component/jooq/JooqConsumer.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.component.jooq;
+import java.util.ArrayList;
import java.util.LinkedList;
+import java.util.List;
import java.util.Queue;
import org.apache.camel.Exchange;
@@ -35,6 +37,8 @@ public class JooqConsumer extends
ScheduledBatchPollingConsumer {
private static final class DataHolder {
private Exchange exchange;
+ private UpdatableRecord<?> record;
+ private boolean consumed;
private DataHolder() {
}
@@ -63,23 +67,38 @@ public class JooqConsumer extends
ScheduledBatchPollingConsumer {
// okay we have some response from jooq so lets mark the consumer as
ready
forceConsumerAsReady();
+ List<DataHolder> holders = new ArrayList<>(results.size());
for (UpdatableRecord<?> result : results) {
DataHolder holder = new DataHolder();
holder.exchange = createExchange(result);
- answer.add(holder);
+ holder.record = result;
+ holders.add(holder);
}
+ answer.addAll(holders);
int messagePolled = processBatch(CastUtils.cast(answer));
if (configuration.isConsumeDelete()) {
- context.batchDelete(results).execute();
+ // only delete the entities whose exchange was processed
successfully, so a failed (or not processed)
+ // entity stays in the table and is consumed again by the next poll
+ List<UpdatableRecord<?>> consumed = new
ArrayList<>(holders.size());
+ for (DataHolder holder : holders) {
+ if (holder.consumed) {
+ consumed.add(holder.record);
+ }
+ }
+ if (!consumed.isEmpty()) {
+ context.batchDelete(consumed).execute();
+ }
}
return messagePolled;
}
protected Exchange createExchange(Object result) {
- Exchange exchange = createExchange(true);
+ // not auto released: the pooled exchange factory resets an auto
released exchange when its unit of work is done,
+ // before processBatch reads whether it failed (it releases the
exchange itself)
+ Exchange exchange = createExchange(false);
exchange.getIn().setBody(result);
return exchange;
}
@@ -93,11 +112,19 @@ public class JooqConsumer extends
ScheduledBatchPollingConsumer {
public int processBatch(Queue<Object> exchanges) throws Exception {
int total = exchanges.size();
- if (isBatchAllowed()) {
- for (int i = 0; i < total; i++) {
- DataHolder holder =
org.apache.camel.util.ObjectHelper.cast(DataHolder.class, exchanges.poll());
- getProcessor().process(holder.exchange);
+ // only loop while we are allowed to run: when a graceful shutdown
starts, the remaining entities are not
+ // processed, so they are not deleted and the next start consumes them
+ for (int index = 0; index < total && isBatchAllowed(); index++) {
+ DataHolder holder =
org.apache.camel.util.ObjectHelper.cast(DataHolder.class, exchanges.poll());
+ Exchange exchange = holder.exchange;
+ try {
+ getProcessor().process(exchange);
+ } catch (Exception e) {
+ exchange.setException(e);
+ getExceptionHandler().handleException("Error processing
exchange", exchange, e);
}
+ holder.consumed = !exchange.isFailed() &&
!exchange.isRollbackOnly();
+ releaseExchange(exchange, false);
}
return total;
diff --git
a/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedPooledExchangeTest.java
b/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedPooledExchangeTest.java
new file mode 100644
index 000000000000..f443dc4362bc
--- /dev/null
+++
b/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedPooledExchangeTest.java
@@ -0,0 +1,34 @@
+/*
+ * 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.jooq;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+
+/**
+ * The same cases with the pooled exchange factory, which resets an exchange
when its unit of work is done: the consumer
+ * must read the outcome of the exchange before it is released.
+ */
+class JooqConsumerDeleteFailedPooledExchangeTest extends
JooqConsumerDeleteFailedTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ camelContext.getCamelContextExtension().setExchangeFactory(new
PooledExchangeFactory());
+ return camelContext;
+ }
+}
diff --git
a/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedTest.java
b/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedTest.java
new file mode 100644
index 000000000000..1bd7347c8f00
--- /dev/null
+++
b/components/camel-jooq/src/test/java/org/apache/camel/component/jooq/JooqConsumerDeleteFailedTest.java
@@ -0,0 +1,98 @@
+/*
+ * 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.jooq;
+
+import java.util.List;
+
+import org.apache.camel.RoutesBuilder;
+import org.apache.camel.ShutdownRunningTask;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.jooq.db.tables.records.AuthorRecord;
+import org.junit.jupiter.api.Test;
+
+import static org.apache.camel.component.jooq.db.Tables.AUTHOR;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * With consumeDelete (the default) an entity must only be deleted when its
exchange was processed successfully: a
+ * failed or rollback only exchange, or an entity that was not processed, must
leave the row in the table, so the next
+ * poll consumes it again.
+ */
+class JooqConsumerDeleteFailedTest extends BaseJooqTest {
+
+ @Test
+ void testFailedExchangeKeepsRow() throws Exception {
+ create.executeInsert(new AuthorRecord(1, null, "ok", null, null,
null));
+ create.executeInsert(new AuthorRecord(2, null, "fail", null, null,
null));
+
+ JooqConsumer consumer = (JooqConsumer)
context.getRoute("delete-failed").getConsumer();
+ // one poll in the test thread (the scheduler is not started)
+ assertEquals(2, consumer.poll());
+
+ List<Integer> remaining =
create.select(AUTHOR.ID).from(AUTHOR).orderBy(AUTHOR.ID).fetch(AUTHOR.ID);
+ assertEquals(List.of(2), remaining, "only the successfully processed
author must be deleted");
+ }
+
+ @Test
+ void testRollbackOnlyExchangeKeepsRow() throws Exception {
+ create.executeInsert(new AuthorRecord(1, null, "ok", null, null,
null));
+ create.executeInsert(new AuthorRecord(2, null, "rollback", null, null,
null));
+
+ JooqConsumer consumer = (JooqConsumer)
context.getRoute("delete-failed").getConsumer();
+ assertEquals(2, consumer.poll());
+
+ List<Integer> remaining =
create.select(AUTHOR.ID).from(AUTHOR).orderBy(AUTHOR.ID).fetch(AUTHOR.ID);
+ assertEquals(List.of(2), remaining, "an author whose exchange was
marked rollback only must not be deleted");
+ }
+
+ @Test
+ void testNotProcessedEntitiesKeepRows() throws Exception {
+ create.executeInsert(new AuthorRecord(1, null, "ok", null, null,
null));
+ create.executeInsert(new AuthorRecord(3, null, "ok", null, null,
null));
+
+ JooqConsumer consumer = (JooqConsumer)
context.getRoute("delete-failed").getConsumer();
+ // what the shutdown strategy calls when a graceful shutdown starts
while a poll has fetched the entities:
+ // the batch is then not processed
+ consumer.deferShutdown(ShutdownRunningTask.CompleteCurrentTaskOnly);
+ consumer.poll();
+
+ assertEquals(2, create.fetchCount(AUTHOR), "entities that were not
processed must not be deleted");
+ }
+
+ @Override
+ protected RoutesBuilder createRouteBuilder() {
+ JooqComponent jooqComponent = (JooqComponent)
context().getComponent("jooq");
+ JooqConfiguration jooqConfiguration = new JooqConfiguration();
+ jooqConfiguration.setDatabaseConfiguration(create.configuration());
+ jooqComponent.setConfiguration(jooqConfiguration);
+
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("jooq://org.apache.camel.component.jooq.db.tables.records.AuthorRecord?startScheduler=false")
+ .routeId("delete-failed")
+ .choice().when(simple("${body.lastName} ==
'rollback'")).markRollbackOnly().end()
+ .process(exchange -> {
+ AuthorRecord author =
exchange.getIn().getBody(AuthorRecord.class);
+ if ("fail".equals(author.getLastName())) {
+ throw new IllegalStateException("Simulated
failure for author " + author.getId());
+ }
+ });
+ }
+ };
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index bd76cca82786..99574885768e 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -1158,6 +1158,16 @@ exception in the route (for example with
`onException(...).handled(true)`) to ac
event that always fails now gets `RETRY` every time, until a Dapr resiliency
policy for the pub/sub component or a
dead-letter topic on the subscription limits the redeliveries.
+=== camel-jooq - an entity whose exchange failed is not deleted
+
+With `consumeDelete=true` (the default) the consumer deleted every entity of a
poll after the batch, also the
+entities whose exchange failed (a failed route sets the exception on the
exchange instead of throwing it), so those
+rows were lost. The consumer now deletes only the entities whose exchange
completed successfully; an entity whose
+exchange failed (or was marked rollback only), or that was not processed
because the route was stopping (when a
+graceful shutdown starts in the middle of a batch, the remaining entities are
no longer processed), stays in the table
+and is consumed again by the next poll, so an entity that always fails is now
processed at every poll. Handle
+the exception in the route (for example with `onException(...).handled(true)`)
to delete the entity anyway.
+
=== Components and Language removal
==== camel-csimple, camel-csimple-joor and csimple-maven-plugin
