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..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 @@ -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; @@ -41,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; @@ -50,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 @@ -61,16 +60,44 @@ 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.waitUntilReady(Duration.ofSeconds(30)); 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. + // 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 @@ -84,7 +111,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(),