Tests - improve consistency

Adding consistency utility functions to wait for all nodes to have
a consistent view of a particular resource.

Change-Id: I5ecd82ca0bc88147abf73819b2840fc7934ba1a6
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/183021
Reviewed-by: Graham Pople <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: programmatix

core-io/src/main/java/com/couchbase/client/core/util/ConsistencyUtil.java

@@ -20,6 +20,7 @@
 import com.couchbase.client.core.error.ParsingFailureException;
 import com.couchbase.client.core.error.ScopeNotFoundException;
 import com.couchbase.client.core.service.ServiceType;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.java.analytics.AnalyticsMetaData;
 import com.couchbase.client.java.analytics.AnalyticsResult;
 import com.couchbase.client.java.analytics.AnalyticsStatus;
@@ -101,8 +102,10 @@ static void setup() {
     waitForService(bucket, ServiceType.ANALYTICS);
 
     collectionManager.createScope(scopeName);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), bucket.name(), scopeName);
     CollectionSpec collSpec = CollectionSpec.create(collectionName, scopeName);
     collectionManager.createCollection(collSpec);
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), bucket.name(), collSpec.scopeName(), collSpec.name());
     waitUntilCondition(() -> collectionExists(collectionManager, collSpec));
 
     waitForQueryIndexerToHaveKeyspace(cluster, config().bucketname());

java-client/src/integrationTest/java/com/couchbase/client/java/AnalyticsCollectionIntegrationTest.java

@@ -18,6 +18,7 @@
 import com.couchbase.client.core.error.UnambiguousTimeoutException;
 import com.couchbase.client.core.io.CollectionIdentifier;
 import com.couchbase.client.core.retry.RetryReason;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.java.kv.GetResult;
 import com.couchbase.client.java.kv.MutationResult;
 import com.couchbase.client.java.manager.collection.CollectionSpec;
@@ -61,6 +62,7 @@ void recognizesCollectionAfterCreation() {
     String collId = UUID.randomUUID().toString().substring(0, 10);
     CollectionSpec collectionSpec = CollectionSpec.create(collId, CollectionIdentifier.DEFAULT_SCOPE);
     bucket.collections().createCollection(collectionSpec);
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), bucket.name(), collectionSpec.scopeName(), collectionSpec.name());
 
     Collection collection = bucket.collection(collId);
 

java-client/src/integrationTest/java/com/couchbase/client/java/KeyValueCollectionIntegrationTest.java

@@ -20,6 +20,7 @@
 import com.couchbase.client.core.error.QueryException;
 import com.couchbase.client.core.json.Mapper;
 import com.couchbase.client.core.service.ServiceType;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.java.json.JsonArray;
 import com.couchbase.client.java.json.JsonObject;
 import com.couchbase.client.java.kv.MutationResult;
@@ -98,9 +99,11 @@ static void beforeAll() {
     CollectionSpec collSpec = CollectionSpec.create(COLLECTION_NAME, SCOPE_NAME);
 
     collectionManager.createScope(SCOPE_NAME);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), bucket.name(), SCOPE_NAME);
     waitUntilCondition(() -> scopeExists(collectionManager, SCOPE_NAME));
 
     collectionManager.createCollection(collSpec);
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), bucket.name(), collSpec.scopeName(), collSpec.name());
     waitUntilCondition(() -> collectionExists(collectionManager, collSpec));
 
     waitForQueryIndexerToHaveKeyspace(cluster, COLLECTION_NAME);

java-client/src/integrationTest/java/com/couchbase/client/java/QueryCollectionIntegrationTest.java

@@ -30,6 +30,7 @@
 import com.couchbase.client.core.msg.ResponseStatus;
 import com.couchbase.client.core.service.ServiceType;
 import com.couchbase.client.core.util.CbCollections;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.core.util.UrlQueryStringBuilder;
 import com.couchbase.client.java.json.JsonObject;
 import com.couchbase.client.java.kv.MutationResult;
