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 6b79e833bb80 CAMEL-25221: camel-couchbase - remove a document only
after its exchange completed (#27347)
6b79e833bb80 is described below
commit 6b79e833bb80cba824d2f058bee3720c026e529c
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 5 22:37:58 2026 +0530
CAMEL-25221: camel-couchbase - remove a document only after its exchange
completed (#27347)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../component/couchbase/CouchbaseConsumer.java | 153 +++++--
.../CouchbaseConsumerProcessedStrategyTest.java | 448 +++++++++++++++++++++
.../component/couchbase/CouchbaseConsumerTest.java | 3 +-
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 24 +-
4 files changed, 588 insertions(+), 40 deletions(-)
diff --git
a/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
b/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
index 8200004c8cf7..897268a3b57f 100644
---
a/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
+++
b/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseConsumer.java
@@ -18,6 +18,8 @@ package org.apache.camel.component.couchbase;
import java.util.ArrayDeque;
import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -38,6 +40,7 @@ import org.apache.camel.Processor;
import org.apache.camel.resume.ResumeAware;
import org.apache.camel.resume.ResumeStrategy;
import org.apache.camel.support.ScheduledBatchPollingConsumer;
+import org.apache.camel.support.SynchronizationAdapter;
import org.apache.camel.support.resume.ResumeStrategyHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -55,6 +58,11 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
static final String SQL_DOCUMENT_ID_ALIAS = "__id";
private final Lock lock = new ReentrantLock();
+ /**
+ * IDs of the documents whose exchange was created by a poll and has not
completed or failed yet, with
+ * {@code consumerProcessedStrategy=delete}. See {@link
#isInFlight(String)}.
+ */
+ private final Set<String> inFlight = ConcurrentHashMap.newKeySet();
private final CouchbaseEndpoint endpoint;
private Bucket bucket;
private Scope scope;
@@ -144,17 +152,25 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
protected int poll() throws Exception {
lock.lock();
try {
- if (useSqlQuery) {
- return pollWithSqlQuery();
- } else {
- return pollWithView();
+ Queue<Object> exchanges = new ArrayDeque<>();
+ try {
+ if (useSqlQuery) {
+ pollWithSqlQuery(exchanges);
+ } else {
+ pollWithView(exchanges);
+ }
+ } catch (Exception e) {
+ // the poll failed part way, so none of the exchanges created
so far is handed to the route
+ releaseUndelivered(exchanges);
+ throw e;
}
+ return processBatch(exchanges);
} finally {
lock.unlock();
}
}
- private int pollWithSqlQuery() throws Exception {
+ private void pollWithSqlQuery(Queue<Object> exchanges) throws Exception {
QueryOptions queryOptions = QueryOptions.queryOptions()
.scanConsistency(QueryScanConsistency.REQUEST_PLUS);
@@ -167,7 +183,7 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
}
String consumerProcessedStrategy =
endpoint.getConsumerProcessedStrategy();
- Queue<Object> exchanges = new ArrayDeque<>();
+ boolean removeOnCompletion =
"delete".equalsIgnoreCase(consumerProcessedStrategy);
for (JsonObject row : result.rowsAsObject()) {
String id = row.getString(SQL_DOCUMENT_ID_ALIAS);
@@ -177,6 +193,9 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
SQL_DOCUMENT_ID_ALIAS, SQL_DOCUMENT_ID_ALIAS);
continue;
}
+ if (removeOnCompletion && isInFlight(id)) {
+ continue;
+ }
Object doc;
if (endpoint.isFullDocument()) {
@@ -190,12 +209,8 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
exchange.getIn().setBody(doc);
exchange.getIn().setHeader(HEADER_ID, id);
- if ("delete".equalsIgnoreCase(consumerProcessedStrategy)) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Deleting doc with ID {}", id);
- }
- CouchbaseCollectionOperation.removeDocument(collection, id,
endpoint.getWriteQueryTimeout(),
- endpoint.getConsumerRetryPause());
+ if (removeOnCompletion) {
+ removeDocumentOnCompletion(exchange, id);
} else if ("filter".equalsIgnoreCase(consumerProcessedStrategy)) {
if (LOG.isTraceEnabled()) {
LOG.trace("Filtering out ID {}", id);
@@ -212,12 +227,10 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
exchanges.add(exchange);
}
-
- return processBatch(exchanges);
}
@SuppressWarnings("deprecation")
- private int pollWithView() throws Exception {
+ private void pollWithView(Queue<Object> exchanges) throws Exception {
ViewResult result = bucket.viewQuery(endpoint.getDesignDocumentName(),
endpoint.getViewName(), this.viewOptions);
// okay we have some response from CouchBase so lets mark the consumer
as ready
@@ -228,11 +241,14 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
}
String consumerProcessedStrategy =
endpoint.getConsumerProcessedStrategy();
+ boolean removeOnCompletion =
"delete".equalsIgnoreCase(consumerProcessedStrategy);
- Queue<Object> exchanges = new ArrayDeque<>();
for (ViewRow row : result.rows()) {
Object doc;
String id = row.id().get();
+ if (removeOnCompletion && isInFlight(id)) {
+ continue;
+ }
if (endpoint.isFullDocument()) {
doc = CouchbaseCollectionOperation.getDocument(collection, id,
endpoint.getQueryTimeout(),
endpoint.getConsumerRetryPause());
@@ -255,12 +271,8 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
exchange.getIn().setHeader(HEADER_DESIGN_DOCUMENT_NAME,
designDocumentName);
exchange.getIn().setHeader(HEADER_VIEWNAME, viewName);
- if ("delete".equalsIgnoreCase(consumerProcessedStrategy)) {
- if (LOG.isTraceEnabled()) {
- LOG.trace("Deleting doc with ID {}", id);
- }
- CouchbaseCollectionOperation.removeDocument(collection, id,
endpoint.getWriteQueryTimeout(),
- endpoint.getConsumerRetryPause());
+ if (removeOnCompletion) {
+ removeDocumentOnCompletion(exchange, id);
} else if ("filter".equalsIgnoreCase(consumerProcessedStrategy)) {
if (LOG.isTraceEnabled()) {
LOG.trace("Filtering out ID {}", id);
@@ -273,8 +285,87 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
logDetails(id, doc, key, designDocumentName, viewName, exchange);
exchanges.add(exchange);
}
+ }
+
+ /**
+ * Whether the exchange of this document, created by an earlier poll, has
not completed or failed yet.
+ * <p/>
+ * With {@code consumerProcessedStrategy=delete} the document is removed
only when its exchange completes. If the
+ * route hands the exchange over to another thread (for example {@code
seda} with
+ * {@code waitForTaskToComplete=Never}, or an asynchronous producer), that
can be after the next poll has read the
+ * same document again, which would deliver it twice. Such a document is
skipped until its exchange is done, as the
+ * in-progress repository of the file and aws2-s3 consumers does. A
skipped row does not count towards
+ * {@code maxMessagesPerPoll}, as no exchange is created for it.
+ * <p/>
+ * The other strategies leave the document in place, so reading it again
on the next poll is what they do anyway;
+ * they are not guarded. The guard is kept per consumer, it does not stop
another consumer, for example on another
+ * node, from reading the same document.
+ */
+ private boolean isInFlight(String id) {
+ // only the poll adds an ID, under the poll lock, so checking first
and adding later does not race
+ if (inFlight.contains(id)) {
+ LOG.trace("Skipping document with ID {} as its exchange is still
in progress", id);
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Removes the document once its exchange has been processed successfully.
+ * <p/>
+ * Removing it while the exchanges are built, before any of them is handed
to the route, loses the document when its
+ * exchange fails, or when it is never delivered at all because the batch
is cut short by {@code maxMessagesPerPoll}
+ * or by the consumer stopping. The on-completion runs only for an
exchange that went through the route and
+ * completed: a failed exchange keeps its document for the next poll, and
an undelivered one is not touched.
+ * <p/>
+ * The on-completion is handed over with the exchange, so that the
document is not removed before an asynchronous
+ * part of the route has processed it. Until then the document is marked
as in progress (see
+ * {@link #isInFlight(String)}), and it is released when the exchange
completes or fails, or by
+ * {@link #releaseUndelivered(Queue)} when the exchange is never handed to
the route.
+ */
+ private void removeDocumentOnCompletion(Exchange exchange, String id) {
+ inFlight.add(id);
+ exchange.getExchangeExtension().addOnCompletion(new
SynchronizationAdapter() {
+ @Override
+ public void onComplete(Exchange exchange) {
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Deleting doc with ID {}", id);
+ }
+ try {
+ CouchbaseCollectionOperation.removeDocument(collection,
id, endpoint.getWriteQueryTimeout(),
+ endpoint.getConsumerRetryPause());
+ } catch (Exception e) {
+ getExceptionHandler().handleException("Error removing
document with ID " + id, exchange, e);
+ } finally {
+ inFlight.remove(id);
+ }
+ }
- return processBatch(exchanges);
+ @Override
+ public void onFailure(Exchange exchange) {
+ // the document is kept, so a later poll may consume it again
+ inFlight.remove(id);
+ }
+ });
+ }
+
+ /**
+ * Releases exchanges that were created by a poll but never handed to the
route, and their documents.
+ * <p/>
+ * Nothing else will release these, and a pooled exchange that is never
released never returns to the pool. With
+ * {@code consumerProcessedStrategy=delete} their documents are left in
place, as the removal only runs on
+ * completion, and are no longer marked as in progress, so a later poll
picks them up.
+ */
+ private void releaseUndelivered(Queue<Object> exchanges) {
+ Object next;
+ while ((next = exchanges.poll()) != null) {
+ Exchange exchange = (Exchange) next;
+ String id = exchange.getIn().getHeader(HEADER_ID, String.class);
+ if (id != null) {
+ inFlight.remove(id);
+ }
+ releaseExchange(exchange, false);
+ }
}
@Override
@@ -297,16 +388,8 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
}
// Anything still queued was never handed to the route - the batch was
cut short because the consumer is
- // stopping, or the poll returned more rows than maxMessagesPerPoll.
Nothing else will release these, and a
- // pooled exchange that is never released never returns to the pool.
- //
- // Releasing them is not the same as handling them: with
consumerProcessedStrategy=delete the document was
- // already removed during the poll, above, so these rows are lost
rather than redelivered. That predates
- // this method and is not fixed here - see CAMEL-25221
- Exchange remaining;
- while ((remaining = (Exchange) exchanges.poll()) != null) {
- releaseExchange(remaining, false);
- }
+ // stopping, or the poll returned more rows than maxMessagesPerPoll
+ releaseUndelivered(exchanges);
return answer;
}
@@ -320,9 +403,7 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
* exchange had been delivered.
* <p/>
* The route's own error handler has already logged the exhausted failure
by this point, so this is not the only
- * record of it; what it adds is that the consumer no longer treats a
failed exchange as a delivered one. It matters
- * most with {@code consumerProcessedStrategy=delete}, where the document
is removed during the poll, before the
- * route runs, so a failure means the document is gone.
+ * record of it; what it adds is that the consumer no longer treats a
failed exchange as a delivered one.
*/
private void processExchange(Exchange exchange) {
try {
diff --git
a/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerProcessedStrategyTest.java
b/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerProcessedStrategyTest.java
new file mode 100644
index 000000000000..132e45cab608
--- /dev/null
+++
b/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerProcessedStrategyTest.java
@@ -0,0 +1,448 @@
+/*
+ * 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.couchbase;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.couchbase.client.java.Bucket;
+import com.couchbase.client.java.Cluster;
+import com.couchbase.client.java.ClusterOptions;
+import com.couchbase.client.java.Collection;
+import com.couchbase.client.java.Scope;
+import com.couchbase.client.java.json.JsonObject;
+import com.couchbase.client.java.kv.GetOptions;
+import com.couchbase.client.java.kv.RemoveOptions;
+import com.couchbase.client.java.query.QueryOptions;
+import com.couchbase.client.java.query.QueryResult;
+import com.couchbase.client.java.view.ViewOptions;
+import com.couchbase.client.java.view.ViewResult;
+import com.couchbase.client.java.view.ViewRow;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.ShutdownRunningTask;
+import org.apache.camel.builder.NotifyBuilder;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.seda.SedaEndpoint;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.ExceptionHandler;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.mockingDetails;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * With {@code consumerProcessedStrategy=delete} a document must be removed
only once its exchange has been processed
+ * successfully: a failed exchange, or a row that is never handed to the
route, keeps its document for the next poll.
+ * <p/>
+ * The consumer runs in a real route, so the exchange goes through a unit of
work, but its scheduler is not started and
+ * the test calls {@code poll()} itself against a mocked cluster.
+ */
+class CouchbaseConsumerProcessedStrategyTest {
+
+ private static final String URI =
"couchbase:http://localhost:8091?bucket=bucket&username=user&password=secret";
+ +
"&consumerProcessedStrategy=delete&startScheduler=false";
+
+ private DefaultCamelContext context;
+ private MockedStatic<Cluster> clusters;
+ private Bucket bucket;
+ private Scope scope;
+ private Collection collection;
+
+ @BeforeEach
+ void setUp() {
+ bucket = mock(Bucket.class);
+ scope = mock(Scope.class);
+ collection = mock(Collection.class);
+ when(bucket.defaultScope()).thenReturn(scope);
+ when(bucket.defaultCollection()).thenReturn(collection);
+ Cluster cluster = mock(Cluster.class);
+ when(cluster.bucket(anyString())).thenReturn(bucket);
+ clusters = mockStatic(Cluster.class);
+ clusters.when(() -> Cluster.connect(anyString(),
any(ClusterOptions.class))).thenReturn(cluster);
+ context = new DefaultCamelContext();
+ }
+
+ @AfterEach
+ void tearDown() {
+ // exchanges still queued for a started route would hold up the
graceful shutdown
+ if (context.hasEndpoint("seda://async") instanceof SedaEndpoint seda) {
+ seda.purgeQueue();
+ }
+ context.stop();
+ clusters.close();
+ }
+
+ private CouchbaseConsumer startRoute(String uri, Processor processor)
throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(uri).routeId("couchbase").process(processor);
+ }
+ });
+ context.start();
+ return (CouchbaseConsumer) context.getRoute("couchbase").getConsumer();
+ }
+
+ /**
+ * The couchbase route hands every exchange over to a {@code seda} queue
without waiting for it, so the
+ * on-completion that removes the document goes with the copy in the
queue. The route consuming that queue is
+ * started by the test when the handed over exchanges should complete, so
until then they are in flight.
+ */
+ private CouchbaseConsumer startHandoverRoutes(String uri, Processor
before, Processor async) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(uri).routeId("couchbase").process(before).to("seda:async?waitForTaskToComplete=Never");
+
from("seda:async").routeId("async").autoStartup(false).process(async);
+ }
+ });
+ context.start();
+ return (CouchbaseConsumer) context.getRoute("couchbase").getConsumer();
+ }
+
+ private CouchbaseConsumer startHandoverRoutes(String uri, Processor async)
throws Exception {
+ return startHandoverRoutes(uri, exchange -> {
+ }, async);
+ }
+
+ /** The IDs of the handed over exchanges still waiting in the queue, in
order. */
+ private List<Object> inFlightIds() {
+ return context.getEndpoint("seda:async",
SedaEndpoint.class).getQueue().stream()
+ .map(exchange ->
exchange.getMessage().getHeader(CouchbaseConstants.HEADER_ID))
+ .toList();
+ }
+
+ private void completeHandedOverExchanges(int count) throws Exception {
+ NotifyBuilder done = new
NotifyBuilder(context).fromRoute("async").whenDone(count).create();
+ context.getRouteController().startRoute("async");
+ // the on-completions run before the exchange is signalled as done
+ assertTrue(done.matches(10, TimeUnit.SECONDS), "the handed over
exchanges did not complete");
+ context.getRouteController().stopRoute("async");
+ }
+
+ private void queryReturns(String... ids) {
+ List<JsonObject> rows = new ArrayList<>();
+ for (String id : ids) {
+
rows.add(JsonObject.create().put(CouchbaseConsumer.SQL_DOCUMENT_ID_ALIAS,
id).put("name", id));
+ }
+ QueryResult result = mock(QueryResult.class);
+ when(result.rowsAsObject()).thenReturn(rows);
+ when(scope.query(anyString(),
any(QueryOptions.class))).thenReturn(result);
+ }
+
+ private void verifyRemoved(String id) {
+ verify(collection).remove(eq(id), any(RemoveOptions.class));
+ }
+
+ private void verifyNotRemoved(String id) {
+ verify(collection, never()).remove(eq(id), any(RemoveOptions.class));
+ }
+
+ private boolean removedYet() {
+ return mockingDetails(collection).getInvocations().stream()
+ .anyMatch(invocation ->
"remove".equals(invocation.getMethod().getName()));
+ }
+
+ @Test
+ void aProcessedDocumentIsRemovedAfterTheRouteHasRun() throws Exception {
+ queryReturns("doc-1");
+ List<Boolean> removedWhileRouting = new ArrayList<>();
+ CouchbaseConsumer consumer = startRoute(URI, exchange ->
removedWhileRouting.add(removedYet()));
+
+ assertEquals(1, consumer.poll());
+
+ assertEquals(List.of(false), removedWhileRouting, "the document must
still exist while the route processes it");
+ verifyRemoved("doc-1");
+ }
+
+ @Test
+ void aFailedExchangeKeepsItsDocument() throws Exception {
+ queryReturns("doc-1", "doc-2");
+ CouchbaseConsumer consumer = startRoute(URI, exchange -> {
+ if
("doc-1".equals(exchange.getMessage().getHeader(CouchbaseConstants.HEADER_ID)))
{
+ throw new IllegalStateException("the route blew up");
+ }
+ });
+
+ consumer.poll();
+
+ verifyNotRemoved("doc-1");
+ verifyRemoved("doc-2");
+ }
+
+ @Test
+ void rowsBeyondMaxMessagesPerPollKeepTheirDocuments() throws Exception {
+ queryReturns("doc-1", "doc-2", "doc-3");
+ CouchbaseConsumer consumer = startRoute(URI, exchange -> {
+ });
+ consumer.setMaxMessagesPerPoll(1);
+
+ consumer.poll();
+
+ verifyRemoved("doc-1");
+ verifyNotRemoved("doc-2");
+ verifyNotRemoved("doc-3");
+ }
+
+ @Test
+ void rowsLeftWhenTheConsumerStopsMidBatchKeepTheirDocuments() throws
Exception {
+ queryReturns("doc-1", "doc-2", "doc-3");
+ AtomicReference<CouchbaseConsumer> consumerRef = new
AtomicReference<>();
+ // what the shutdown strategy does to a consumer that should only
complete its current task
+ CouchbaseConsumer consumer = startRoute(URI,
+ exchange ->
consumerRef.get().deferShutdown(ShutdownRunningTask.CompleteCurrentTaskOnly));
+ consumerRef.set(consumer);
+
+ consumer.poll();
+
+ verifyRemoved("doc-1");
+ verifyNotRemoved("doc-2");
+ verifyNotRemoved("doc-3");
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void aFailedExchangeKeepsItsDocumentWhenConsumingAView() throws Exception {
+ ViewRow failing = viewRow("doc-1");
+ ViewRow succeeding = viewRow("doc-2");
+ ViewResult result = mock(ViewResult.class);
+ when(result.rows()).thenReturn(List.of(failing, succeeding));
+ when(bucket.viewQuery(anyString(), anyString(),
any(ViewOptions.class))).thenReturn(result);
+
+ CouchbaseConsumer consumer = startRoute(URI +
"&useView=true&designDocumentName=dd&viewName=view", exchange -> {
+ if
("doc-1".equals(exchange.getMessage().getHeader(CouchbaseConstants.HEADER_ID)))
{
+ throw new IllegalStateException("the route blew up");
+ }
+ });
+
+ consumer.poll();
+
+ verifyNotRemoved("doc-1");
+ verifyRemoved("doc-2");
+ }
+
+ @Test
+ void aFailedRemovalIsReportedToTheExceptionHandler() throws Exception {
+ queryReturns("doc-1");
+ RuntimeException failure = new IllegalStateException("cannot remove");
+ when(collection.remove(eq("doc-1"),
any(RemoveOptions.class))).thenThrow(failure);
+ CouchbaseConsumer consumer = startRoute(URI, exchange -> {
+ });
+ AtomicReference<Throwable> reported = new AtomicReference<>();
+ consumer.setExceptionHandler(new CapturingExceptionHandler(reported));
+
+ consumer.poll();
+
+ verify(collection, times(1)).remove(eq("doc-1"),
any(RemoveOptions.class));
+ assertSame(failure, reported.get());
+ }
+
+ @Test
+ void otherStrategiesDoNotRemoveDocuments() throws Exception {
+ queryReturns("doc-1");
+ CouchbaseConsumer consumer = startRoute(URI.replace("=delete",
"=none"), exchange -> {
+ });
+
+ consumer.poll();
+
+ assertFalse(removedYet());
+ }
+
+ @Test
+ void aDocumentWhoseHandedOverExchangeIsInFlightIsNotDeliveredAgain()
throws Exception {
+ queryReturns("doc-1");
+ CouchbaseConsumer consumer = startHandoverRoutes(URI, exchange -> {
+ });
+
+ assertEquals(1, consumer.poll());
+ assertEquals(0, consumer.poll(), "the document must not be delivered
again while its exchange is in flight");
+
+ assertEquals(List.of("doc-1"), inFlightIds());
+ verifyNotRemoved("doc-1");
+ }
+
+ @Test
+ void
aDocumentWhoseHandedOverExchangeIsInFlightIsNotDeliveredAgainWhenConsumingAView()
throws Exception {
+ ViewRow row = viewRow("doc-1");
+ ViewResult result = mock(ViewResult.class);
+ when(result.rows()).thenReturn(List.of(row));
+ when(bucket.viewQuery(anyString(), anyString(),
any(ViewOptions.class))).thenReturn(result);
+ CouchbaseConsumer consumer = startHandoverRoutes(URI +
"&useView=true&designDocumentName=dd&viewName=view",
+ exchange -> {
+ });
+
+ assertEquals(1, consumer.poll());
+ assertEquals(0, consumer.poll(), "the document must not be delivered
again while its exchange is in flight");
+
+ assertEquals(List.of("doc-1"), inFlightIds());
+ }
+
+ @Test
+ void aHandedOverExchangeRemovesItsDocumentOnceWhenItCompletes() throws
Exception {
+ queryReturns("doc-1");
+ List<Object> processed = new CopyOnWriteArrayList<>();
+ CouchbaseConsumer consumer = startHandoverRoutes(URI,
+ exchange ->
processed.add(exchange.getMessage().getHeader(CouchbaseConstants.HEADER_ID)));
+
+ consumer.poll();
+ consumer.poll();
+ verifyNotRemoved("doc-1");
+
+ completeHandedOverExchanges(1);
+
+ assertEquals(List.of("doc-1"), processed);
+ verify(collection, times(1)).remove(eq("doc-1"),
any(RemoveOptions.class));
+ }
+
+ @Test
+ void aFailedHandedOverExchangeReleasesItsDocumentForTheNextPoll() throws
Exception {
+ queryReturns("doc-1");
+ CouchbaseConsumer consumer = startHandoverRoutes(URI, exchange -> {
+ throw new IllegalStateException("the route blew up");
+ });
+
+ assertEquals(1, consumer.poll());
+ assertEquals(0, consumer.poll());
+
+ completeHandedOverExchanges(1);
+
+ verifyNotRemoved("doc-1");
+ assertEquals(1, consumer.poll(), "a failed document must be consumed
again by the next poll");
+ assertEquals(List.of("doc-1"), inFlightIds());
+ }
+
+ @Test
+ void rowsBeyondMaxMessagesPerPollAreNotLeftInFlight() throws Exception {
+ queryReturns("doc-1", "doc-2", "doc-3");
+ CouchbaseConsumer consumer = startHandoverRoutes(URI, exchange -> {
+ });
+ consumer.setMaxMessagesPerPoll(1);
+
+ consumer.poll();
+ // doc-1 is still in flight and skipped, it does not take the place of
doc-2
+ consumer.poll();
+ consumer.poll();
+
+ assertEquals(List.of("doc-1", "doc-2", "doc-3"), inFlightIds());
+ }
+
+ @Test
+ void rowsLeftWhenTheConsumerStopsMidBatchAreNotLeftInFlight() throws
Exception {
+ queryReturns("doc-1", "doc-2", "doc-3");
+ AtomicReference<CouchbaseConsumer> consumerRef = new
AtomicReference<>();
+ AtomicBoolean stopping = new AtomicBoolean(true);
+ // what the shutdown strategy does to a consumer that should only
complete its current task
+ CouchbaseConsumer consumer = startHandoverRoutes(URI, exchange -> {
+ if (stopping.get()) {
+
consumerRef.get().deferShutdown(ShutdownRunningTask.CompleteCurrentTaskOnly);
+ }
+ }, exchange -> {
+ });
+ consumerRef.set(consumer);
+
+ consumer.poll();
+ assertEquals(List.of("doc-1"), inFlightIds());
+
+ // the consumer carries on, as after a shutdown that timed out and was
cancelled
+ stopping.set(false);
+ consumer.deferShutdown(ShutdownRunningTask.CompleteAllTasks);
+ consumer.poll();
+
+ assertEquals(List.of("doc-1", "doc-2", "doc-3"), inFlightIds());
+ }
+
+ @Test
+ void aPollThatFailsPartWayDoesNotLeaveItsRowsInFlight() throws Exception {
+ queryReturns("doc-1", "doc-2");
+ RuntimeException failure = new IllegalStateException("cannot get
document");
+ when(collection.get(eq("doc-2"),
any(GetOptions.class))).thenThrow(failure).thenReturn(null);
+ CouchbaseConsumer consumer = startHandoverRoutes(URI, exchange -> {
+ });
+
+ assertSame(failure, assertThrows(IllegalStateException.class,
consumer::poll));
+ assertEquals(List.of(), inFlightIds());
+
+ assertEquals(2, consumer.poll());
+ assertEquals(List.of("doc-1", "doc-2"), inFlightIds());
+ }
+
+ @Test
+ void otherStrategiesStillDeliverADocumentWhoseExchangeIsInFlight() throws
Exception {
+ queryReturns("doc-1");
+ CouchbaseConsumer consumer =
startHandoverRoutes(URI.replace("=delete", "=none"), exchange -> {
+ });
+
+ assertEquals(1, consumer.poll());
+ assertEquals(1, consumer.poll());
+
+ assertEquals(List.of("doc-1", "doc-1"), inFlightIds());
+ }
+
+ private static ViewRow viewRow(String id) {
+ ViewRow row = mock(ViewRow.class);
+ when(row.id()).thenReturn(Optional.of(id));
+ when(row.valueAs(Object.class)).thenReturn(Optional.of(id));
+ when(row.keyAs(String.class)).thenReturn(Optional.of(id));
+ return row;
+ }
+
+ private static final class CapturingExceptionHandler implements
ExceptionHandler {
+
+ private final AtomicReference<Throwable> captured;
+
+ private CapturingExceptionHandler(AtomicReference<Throwable> captured)
{
+ this.captured = captured;
+ }
+
+ @Override
+ public void handleException(Throwable exception) {
+ captured.set(exception);
+ }
+
+ @Override
+ public void handleException(String message, Throwable exception) {
+ captured.set(exception);
+ }
+
+ @Override
+ public void handleException(String message, Exchange exchange,
Throwable exception) {
+ captured.set(exception);
+ }
+ }
+}
diff --git
a/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerTest.java
b/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerTest.java
index bf3ffe78e972..16a89ae8c899 100644
---
a/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerTest.java
+++
b/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerTest.java
@@ -97,8 +97,7 @@ class CouchbaseConsumerTest {
/**
* A failing route does not throw out of {@code process()} - the failure
is left on the exchange - so a consumer
- * that only catches never learns about the common case. With {@code
consumerProcessedStrategy=delete} the document
- * is already gone by then, so an unreported failure loses the message
outright.
+ * that only catches never learns about the common case.
*/
@Test
void aRouteFailureIsReportedToTheExceptionHandler() throws Exception {
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 46a3d152e3c9..b95f0c3b64ab 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
@@ -588,8 +588,28 @@ poll as if the exchange had been delivered.
This is not a change to error handling. An exchange that failed while being
routed has already passed through the
route's error handler, which logs the exhausted failure, and
`BridgeExceptionHandlerToErrorHandler` deliberately
falls back to its logging handler for such an exchange rather than bridging it
a second time. What changes is that
-the consumer itself no longer stays silent, and that with
`consumerProcessedStrategy=delete` the loss is now visible:
-the document is removed during the poll, before the route runs.
+the consumer itself no longer stays silent.
+
+==== consumerProcessedStrategy=delete
+
+With `consumerProcessedStrategy=delete` the consumer now removes a document
once its exchange has completed
+successfully. It used to remove every document of a poll while building the
exchanges, before any of them was handed
+to the route, so the document was gone even when its exchange failed, and also
when it was never handed to the route
+at all, because the consumer was stopping or the poll returned more rows than
`maxMessagesPerPoll`.
+
+A document whose exchange fails, and a document that was not handed to the
route, are now kept and consumed again by a
+later poll. A route that fails for the same document every time therefore
receives it on every poll; handle the
+failure in the route (for example with a dead letter channel or a handled
`onException`) so that the exchange
+completes and the document is removed. While the route runs, the document is
still in the bucket. A failure to remove
+the document is reported through the consumer's exception handler.
+
+If the route hands the exchange over to another thread (for example `seda`
with `waitForTaskToComplete=Never`, or an
+asynchronous producer), the document is removed when that exchange completes,
which can be after the next poll. Until
+its exchange has completed or failed, a document is skipped by later polls of
the same consumer, so it is not delivered
+twice, in the same way as the in-progress repository of the file and aws2-s3
consumers. A skipped document does not
+count towards `maxMessagesPerPoll`. This only applies to the same consumer:
another consumer reading the same bucket,
+for example on another node, can still read the document before it is removed.
With the other strategies the document
+stays in the bucket, so later polls read it again as before.
=== camel-smpp - received exchanges now come from the consumer
