From 0e030c8b40dceddce53fa25b8954d1d04f067086 Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Mon, 9 Nov 2026 14:34:38 +0000 Subject: [PATCH 1/3] CAMEL-23626: Remove deprecated MapReduce View from couchbase-log example Replace useView=true + DesignDocument/View setup with the SQL++ (N1QL) consumer mode, which is the default (useView=false). The test now creates a primary index on the bucket so the auto-generated SQL++ query works. --- .../apache/camel/example/MyRouteBuilder.java | 2 +- .../apache/camel/example/CouchbaseTest.java | 22 +++++-------------- 2 files changed, 7 insertions(+), 17 deletions(-) diff --git a/couchbase-log/src/main/java/org/apache/camel/example/MyRouteBuilder.java b/couchbase-log/src/main/java/org/apache/camel/example/MyRouteBuilder.java index a828dd2bd..19c4268e2 100644 --- a/couchbase-log/src/main/java/org/apache/camel/example/MyRouteBuilder.java +++ b/couchbase-log/src/main/java/org/apache/camel/example/MyRouteBuilder.java @@ -23,7 +23,7 @@ public class MyRouteBuilder extends RouteBuilder { @Override public void configure() throws Exception { - from("couchbase:http://{{couchbase.host}}:{{couchbase.port}}?bucket={{couchbase.bucket}}&username={{couchbase.username}}&password={{couchbase.password}}&consumerProcessedStrategy={{couchbase.consumerProcessedStrategy}}&useView=true&connectionString={{couchbase.connectionString}}") + from("couchbase:http://{{couchbase.host}}:{{couchbase.port}}?bucket={{couchbase.bucket}}&username={{couchbase.username}}&password={{couchbase.password}}&consumerProcessedStrategy={{couchbase.consumerProcessedStrategy}}&connectionString={{couchbase.connectionString}}") .to("log:info"); } diff --git a/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java b/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java index 8b320677b..e56ef99f0 100644 --- a/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java +++ b/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java @@ -17,7 +17,6 @@ package org.apache.camel.example; import java.time.Duration; -import java.util.Collections; import java.util.Properties; import java.util.concurrent.TimeUnit; @@ -26,12 +25,8 @@ import com.couchbase.client.java.json.JsonObject; import com.couchbase.client.java.manager.bucket.BucketSettings; import com.couchbase.client.java.manager.bucket.BucketType; -import com.couchbase.client.java.manager.bucket.StorageBackend; -import com.couchbase.client.java.manager.view.DesignDocument; -import com.couchbase.client.java.manager.view.View; -import com.couchbase.client.java.view.DesignDocumentNamespace; +import com.couchbase.client.java.manager.query.CreatePrimaryQueryIndexOptions; import org.apache.camel.builder.NotifyBuilder; -import org.apache.camel.component.couchbase.CouchbaseConstants; import org.apache.camel.main.MainConfigurationProperties; import org.apache.camel.test.infra.couchbase.services.CouchbaseService; import org.apache.camel.test.infra.couchbase.services.CouchbaseServiceFactory; @@ -61,16 +56,12 @@ static void init() { CLUSTER = Cluster.connect( SERVICE.getConnectionString(), SERVICE.username(), SERVICE.password() ); - DesignDocument designDoc = new DesignDocument( - CouchbaseConstants.DEFAULT_DESIGN_DOCUMENT_NAME, - Collections.singletonMap( - CouchbaseConstants.DEFAULT_VIEWNAME, - new View("function (doc, meta) { emit(meta.id, doc);}") - ) - ); CLUSTER.buckets().createBucket( - BucketSettings.create(BUCKET).bucketType(BucketType.COUCHBASE).storageBackend(StorageBackend.COUCHSTORE).flushEnabled(true)); - CLUSTER.bucket(BUCKET).viewIndexes().upsertDesignDocument(designDoc, DesignDocumentNamespace.PRODUCTION); + BucketSettings.create(BUCKET).bucketType(BucketType.COUCHBASE).flushEnabled(true)); + CLUSTER.bucket(BUCKET).waitUntilReady(Duration.ofSeconds(30)); + // Create a primary index so the SQL++ consumer can query the bucket + CLUSTER.queryIndexes().createPrimaryIndex(BUCKET, + CreatePrimaryQueryIndexOptions.createPrimaryQueryIndexOptions().ignoreIfExists(true)); } @AfterAll @@ -84,7 +75,6 @@ static void destroy() { @Override public void configureContext(CamelContextConfiguration camelContextConfiguration) { super.configureContext(camelContextConfiguration); - System.out.println("#### " + SERVICE.getConnectionString()); Properties overridenProperties = asProperties( "couchbase.connectionString", SERVICE.getConnectionString(), "couchbase.host", SERVICE.hostname(), From 04136cc0a6a4db361c8ba2ac179bb86e6edd4490 Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 12 Nov 2026 09:03:38 +0000 Subject: [PATCH 2/3] CAMEL-23626: Retry primary index creation until query service is ready The Couchbase query service may not yet be reachable when the KV bucket becomes ready. The SDK topology may also not reflect the external port mappings until after the first bootstrap completes. Use fresh Cluster connections with retry (10 attempts, 3s apart) to handle both cases. --- .../apache/camel/example/CouchbaseTest.java | 41 +++++++++++++++++-- 1 file changed, 38 insertions(+), 3 deletions(-) diff --git a/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java b/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java index e56ef99f0..4de542bfb 100644 --- a/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java +++ b/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java @@ -36,6 +36,8 @@ import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.RegisterExtension; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import static org.apache.camel.util.PropertiesHelper.asProperties; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -45,6 +47,8 @@ */ class CouchbaseTest extends CamelMainTestSupport { + private static final Logger LOG = LoggerFactory.getLogger(CouchbaseTest.class); + private static final String BUCKET = "test-bucket-" + System.currentTimeMillis(); @RegisterExtension @@ -59,9 +63,40 @@ static void init() { CLUSTER.buckets().createBucket( BucketSettings.create(BUCKET).bucketType(BucketType.COUCHBASE).flushEnabled(true)); CLUSTER.bucket(BUCKET).waitUntilReady(Duration.ofSeconds(30)); - // Create a primary index so the SQL++ consumer can query the bucket - CLUSTER.queryIndexes().createPrimaryIndex(BUCKET, - CreatePrimaryQueryIndexOptions.createPrimaryQueryIndexOptions().ignoreIfExists(true)); + // Create a primary index so the SQL++ consumer can query the bucket. + // Retry because the query service may need extra time to become reachable + // after the KV bucket is ready (Couchbase initialises services in parallel, + // and Testcontainers' external-port remapping may not yet be reflected in + // the SDK's cluster topology on the first attempt). + createPrimaryIndexWithRetry(SERVICE.getConnectionString(), SERVICE.username(), SERVICE.password()); + } + + /** + * Creates a primary index on {@link #BUCKET}, retrying with fresh cluster connections until the query service is + * reachable or the maximum number of attempts is exhausted. + */ + private static void createPrimaryIndexWithRetry(String connectionString, String username, String password) { + int maxAttempts = 10; + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + // Use a short-lived cluster per attempt so we always get a fresh topology snapshot. + try (Cluster c = Cluster.connect(connectionString, username, password)) { + c.queryIndexes().createPrimaryIndex(BUCKET, + CreatePrimaryQueryIndexOptions.createPrimaryQueryIndexOptions().ignoreIfExists(true)); + return; + } catch (Exception e) { + if (attempt == maxAttempts) { + throw new RuntimeException("Failed to create primary index after " + maxAttempts + " attempts", e); + } + LOG.info("Query service not yet available (attempt {}/{}), retrying in 3s: {}", + attempt, maxAttempts, e.getMessage()); + try { + Thread.sleep(3000); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted while waiting for query service", ie); + } + } + } } @AfterAll From 5d3d704096756b173577861fbdb088121f8ab178 Mon Sep 17 00:00:00 2001 From: Guillaume Nodet Date: Thu, 12 Nov 2026 09:07:53 +0000 Subject: [PATCH 3/3] CAMEL-23626: Wait for cluster readiness before bucket and index setup Add cluster.waitUntilReady() after Cluster.connect() to ensure the query service is registered in the cluster topology before attempting to create buckets and primary indexes. Without this, the Couchbase SDK may report ServiceNotAvailableException on createPrimaryIndex because the cluster topology snapshot obtained at connect time does not yet include the query service endpoint (Couchbase 8.0 initialises services in parallel). Co-authored-by: Claude Sonnet 4.6 --- .../src/test/java/org/apache/camel/example/CouchbaseTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java b/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java index 4de542bfb..82fa97759 100644 --- a/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java +++ b/couchbase-log/src/test/java/org/apache/camel/example/CouchbaseTest.java @@ -60,6 +60,7 @@ static void init() { CLUSTER = Cluster.connect( SERVICE.getConnectionString(), SERVICE.username(), SERVICE.password() ); + CLUSTER.waitUntilReady(Duration.ofSeconds(30)); CLUSTER.buckets().createBucket( BucketSettings.create(BUCKET).bucketType(BucketType.COUCHBASE).flushEnabled(true)); CLUSTER.bucket(BUCKET).waitUntilReady(Duration.ofSeconds(30));