@@ -249,6 +250,7 @@ void kvQuotaLimitScopesDataSize() throws Exception {
 
     CollectionManager collectionManager = adminCluster.bucket(config().bucketname()).collections();
     collectionManager.createCollection(CollectionSpec.create(scopeName, scopeName));
+    ConsistencyUtil.waitUntilCollectionPresent(adminCluster.core(), config().bucketname(), scopeName, scopeName);
 
     Collection collection = adminCluster.bucket(config().bucketname()).scope(scopeName).collection(scopeName);
 
@@ -422,6 +424,7 @@ void scopeQuotaLimitExceedMaxCollections() throws Exception {
 
     try {
       collectionManager.createCollection(CollectionSpec.create("collection1", scopeName));
+      ConsistencyUtil.waitUntilCollectionPresent(adminCluster.core(), config().bucketname(), scopeName, "collection1");
 
       assertThrows(QuotaLimitedException.class, () ->
         collectionManager.createCollection(CollectionSpec.create("collection2", scopeName)));
@@ -626,6 +629,7 @@ void searchQuotaLimitScopesMaxIndexes() throws Exception {
     try {
       CollectionSpec collectionSpec = CollectionSpec.create(collectionName, scopeName);
       collectionManager.createCollection(collectionSpec);
+      ConsistencyUtil.waitUntilCollectionPresent(adminCluster.core(), config().bucketname(), collectionSpec.scopeName(), collectionSpec.name());
       waitUntilCondition(() -> collectionExists(collectionManager, collectionSpec));
 
       waitForService(adminCluster.bucket(config().bucketname()), ServiceType.SEARCH);

java-client/src/integrationTest/java/com/couchbase/client/java/RateLimitingIntegrationTest.java

@@ -17,6 +17,7 @@
 package com.couchbase.client.java;
 
 import com.couchbase.client.core.service.ServiceType;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.java.kv.MutationResult;
 import com.couchbase.client.java.kv.MutationState;
 import com.couchbase.client.java.manager.search.SearchIndex;
@@ -56,7 +57,7 @@
  */
 @Disabled
 @Flaky
-@IgnoreWhen(missesCapabilities = Capabilities.SEARCH, clusterTypes = ClusterType.CAVES)
+@IgnoreWhen(missesCapabilities = Capabilities.SEARCH, clusterTypes = ClusterType.CAVES, clusterVersionIsBelow = ConsistencyUtil.CLUSTER_VERSION_MB_50101)
 class SearchIntegrationTest extends JavaIntegrationTest {
   private static final Logger LOGGER = LoggerFactory.getLogger(SearchIntegrationTest.class);
 
@@ -73,6 +74,7 @@ static void setup() {
     bucket.waitUntilReady(WAIT_UNTIL_READY_DEFAULT);
     waitForService(bucket, ServiceType.SEARCH);
     cluster.searchIndexes().upsertIndex(new SearchIndex(indexName, config().bucketname()));
+    ConsistencyUtil.waitUntilSearchIndexPresent(cluster.core(), indexName);
   }
 
   @AfterAll

java-client/src/integrationTest/java/com/couchbase/client/java/SearchIntegrationTest.java

@@ -24,6 +24,7 @@
 import com.couchbase.client.core.error.HttpStatusCodeException;
 import com.couchbase.client.core.msg.kv.DurabilityLevel;
 import com.couchbase.client.core.service.ServiceType;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.java.Bucket;
 import com.couchbase.client.java.Cluster;
 import com.couchbase.client.java.Collection;
@@ -36,6 +37,8 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.parallel.Execution;
+import org.junit.jupiter.api.parallel.ExecutionMode;
 
 import java.time.Duration;
 import java.util.HashSet;
@@ -58,6 +61,7 @@
  * Verifies the functionality of the bucket manager.
  */
 @IgnoreWhen(clusterTypes = { ClusterType.MOCKED, ClusterType.CAVES, ClusterType.CAPELLA })
+@Execution(ExecutionMode.CONCURRENT)
 class BucketManagerIntegrationTest extends JavaIntegrationTest {
 
   private static Cluster cluster;
@@ -93,6 +97,7 @@ static void tearDown() {
   }
 
   private void waitUntilHealthy(String bucket) {
+    ConsistencyUtil.waitUntilBucketPresent(cluster.core(), bucket);
     Util.waitUntilCondition(() -> {
       try {
         BucketSettings bkt = buckets.getBucket(bucket);
@@ -104,6 +109,7 @@ private void waitUntilHealthy(String bucket) {
   }
 
   private void waitUntilDropped(String bucket) {
+    ConsistencyUtil.waitUntilBucketDropped(cluster.core(), bucket);
     Util.waitUntilCondition(() -> {
       try {
         buckets.getBucket(bucket);
@@ -223,7 +229,7 @@ void createCouchbaseBucketWithStorageBackendMagma() {
     assertEquals(StorageBackend.MAGMA, settings.storageBackend());
   }
 
-    @Test
+  @Test
   void shouldPickNoDurabilityLevelIfNotSpecified() {
     String name = UUID.randomUUID().toString();
     createBucket(BucketSettings.create(name)
@@ -245,7 +251,6 @@ void createAndDropBucket() {
     String name = UUID.randomUUID().toString();
 
     createBucket(BucketSettings.create(name));
-    waitUntilHealthy(name);
     assertTrue(buckets.getAllBuckets().containsKey(name));
 
     buckets.dropBucket(name);
@@ -315,7 +320,6 @@ void createWithMoreThanOneReplica() {
     String name = UUID.randomUUID().toString();
 
     createBucket(BucketSettings.create(name).numReplicas(3));
-    waitUntilHealthy(name);
 
     BucketSettings bucket = buckets.getBucket(name);
     assertEquals(3, bucket.numReplicas());
@@ -327,7 +331,6 @@ void customConflictResolution() {
     String bucketName = UUID.randomUUID().toString();
     try {
       createBucket(BucketSettings.create(bucketName).ramQuotaMB(100).conflictResolutionType(ConflictResolutionType.CUSTOM));
-      waitUntilHealthy(bucketName);
 
       BucketSettings bucket = buckets.getBucket(bucketName);
       assertEquals(ConflictResolutionType.CUSTOM, bucket.conflictResolutionType());
@@ -382,6 +385,7 @@ private void assertCreatedBucket(final BucketSettings settings) {
 
   private void createBucket(BucketSettings settings) {
     buckets.createBucket(settings);
+    ConsistencyUtil.waitUntilBucketPresent(cluster.core(), settings.name());
     bucketsToDrop.add(settings.name());
     waitUntilHealthy(settings.name());
   }

java-client/src/integrationTest/java/com/couchbase/client/java/manager/bucket/BucketManagerIntegrationTest.java

@@ -22,6 +22,7 @@
 import com.couchbase.client.core.error.ScopeExistsException;
 import com.couchbase.client.core.error.ScopeNotFoundException;
 import com.couchbase.client.core.service.ServiceType;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.java.Bucket;
 import com.couchbase.client.java.Cluster;
 import com.couchbase.client.java.util.JavaIntegrationTest;
@@ -73,13 +74,15 @@ void shouldCreateScopeAndCollection() {
     assertThrows(ScopeNotFoundException.class, () -> collections.createCollection(collSpec));
 
     collections.createScope(scopeName);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), config().bucketname(), scopeName);
 
     waitUntilCondition(() -> {
       Optional<ScopeSpec> scope = collections.getAllScopes().stream().filter(ss -> ss.name().equals(scopeName)).findFirst();
       return scope.isPresent() && scope.get().equals(scopeSpec);
     });
 
     collections.createCollection(collSpec);
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), collSpec.scopeName(), collSpec.name());
     waitUntilCondition(() -> {
       boolean collExists = collectionExists(collections, collSpec);
       if (collExists) {
@@ -95,6 +98,7 @@ void shouldThrowWhenScopeAlreadyExists() {
     String scope = randomString();
 
     collections.createScope(scope);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), config().bucketname(), scope);
     waitUntilCondition(() -> scopeExists(collections, scope));
     assertThrows(ScopeExistsException.class, () -> collections.createScope(scope));
   }
@@ -103,10 +107,12 @@ void shouldThrowWhenScopeAlreadyExists() {
   void shouldThrowWhenCollectionAlreadyExists() {
     String scope = randomString();
     collections.createScope(scope);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), config().bucketname(), scope);
     waitUntilCondition(() -> scopeExists(collections, scope));
 
     CollectionSpec collectionSpec = CollectionSpec.create(randomString(), scope);
     collections.createCollection(collectionSpec);
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), collectionSpec.scopeName(), collectionSpec.name());
 
     assertThrows(CollectionExistsException.class, () -> collections.createCollection(collectionSpec));
   }
@@ -124,16 +130,21 @@ void shouldDropScopeAndCollections() {
     assertThrows(ScopeNotFoundException.class, () -> collections.dropCollection(collectionSpec1));
 
     collections.createScope(scope);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), config().bucketname(), scope);
     waitUntilCondition(() -> scopeExists(collections, scope));
 
     collections.createCollection(collectionSpec1);
     collections.createCollection(collectionSpec2);
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), collectionSpec1.scopeName(), collectionSpec1.name());
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), collectionSpec2.scopeName(), collectionSpec2.name());
 
     collections.dropCollection(collectionSpec1);
+    ConsistencyUtil.waitUntilCollectionDropped(cluster.core(), config().bucketname(), collectionSpec1.scopeName(), collectionSpec1.name());
     waitUntilCondition(() -> !collectionExists(collections, collectionSpec1));
     assertThrows(CollectionNotFoundException.class, () -> collections.dropCollection(collectionSpec1));
 
     collections.dropScope(scope);
+    ConsistencyUtil.waitUntilScopeDropped(cluster.core(), config().bucketname(), scope);
     waitUntilCondition(() -> !scopeExists(collections, scope));
 
     assertThrows(ScopeNotFoundException.class, () -> collections.dropCollection(collectionSpec2));
@@ -149,10 +160,13 @@ void shouldCreateCollectionWithMaxExpiry() {
     CollectionSpec collectionSpec2 = CollectionSpec.create(collection2, scope);
 
     collections.createScope(scope);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), config().bucketname(), scope);
     waitUntilCondition(() -> scopeExists(collections, scope));
 
     collections.createCollection(collectionSpec1);
     collections.createCollection(collectionSpec2);
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), collectionSpec1.scopeName(), collectionSpec1.name());
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), collectionSpec2.scopeName(), collectionSpec2.name());
 
     waitUntilCondition(() -> collectionExists(collections, collectionSpec1));
     waitUntilCondition(() -> collectionExists(collections, collectionSpec2));
@@ -178,6 +192,7 @@ void createMaxNumOfCollections() {
     int collectionsPerScope = 10;
     ScopeSpec scopeSpec = ScopeSpec.create(scopeName);
     collections.createScope(scopeName);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), config().bucketname(), scopeName);
     waitUntilCondition(() -> scopeExists(collections, scopeName));
 
     List<ScopeSpec> scopeList = collections.getAllScopes();
@@ -193,6 +208,7 @@ void createMaxNumOfCollections() {
     for (int i = 0; i < collectionsPerScope; i++) {
       CollectionSpec collectionSpec = CollectionSpec.create(String.valueOf(collectionsPerScope + i), scopeName);
       collections.createCollection(collectionSpec);
+      ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), collectionSpec.scopeName(), collectionSpec.name());
       waitUntilCondition(() -> collectionExists(collections, collectionSpec));
       waitUntilCondition(() -> collections.getAllScopes().stream().anyMatch(ss -> ss.collections().contains(collectionSpec)));
     }

java-client/src/integrationTest/java/com/couchbase/client/java/manager/collection/CollectionManagerIntegrationTest.java

@@ -17,6 +17,7 @@
 package com.couchbase.client.java.manager.eventing;
 
 import com.couchbase.client.core.error.*;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.java.Bucket;
 import com.couchbase.client.java.Cluster;
 import com.couchbase.client.java.Collection;
@@ -61,8 +62,11 @@ static void setup() {
     Bucket bucket = cluster.bucket(config().bucketname());
 
     bucket.collections().createScope("eventing");
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), config().bucketname(), "eventing");
     bucket.collections().createCollection(CollectionSpec.create("source", "eventing"));
     bucket.collections().createCollection(CollectionSpec.create("meta", "eventing"));
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), "source", "eventing");
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), "meta", "eventing");
 
     sourceCollection = bucket.scope("eventing").collection("source");
     metaCollection = bucket.scope("eventing").collection("meta");

