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 d61386813417 CAMEL-25361: camel-couchbase - use the view value as the
body and document the fullDocument default (#27588)
d61386813417 is described below
commit d61386813417f5917ea4bae1b3282fe583273a1f
Author: allthingssecurity <[email protected]>
AuthorDate: Sat Oct 10 01:20:50 2026 +0530
CAMEL-25361: camel-couchbase - use the view value as the body and document
the fullDocument default (#27588)
With useView=true and fullDocument=false the consumer body was the Optional
returned by the Couchbase SDK 3 instead of the view value. The body is now
the
value, or null when the view emitted none. The fullDocument option is now
documented with its actual default (true). The SQL++ path is unchanged.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../apache/camel/catalog/components/couchbase.json | 2 +-
.../camel/component/couchbase/couchbase.json | 2 +-
.../component/couchbase/CouchbaseConsumer.java | 3 +-
.../component/couchbase/CouchbaseEndpoint.java | 2 +-
.../couchbase/CouchbaseConsumerBodyTest.java | 174 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 13 ++
.../dsl/CouchbaseEndpointBuilderFactory.java | 4 +-
7 files changed, 194 insertions(+), 6 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/couchbase.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/couchbase.json
index 3f962e5a755d..e3153aa368a1 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/couchbase.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/couchbase.json
@@ -50,7 +50,7 @@
"consumerRetryPause": { "index": 8, "kind": "parameter", "displayName":
"Consumer Retry Pause", "group": "consumer", "label": "consumer", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 5000, "description": "Define the
consumer retry pause between different attempts" },
"descending": { "index": 9, "kind": "parameter", "displayName":
"Descending", "group": "consumer", "label": "consumer", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "description": "Define if this
operation is descending or not" },
"designDocumentName": { "index": 10, "kind": "parameter", "displayName":
"Design Document Name", "group": "consumer", "label": "consumer", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": true,
"autowired": false, "secret": false, "defaultValue": "beer", "description":
"The design document name to use. Deprecated: use the statement option with SQL
queries instead." },
- "fullDocument": { "index": 11, "kind": "parameter", "displayName": "Full
Document", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "If true consumer will
return complete document instead data defined in view" },
+ "fullDocument": { "index": 11, "kind": "parameter", "displayName": "Full
Document", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": true, "description": "If true consumer will
return complete document instead data defined in view" },
"limit": { "index": 12, "kind": "parameter", "displayName": "Limit",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": -1, "description": "The output limit to use" },
"rangeEndKey": { "index": 13, "kind": "parameter", "displayName": "Range
End Key", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "description": "Define a range for the end key" },
"rangeStartKey": { "index": 14, "kind": "parameter", "displayName": "Range
Start Key", "group": "consumer", "label": "consumer", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "description": "Define a range for the
start key" },
diff --git
a/components/camel-couchbase/src/generated/resources/META-INF/org/apache/camel/component/couchbase/couchbase.json
b/components/camel-couchbase/src/generated/resources/META-INF/org/apache/camel/component/couchbase/couchbase.json
index 3f962e5a755d..e3153aa368a1 100644
---
a/components/camel-couchbase/src/generated/resources/META-INF/org/apache/camel/component/couchbase/couchbase.json
+++
b/components/camel-couchbase/src/generated/resources/META-INF/org/apache/camel/component/couchbase/couchbase.json
@@ -50,7 +50,7 @@
"consumerRetryPause": { "index": 8, "kind": "parameter", "displayName":
"Consumer Retry Pause", "group": "consumer", "label": "consumer", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 5000, "description": "Define the
consumer retry pause between different attempts" },
"descending": { "index": 9, "kind": "parameter", "displayName":
"Descending", "group": "consumer", "label": "consumer", "required": false,
"type": "boolean", "javaType": "boolean", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": false, "description": "Define if this
operation is descending or not" },
"designDocumentName": { "index": 10, "kind": "parameter", "displayName":
"Design Document Name", "group": "consumer", "label": "consumer", "required":
false, "type": "string", "javaType": "java.lang.String", "deprecated": true,
"autowired": false, "secret": false, "defaultValue": "beer", "description":
"The design document name to use. Deprecated: use the statement option with SQL
queries instead." },
- "fullDocument": { "index": 11, "kind": "parameter", "displayName": "Full
Document", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": false, "description": "If true consumer will
return complete document instead data defined in view" },
+ "fullDocument": { "index": 11, "kind": "parameter", "displayName": "Full
Document", "group": "consumer", "label": "consumer", "required": false, "type":
"boolean", "javaType": "boolean", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": true, "description": "If true consumer will
return complete document instead data defined in view" },
"limit": { "index": 12, "kind": "parameter", "displayName": "Limit",
"group": "consumer", "label": "consumer", "required": false, "type": "integer",
"javaType": "int", "deprecated": false, "autowired": false, "secret": false,
"defaultValue": -1, "description": "The output limit to use" },
"rangeEndKey": { "index": 13, "kind": "parameter", "displayName": "Range
End Key", "group": "consumer", "label": "consumer", "required": false, "type":
"string", "javaType": "java.lang.String", "deprecated": false, "autowired":
false, "secret": false, "description": "Define a range for the end key" },
"rangeStartKey": { "index": 14, "kind": "parameter", "displayName": "Range
Start Key", "group": "consumer", "label": "consumer", "required": false,
"type": "string", "javaType": "java.lang.String", "deprecated": false,
"autowired": false, "secret": false, "description": "Define a range for the
start key" },
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 897268a3b57f..09fba1970d87 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
@@ -253,7 +253,8 @@ public class CouchbaseConsumer extends
ScheduledBatchPollingConsumer implements
doc = CouchbaseCollectionOperation.getDocument(collection, id,
endpoint.getQueryTimeout(),
endpoint.getConsumerRetryPause());
} else {
- doc = row.valueAs(Object.class);
+ // the SDK returns the value as an Optional, which is empty
when the view emitted no value (null)
+ doc = row.valueAs(Object.class).orElse(null);
}
// Use String.class instead of the shaded JsonNode class to avoid
conflicts
diff --git
a/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java
b/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java
index 5ca1a7f9b9e3..d83117e80425 100644
---
a/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java
+++
b/components/camel-couchbase/src/main/java/org/apache/camel/component/couchbase/CouchbaseEndpoint.java
@@ -149,7 +149,7 @@ public class CouchbaseEndpoint extends
ScheduledPollEndpoint implements Endpoint
private String rangeStartKey;
@UriParam(label = "consumer")
private String rangeEndKey = "";
- @UriParam(label = "consumer", defaultValue = "false")
+ @UriParam(label = "consumer", defaultValue = "true")
private boolean fullDocument = true;
@UriParam(label = "consumer", defaultValue = "5000")
private int consumerRetryPause = DEFAULT_PAUSE_BETWEEN_RETRIES;
diff --git
a/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerBodyTest.java
b/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerBodyTest.java
new file mode 100644
index 000000000000..b92371521b61
--- /dev/null
+++
b/components/camel-couchbase/src/test/java/org/apache/camel/component/couchbase/CouchbaseConsumerBodyTest.java
@@ -0,0 +1,174 @@
+/*
+ * 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.Map;
+import java.util.Optional;
+
+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.GetResult;
+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.builder.RouteBuilder;
+import org.apache.camel.impl.DefaultCamelContext;
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+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.when;
+
+/**
+ * The message body the consumer creates for a row, with and without {@code
fullDocument}.
+ * <p/>
+ * The scheduler of the consumer is not started; the test calls {@code poll()}
itself against a mocked cluster.
+ */
+class CouchbaseConsumerBodyTest {
+
+ private static final String URI =
"couchbase:http://localhost:8091?bucket=bucket&username=user&password=secret";
+ + "&startScheduler=false";
+ private static final String VIEW =
"&useView=true&designDocumentName=dd&viewName=view";
+
+ private DefaultCamelContext context;
+ private MockedStatic<Cluster> clusters;
+ private Bucket bucket;
+ private Scope scope;
+ private Collection collection;
+ private final List<Object> bodies = new ArrayList<>();
+
+ @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() {
+ context.stop();
+ clusters.close();
+ }
+
+ private CouchbaseConsumer startRoute(String uri) throws Exception {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from(uri).routeId("couchbase").process(exchange ->
bodies.add(exchange.getMessage().getBody()));
+ }
+ });
+ context.start();
+ return (CouchbaseConsumer) context.getRoute("couchbase").getConsumer();
+ }
+
+ @SuppressWarnings("deprecation")
+ private void viewReturns(ViewRow... rows) {
+ ViewResult result = mock(ViewResult.class);
+ when(result.rows()).thenReturn(List.of(rows));
+ when(bucket.viewQuery(anyString(), anyString(),
any(ViewOptions.class))).thenReturn(result);
+ }
+
+ private void queryReturns(JsonObject... rows) {
+ QueryResult result = mock(QueryResult.class);
+ when(result.rowsAsObject()).thenReturn(List.of(rows));
+ when(scope.query(anyString(),
any(QueryOptions.class))).thenReturn(result);
+ }
+
+ /** A view row as the SDK returns it: the value is an {@code Optional},
empty when the view emitted null. */
+ private static ViewRow viewRow(String id, Object value) {
+ ViewRow row = mock(ViewRow.class);
+ when(row.id()).thenReturn(Optional.of(id));
+ // pins the valueAs(Class) and keyAs(Class) overloads that
CouchbaseConsumer calls: if the consumer moves to
+ // another overload such as valueAs(TypeRef), stub that one here too,
as an unstubbed call returns null and the
+ // tests would fail with a NullPointerException instead of a clear
assertion
+ when(row.valueAs(Object.class)).thenReturn(Optional.ofNullable(value));
+ when(row.keyAs(String.class)).thenReturn(Optional.of(id));
+ return row;
+ }
+
+ @Test
+ void theValueEmittedByTheViewIsTheBodyWithoutFullDocument() throws
Exception {
+ Map<String, Object> value = Map.of("name", "Alice");
+ viewReturns(viewRow("doc-1", value), viewRow("doc-2", "Bob"));
+ CouchbaseConsumer consumer = startRoute(URI + VIEW +
"&fullDocument=false");
+
+ assertEquals(2, consumer.poll());
+
+ assertEquals(List.of(value, "Bob"), bodies, "the body must be the
value of the view row, not an Optional");
+ }
+
+ @Test
+ void theBodyIsNullWhenTheViewEmitsNoValue() throws Exception {
+ viewReturns(viewRow("doc-1", null));
+ CouchbaseConsumer consumer = startRoute(URI + VIEW +
"&fullDocument=false");
+
+ assertEquals(1, consumer.poll());
+
+ assertEquals(1, bodies.size());
+ assertNull(bodies.get(0), "the body must be null when the view emitted
null, not an empty Optional");
+ }
+
+ @Test
+ void theQueryRowIsTheBodyWithoutFullDocument() throws Exception {
+ JsonObject row =
JsonObject.create().put(CouchbaseConsumer.SQL_DOCUMENT_ID_ALIAS,
"doc-1").put("name", "Alice");
+ queryReturns(row);
+ CouchbaseConsumer consumer = startRoute(URI + "&fullDocument=false");
+
+ assertEquals(1, consumer.poll());
+
+ assertEquals(List.of(row.toString()), bodies, "the body must be the
SQL++ row as a JSON string");
+ }
+
+ @Test
+ void theFullDocumentIsTheBodyByDefault() throws Exception {
+
queryReturns(JsonObject.create().put(CouchbaseConsumer.SQL_DOCUMENT_ID_ALIAS,
"doc-1").put("name", "Alice"));
+ GetResult document = mock(GetResult.class);
+ when(collection.get(eq("doc-1"),
any(GetOptions.class))).thenReturn(document);
+ // fullDocument is not set: it defaults to true, so the consumer
fetches the document by its ID
+ CouchbaseConsumer consumer = startRoute(URI);
+
+ assertEquals(1, consumer.poll());
+
+ assertEquals(1, bodies.size());
+ assertSame(document, bodies.get(0), "the body must be the document
fetched by the ID of the row");
+ }
+}
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 adfdf36b2044..6cb1f2aad4a6 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
@@ -679,6 +679,12 @@ route's error handler, which logs the exhausted failure,
and `BridgeExceptionHan
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.
+==== useView=true with fullDocument=false
+
+With `useView=true` and `fullDocument=false` the message body is now the value
the view emitted for the row, or `null`
+when the view emitted no value. It used to be a `java.util.Optional` holding
that value. A route that unwrapped the
+`Optional` itself must now use the body directly.
+
==== consumerProcessedStrategy=delete
With `consumerProcessedStrategy=delete` the consumer now removes a document
once its exchange has completed
@@ -700,6 +706,13 @@ count towards `maxMessagesPerPoll`. This only applies to
the same consumer: anot
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.
+A view or SQL++ query that returns several rows for one document (for example
a view that emits one row per element of
+an array in the document) now delivers only the first of these rows with
`consumerProcessedStrategy=delete`: the
+document is in progress from its first row on, so its other rows are skipped,
and once that exchange has completed the
+document is removed. Previously the first row removed the document while the
poll was built, and the next row of the
+same document failed to read or remove it, as it no longer existed, so the
whole poll failed. With the other
+strategies every row is delivered as before.
+
=== camel-smpp - received exchanges now come from the consumer
Exchanges for received `deliver_sm` and `data_sm` messages are now created
through
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CouchbaseEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CouchbaseEndpointBuilderFactory.java
index 1daf1548b64a..9b5601c73bee 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CouchbaseEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/CouchbaseEndpointBuilderFactory.java
@@ -199,7 +199,7 @@ public interface CouchbaseEndpointBuilderFactory {
*
* The option is a: <code>boolean</code> type.
*
- * Default: false
+ * Default: true
* Group: consumer
*
* @param fullDocument the value to set
@@ -215,7 +215,7 @@ public interface CouchbaseEndpointBuilderFactory {
*
* The option will be converted to a <code>boolean</code> type.
*
- * Default: false
+ * Default: true
* Group: consumer
*
* @param fullDocument the value to set
