(camel) branch main updated: CAMEL-25290: camel-couchdb - the consumer must apply updates=false, and move past the changes it skips (#27322)

[da...@apache.org]

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 c866afdce9d6 CAMEL-25290: camel-couchdb - the consumer must apply updates=false, and move past the changes it skips (#27322)
c866afdce9d6 is described below

commit c866afdce9d6e227610dc0f9d92ed4ed0ac8acea
Author: allthingssecurity <49...@users.noreply.github.com>
AuthorDate: Sun Oct 4 12:33:40 2026 +0530

    CAMEL-25290: camel-couchdb - the consumer must apply updates=false, and move past the changes it skips (#27322)
    
    Co-Authored-By: Claude Opus 5.5 <no...@anthropic.com>
---
 .../camel/component/couchdb/CouchDbConsumer.java   |  16 ++-
 .../couchdb/CouchDbConsumerFilterTest.java         | 110 +++++++++++++++++++++
 2 files changed, 117 insertions(+), 9 deletions(-)

diff --git a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/CouchDbConsumer.java b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/CouchDbConsumer.java
index 53a4e379c011..6be0b6483665 100644
--- a/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/CouchDbConsumer.java
+++ b/components/camel-couchdb/src/main/java/org/apache/camel/component/couchdb/CouchDbConsumer.java
@@ -78,19 +78,17 @@ public class CouchDbConsumer extends ScheduledBatchPollingConsumer implements Re
                 = couchClient.pollChanges(endpoint.getStyle(), since, endpoint.getHeartbeat(), getMaxMessagesPerPoll());
 
         for (ChangesResultItem changesResultItem : changesResultResponse.getResult().getResults()) {
-            if (changesResultItem.isDeleted() != null) {
-                if (changesResultItem.isDeleted() && !endpoint.isDeletes()) {
-                    continue;
-                }
-                if (!changesResultItem.isDeleted() && !endpoint.isUpdates()) {
-                    continue;
-                }
+            // CouchDB sets deleted only on the changes of deleted documents
+            boolean deleted = Boolean.TRUE.equals(changesResultItem.isDeleted());
+            if (deleted ? !endpoint.isDeletes() : !endpoint.isUpdates()) {
+                // move past the skipped change, or a page of skipped changes is polled again forever
+                since = changesResultItem.getSeq();
+                continue;
             }
 
             lastSequence = changesResultItem.getSeq();
 
-            Exchange exchange = this.createExchange(lastSequence, changesResultItem.getId(), changesResultItem,
-                    changesResultItem.isDeleted() == null ? false : changesResultItem.isDeleted());
+            Exchange exchange = this.createExchange(lastSequence, changesResultItem.getId(), changesResultItem, deleted);
 
             if (LOG.isTraceEnabled()) {
                 LOG.trace("Created exchange [exchange={}, _id={}, seq={}", exchange, changesResultItem.getId(), lastSequence);
diff --git a/components/camel-couchdb/src/test/java/org/apache/camel/component/couchdb/CouchDbConsumerFilterTest.java b/components/camel-couchdb/src/test/java/org/apache/camel/component/couchdb/CouchDbConsumerFilterTest.java
new file mode 100644
index 000000000000..48091adaab5c
--- /dev/null
+++ b/components/camel-couchdb/src/test/java/org/apache/camel/component/couchdb/CouchDbConsumerFilterTest.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.component.couchdb;
+
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import com.ibm.cloud.cloudant.v1.model.ChangesResult;
+import com.ibm.cloud.sdk.core.http.Response;
+import com.ibm.cloud.sdk.core.util.GsonSingleton;
+import org.apache.camel.Exchange;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * The consumer must skip the changes that its options filter out, and move past them.
+ */
+public class CouchDbConsumerFilterTest extends CamelTestSupport {
+
+    // CouchDB sets "deleted" only on the changes of deleted documents
+    private static final String UPDATE = "{\"seq\":\"%s\",\"id\":\"%s\",\"changes\":[{\"rev\":\"1-a\"}]}";
+    private static final String DELETE = "{\"seq\":\"%s\",\"id\":\"%s\",\"changes\":[{\"rev\":\"2-b\"}],\"deleted\":true}";
+
+    private final CouchDbClientWrapper client = mock(CouchDbClientWrapper.class);
+    private final List<Exchange> received = new CopyOnWriteArrayList<>();
+
+    @Test
+    void testDeletesFalseMovesPastDeletedDocuments() throws Exception {
+        CouchDbConsumer consumer = createConsumer("deletes=false");
+        // a whole page of deleted documents, which this consumer does not publish
+        Response<ChangesResult> page = changes(DELETE.formatted("1-x", "doc1"), DELETE.formatted("2-x", "doc2"),
+                DELETE.formatted("3-x", "doc3"));
+        when(client.pollChanges(anyString(), anyString(), anyLong(), anyLong())).thenReturn(page);
+
+        consumer.poll();
+        consumer.poll();
+
+        assertEquals(0, received.size());
+        // the second poll must ask for the changes after the last one seen, not for the same page again
+        verify(client).pollChanges(anyString(), eq("3-x"), anyLong(), anyLong());
+    }
+
+    @Test
+    void testUpdatesFalseSkipsUpdatedDocuments() throws Exception {
+        CouchDbConsumer consumer = createConsumer("updates=false");
+        Response<ChangesResult> page = changes(UPDATE.formatted("1-x", "doc1"), DELETE.formatted("2-x", "doc2"));
+        when(client.pollChanges(anyString(), anyString(), anyLong(), anyLong())).thenReturn(page);
+
+        consumer.poll();
+
+        assertEquals(1, received.size());
+        assertEquals("doc2", received.get(0).getIn().getHeader(CouchDbConstants.HEADER_DOC_ID));
+        assertEquals("DELETE", received.get(0).getIn().getHeader(CouchDbConstants.HEADER_METHOD));
+    }
+
+    @Test
+    void testDefaultPublishesUpdatesAndDeletes() throws Exception {
+        CouchDbConsumer consumer = createConsumer("updates=true");
+        Response<ChangesResult> page = changes(UPDATE.formatted("1-x", "doc1"), DELETE.formatted("2-x", "doc2"));
+        when(client.pollChanges(anyString(), anyString(), anyLong(), anyLong())).thenReturn(page);
+
+        consumer.poll();
+        consumer.poll();
+
+        assertEquals(4, received.size());
+        assertEquals("UPDATE", received.get(0).getIn().getHeader(CouchDbConstants.HEADER_METHOD));
+        assertEquals("DELETE", received.get(1).getIn().getHeader(CouchDbConstants.HEADER_METHOD));
+        verify(client).pollChanges(anyString(), eq("2-x"), anyLong(), anyLong());
+    }
+
+    private CouchDbConsumer createConsumer(String options) throws Exception {
+        when(client.getLatestUpdateSequence()).thenReturn("0");
+        CouchDbEndpoint endpoint = context.getEndpoint("couchdb:http://localhost:5984/camel?" + options, CouchDbEndpoint.class);
+        CouchDbConsumer consumer = new CouchDbConsumer(endpoint, client, received::add);
+        consumer.setMaxMessagesPerPoll(endpoint.getMaxMessagesPerPoll());
+        consumer.init();
+        return consumer;
+    }
+
+    @SuppressWarnings("unchecked")
+    private static Response<ChangesResult> changes(String... items) {
+        String json = "{\"last_seq\":\"9-x\",\"pending\":0,\"results\":[" + String.join(",", items) + "]}";
+        ChangesResult result = GsonSingleton.getGson().fromJson(json, ChangesResult.class);
+        Response<ChangesResult> response = mock(Response.class);
+        when(response.getResult()).thenReturn(result);
+        return response;
+    }
+}