java-client/src/integrationTest/java/com/couchbase/client/java/manager/eventing/EventingFunctionManagerIntegrationTest.java

@@ -19,6 +19,7 @@
 import com.couchbase.client.core.error.IndexExistsException;
 import com.couchbase.client.core.error.IndexNotFoundException;
 import com.couchbase.client.core.service.ServiceType;
+import com.couchbase.client.core.util.ConsistencyUtil;
 import com.couchbase.client.java.Bucket;
 import com.couchbase.client.java.Cluster;
 import com.couchbase.client.java.manager.collection.CollectionSpec;
@@ -51,6 +52,7 @@
 import static com.couchbase.client.java.manager.query.WatchQueryIndexesOptions.watchQueryIndexesOptions;
 import static com.couchbase.client.test.Capabilities.COLLECTIONS;
 import static com.couchbase.client.test.Capabilities.QUERY;
+import static com.couchbase.client.test.Capabilities.SUBDOC_REVIVE_DOCUMENT;
 import static com.couchbase.client.test.ClusterType.CAVES;
 import static com.couchbase.client.test.ClusterType.MOCKED;
 import static java.util.Collections.emptyList;
@@ -91,7 +93,9 @@ static void setup() throws Exception {
     waitForQueryIndexerToHaveKeyspace(cluster, bucketName);
 
     bucket.collections().createScope(scopeName);
+    ConsistencyUtil.waitUntilScopePresent(cluster.core(), config().bucketname(), scopeName);
     bucket.collections().createCollection(CollectionSpec.create(collectionName, scopeName));
+    ConsistencyUtil.waitUntilCollectionPresent(cluster.core(), config().bucketname(), scopeName, collectionName);
     waitForQueryIndexerToHaveKeyspace(cluster, collectionName);
   }