From 6ec58afa17ca4528b92d7d5ded6d49912cf6333e Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Wed, 1 Jul 2020 19:43:32 -0700 Subject: [PATCH 01/15] fix for same collection --- .../implementation/BackoffRetryUtility.java | 12 +++++ .../InvalidPartitionExceptionRetryPolicy.java | 9 +++- .../implementation/ObservableHelper.java | 17 +++++++ .../implementation/RxDocumentClientImpl.java | 35 +++++++++++--- .../implementation/TestConfigurations.java | 6 +-- .../azure/cosmos/AzureKeyCredentialTest.java | 11 ----- ...osContainerContentResponseOnWriteTest.java | 11 ----- .../com/azure/cosmos/CosmosContainerTest.java | 13 ----- .../CosmosItemContentResponseOnWriteTest.java | 11 ----- .../azure/cosmos/rx/CollectionCrudTest.java | 13 ----- .../azure/cosmos/rx/CollectionQueryTest.java | 47 ++++++++++++++++++- .../com/azure/cosmos/rx/TestSuiteBase.java | 6 ++- 12 files changed, 118 insertions(+), 73 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java index f57eeec24e2f..fa0971753f62 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java @@ -2,6 +2,7 @@ // Licensed under the MIT License. package com.azure.cosmos.implementation; +import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.time.Duration; @@ -38,6 +39,17 @@ static public Mono executeRetry(Callable> callbackMethod, }).retryWhen(RetryUtils.toRetryWhenFunc(retryPolicy)); } + static public Flux fluxExecuteRetry(Callable> callbackMethod, IRetryPolicy retryPolicy) { + + return Flux.defer(() -> { + try { + return callbackMethod.call(); + } catch (Exception e) { + return Mono.error(e); + } + }).retryWhen(RetryUtils.toRetryWhenFunc(retryPolicy)); + } + static public Mono executeAsync( Function, Mono> callbackMethod, IRetryPolicy retryPolicy, Function, Mono> inBackoffAlternateCallbackMethod, diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/InvalidPartitionExceptionRetryPolicy.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/InvalidPartitionExceptionRetryPolicy.java index 4c05f78ed843..76e37f801211 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/InvalidPartitionExceptionRetryPolicy.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/InvalidPartitionExceptionRetryPolicy.java @@ -41,7 +41,9 @@ public InvalidPartitionExceptionRetryPolicy(RxCollectionCache collectionCache, @Override public void onBeforeSendRequest(RxDocumentServiceRequest request) { this.request = request; - this.nextPolicy.onBeforeSendRequest(request); + if (this.nextPolicy != null) { + this.nextPolicy.onBeforeSendRequest(request); + } } @Override @@ -73,6 +75,9 @@ public Mono shouldRetry(Exception e) { } } - return this.nextPolicy.shouldRetry(e); + if (this.nextPolicy != null) { + return this.nextPolicy.shouldRetry(e); + } + return Mono.just(ShouldRetryResult.error(e)); } } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ObservableHelper.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ObservableHelper.java index 91309bb3e0b2..5be35ba27c29 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ObservableHelper.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ObservableHelper.java @@ -44,4 +44,21 @@ static public Mono inlineIfPossibleAsObs(Callable> function, IRet return BackoffRetryUtility.executeRetry(() -> function.call(), retryPolicy); } } + + static public Flux fluxInlineIfPossibleAsObs(Callable> function, IRetryPolicy retryPolicy) { + + if (retryPolicy == null) { + // shortcut + return Flux.defer(() -> { + try { + return function.call(); + } catch (Exception e) { + return Mono.error(e); + } + }); + + } else { + return BackoffRetryUtility.fluxExecuteRetry(() -> function.call(), retryPolicy); + } + } } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java index 152da0570689..54f07f99a83b 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java @@ -560,18 +560,39 @@ private String parentResourceLinkToQueryLink(String parentResouceLink, ResourceT } private Flux> createQuery( - String parentResourceLink, + String parentResourceLink, + SqlQuerySpec sqlQuery, + CosmosQueryRequestOptions options, + Class klass, + ResourceType resourceTypeEnum) { + + String resourceLink = parentResourceLinkToQueryLink(parentResourceLink, resourceTypeEnum); + UUID activityId = Utils.randomUUID(); + IDocumentQueryClient queryClient = documentQueryClientImpl(RxDocumentClientImpl.this); + + InvalidPartitionExceptionRetryPolicy invalidPartitionExceptionRetryPolicy = new InvalidPartitionExceptionRetryPolicy( + this.collectionCache, + null, + resourceLink, + options); + + return ObservableHelper.fluxInlineIfPossibleAsObs( + () -> createQueryInternal(resourceLink, sqlQuery, options, klass, resourceTypeEnum, queryClient, activityId), + invalidPartitionExceptionRetryPolicy); + } + + private Flux> createQueryInternal( + String resourceLink, SqlQuerySpec sqlQuery, CosmosQueryRequestOptions options, Class klass, - ResourceType resourceTypeEnum) { + ResourceType resourceTypeEnum, + IDocumentQueryClient queryClient, + UUID activityId) { - String queryResourceLink = parentResourceLinkToQueryLink(parentResourceLink, resourceTypeEnum); - - UUID activityId = Utils.randomUUID(); - IDocumentQueryClient queryClient = documentQueryClientImpl(RxDocumentClientImpl.this); Flux> executionContext = - DocumentQueryExecutionContextFactory.createDocumentQueryExecutionContextAsync(queryClient, resourceTypeEnum, klass, sqlQuery , options, queryResourceLink, false, activityId); + DocumentQueryExecutionContextFactory.createDocumentQueryExecutionContextAsync(queryClient, resourceTypeEnum, klass, sqlQuery , options, resourceLink, false, activityId); + return executionContext.flatMap(iDocumentQueryExecutionContext -> { QueryInfo queryInfo = null; if (iDocumentQueryExecutionContext instanceof PipelinedDocumentQueryExecutionContext) { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TestConfigurations.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TestConfigurations.java index bb852c114816..83f07920a04f 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TestConfigurations.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TestConfigurations.java @@ -39,19 +39,19 @@ public final class TestConfigurations { properties.getProperty("ACCOUNT_KEY", StringUtils.defaultString(Strings.emptyToNull( System.getenv().get("ACCOUNT_KEY")), - "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==")); + "hoTaUW11OCQ16P4Qp7tUM3C4xk20vTQZbh0eSTZHwTPyAQNMwlYqim63ltJJLuhUeDLC081r34vHOn8PHlGmug==")); public final static String SECONDARY_MASTER_KEY = properties.getProperty("SECONDARY_ACCOUNT_KEY", StringUtils.defaultString(Strings.emptyToNull( System.getenv().get("SECONDARY_ACCOUNT_KEY")), - "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==")); + "hoTaUW11OCQ16P4Qp7tUM3C4xk20vTQZbh0eSTZHwTPyAQNMwlYqim63ltJJLuhUeDLC081r34vHOn8PHlGmug==")); public final static String HOST = properties.getProperty("ACCOUNT_HOST", StringUtils.defaultString(Strings.emptyToNull( System.getenv().get("ACCOUNT_HOST")), - "https://localhost:443/")); + "https://cosmos-sdk-tests.documents.azure.com:443/")); public final static String CONSISTENCY = properties.getProperty("ACCOUNT_CONSISTENCY", diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java index 9e161fe0662e..3ef96e0ebb4c 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java @@ -65,17 +65,6 @@ public Object[][] crudArgProvider() { }; } - private CosmosContainerProperties getCollectionDefinition(String collectionName) { - PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition(); - ArrayList paths = new ArrayList<>(); - paths.add("/mypk"); - partitionKeyDef.setPaths(paths); - - return new CosmosContainerProperties( - collectionName, - partitionKeyDef); - } - private InternalObjectNode getDocumentDefinition(String documentId) { final String uuid = UUID.randomUUID().toString(); return new InternalObjectNode(String.format("{ " diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerContentResponseOnWriteTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerContentResponseOnWriteTest.java index efaab4cdf95f..0e529eff2e12 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerContentResponseOnWriteTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerContentResponseOnWriteTest.java @@ -102,17 +102,6 @@ public void replaceContainer_withContentResponseOnWriteDisabled() throws Excepti } - private CosmosContainerProperties getCollectionDefinition(String collectionName) { - PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition(); - ArrayList paths = new ArrayList(); - paths.add("/mypk"); - partitionKeyDef.setPaths(paths); - - return new CosmosContainerProperties( - collectionName, - partitionKeyDef); - } - private void validateContainerResponse(CosmosContainerProperties containerProperties, CosmosContainerResponse createResponse) { // Basic validation diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerTest.java index 65970732ce53..b5caf3ef2868 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerTest.java @@ -53,19 +53,6 @@ public void afterClass() { safeCloseSyncClient(client); } - private CosmosContainerProperties getCollectionDefinition(String collectionName) { - PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition(); - ArrayList paths = new ArrayList(); - paths.add("/mypk"); - partitionKeyDef.setPaths(paths); - - CosmosContainerProperties collectionDefinition = new CosmosContainerProperties( - collectionName, - partitionKeyDef); - - return collectionDefinition; - } - @Test(groups = { "emulator" }, timeOut = TIMEOUT) public void createContainer_withProperties() throws Exception { String collectionName = UUID.randomUUID().toString(); diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosItemContentResponseOnWriteTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosItemContentResponseOnWriteTest.java index 767d96aafc4c..4dcb5b305a9e 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosItemContentResponseOnWriteTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosItemContentResponseOnWriteTest.java @@ -106,17 +106,6 @@ public void deleteItem_withContentResponseOnWriteDisabled() throws Exception { validateMinimalItemResponse(properties, deleteResponse, false); } - private CosmosContainerProperties getCollectionDefinition(String collectionName) { - PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition(); - ArrayList paths = new ArrayList(); - paths.add("/mypk"); - partitionKeyDef.setPaths(paths); - - return new CosmosContainerProperties( - collectionName, - partitionKeyDef); - } - private InternalObjectNode getDocumentDefinition(String documentId) { final String uuid = UUID.randomUUID().toString(); final InternalObjectNode properties = diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionCrudTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionCrudTest.java index f2d660ce53ca..1b05b05355b7 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionCrudTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionCrudTest.java @@ -70,19 +70,6 @@ public Object[][] collectionCrudArgProvider() { }; } - private CosmosContainerProperties getCollectionDefinition(String collectionName) { - PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition(); - ArrayList paths = new ArrayList(); - paths.add("/mypk"); - partitionKeyDef.setPaths(paths); - - CosmosContainerProperties collectionDefinition = new CosmosContainerProperties( - collectionName, - partitionKeyDef); - - return collectionDefinition; - } - @Test(groups = { "emulator" }, timeOut = TIMEOUT, dataProvider = "collectionCrudArgProvider") public void createCollection(String collectionName) throws InterruptedException { CosmosContainerProperties collectionDefinition = getCollectionDefinition(collectionName); diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionQueryTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionQueryTest.java index 1349df5d905b..29bb4aecd19b 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionQueryTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionQueryTest.java @@ -6,6 +6,8 @@ import com.azure.cosmos.CosmosAsyncContainer; import com.azure.cosmos.CosmosAsyncDatabase; import com.azure.cosmos.CosmosClientBuilder; +import com.azure.cosmos.implementation.InternalObjectNode; +import com.azure.cosmos.implementation.RxDocumentClientImpl; import com.azure.cosmos.models.CosmosContainerProperties; import com.azure.cosmos.util.CosmosPagedFlux; import com.azure.cosmos.CosmosDatabaseForTest; @@ -13,6 +15,7 @@ import com.azure.cosmos.models.PartitionKeyDefinition; import com.azure.cosmos.implementation.FeedResponseListValidator; import com.azure.cosmos.implementation.FeedResponseValidator; +import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.commons.lang3.StringUtils; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -27,7 +30,7 @@ import static org.assertj.core.api.Assertions.assertThat; public class CollectionQueryTest extends TestSuiteBase { - private final static int TIMEOUT = 30000; + private final static int TIMEOUT = 300000; private final String databaseId = CosmosDatabaseForTest.generateId(); private List createdCollections = new ArrayList<>(); private CosmosAsyncClient client; @@ -109,6 +112,38 @@ public void queryCollections_NoResults() throws Exception { validateQuerySuccess(queryObservable.byPage(), validator); } + @Test(groups = { "simple" }, timeOut = TIMEOUT) + public void queryByRecreatCollectionWithSameName() throws Exception { + String testCollectionId = UUID.randomUUID().toString(); + CosmosContainerProperties collectionDefinition = getCollectionDefinition(testCollectionId); + CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); + String query = "SELECT * FROM r"; + + // step1: create container and do query + this.createdDatabase.createContainer(collectionDefinition).block(); + CosmosAsyncContainer container = this.createdDatabase.getContainer(collectionDefinition.getId()); + InternalObjectNode document1 = getDocumentDefinition(); + container.createItem(document1).block(); + container.queryItems(query, requestOptions, InternalObjectNode.class); + + // step2: delete the container created on step 1 + safeDeleteCollection(container); + + // step3: create a new collection with the same id as step 1 and do query + this.createdDatabase.createContainer(collectionDefinition).block(); + container = this.createdDatabase.getContainer(collectionDefinition.getId()); + container.createItem(document1).block(); + CosmosPagedFlux queryFlux = container.queryItems(query, requestOptions, InternalObjectNode.class); + FeedResponseListValidator queryValidator = new FeedResponseListValidator.Builder() + .totalSize(1) + .numberOfPages(1) + .build(); + validateQuerySuccess(queryFlux.byPage(10), queryValidator); + + safeDeleteCollection(container); + } + + @BeforeClass(groups = { "simple" }, timeOut = SETUP_TIMEOUT) public void before_CollectionQueryTest() throws Exception { client = getClientBuilder().buildAsyncClient(); @@ -128,4 +163,14 @@ public void afterClass() { safeDeleteDatabase(createdDatabase); safeClose(client); } + + private static InternalObjectNode getDocumentDefinition() { + String uuid = UUID.randomUUID().toString(); + InternalObjectNode doc = new InternalObjectNode(String.format("{ " + + "\"id\": \"%s\", " + + "\"mypk\": \"%s\", " + + "}" + , uuid, uuid)); + return doc; + } } diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java index dead021f8e7e..30805281f90c 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java @@ -516,12 +516,16 @@ private static CosmosAsyncContainer safeCreateCollection(CosmosAsyncClient clien } static protected CosmosContainerProperties getCollectionDefinition() { + return getCollectionDefinition(UUID.randomUUID().toString()); + } + + static protected CosmosContainerProperties getCollectionDefinition(String collectionId) { PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition(); ArrayList paths = new ArrayList(); paths.add("/mypk"); partitionKeyDef.setPaths(paths); - CosmosContainerProperties collectionDefinition = new CosmosContainerProperties(UUID.randomUUID().toString(), partitionKeyDef); + CosmosContainerProperties collectionDefinition = new CosmosContainerProperties(collectionId, partitionKeyDef); return collectionDefinition; } From ce4b6f18a3b9490f7b233f0fd419e29b12a0dbc2 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 2 Jul 2020 00:50:07 -0700 Subject: [PATCH 02/15] change --- .../InvalidPartitionExceptionRetryPolicy.java | 2 +- .../implementation/TestConfigurations.java | 6 +- .../azure/cosmos/AzureKeyCredentialTest.java | 11 ++- ...osContainerContentResponseOnWriteTest.java | 2 - .../com/azure/cosmos/CosmosContainerTest.java | 4 +- .../CosmosItemContentResponseOnWriteTest.java | 3 - ...QueryTest.java => ContainerQueryTest.java} | 69 +++++++++---------- .../com/azure/cosmos/rx/TestSuiteBase.java | 2 +- 8 files changed, 45 insertions(+), 54 deletions(-) rename sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/{CollectionQueryTest.java => ContainerQueryTest.java} (69%) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/InvalidPartitionExceptionRetryPolicy.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/InvalidPartitionExceptionRetryPolicy.java index 76e37f801211..5083b6210aac 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/InvalidPartitionExceptionRetryPolicy.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/InvalidPartitionExceptionRetryPolicy.java @@ -58,7 +58,7 @@ public Mono shouldRetry(Exception e) { // TODO: this is blocking. is that fine? if(this.cosmosQueryRequestOptions != null) { this.clientCollectionCache.refresh( - BridgeInternal.getMetaDataDiagnosticContext(this.request.requestContext.cosmosDiagnostics), + this.request != null ? BridgeInternal.getMetaDataDiagnosticContext(this.request.requestContext.cosmosDiagnostics) : null, collectionLink, ModelBridgeInternal.getPropertiesFromQueryRequestOptions(this.cosmosQueryRequestOptions)); } else { diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TestConfigurations.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TestConfigurations.java index 83f07920a04f..bb852c114816 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TestConfigurations.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/TestConfigurations.java @@ -39,19 +39,19 @@ public final class TestConfigurations { properties.getProperty("ACCOUNT_KEY", StringUtils.defaultString(Strings.emptyToNull( System.getenv().get("ACCOUNT_KEY")), - "hoTaUW11OCQ16P4Qp7tUM3C4xk20vTQZbh0eSTZHwTPyAQNMwlYqim63ltJJLuhUeDLC081r34vHOn8PHlGmug==")); + "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==")); public final static String SECONDARY_MASTER_KEY = properties.getProperty("SECONDARY_ACCOUNT_KEY", StringUtils.defaultString(Strings.emptyToNull( System.getenv().get("SECONDARY_ACCOUNT_KEY")), - "hoTaUW11OCQ16P4Qp7tUM3C4xk20vTQZbh0eSTZHwTPyAQNMwlYqim63ltJJLuhUeDLC081r34vHOn8PHlGmug==")); + "C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw==")); public final static String HOST = properties.getProperty("ACCOUNT_HOST", StringUtils.defaultString(Strings.emptyToNull( System.getenv().get("ACCOUNT_HOST")), - "https://cosmos-sdk-tests.documents.azure.com:443/")); + "https://localhost:443/")); public final static String CONSISTENCY = properties.getProperty("ACCOUNT_CONSISTENCY", diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java index 3ef96e0ebb4c..e8c80957fa06 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java @@ -1,25 +1,24 @@ package com.azure.cosmos; -import com.azure.cosmos.implementation.InternalObjectNode; import com.azure.cosmos.implementation.FailureValidator; +import com.azure.cosmos.implementation.InternalObjectNode; import com.azure.cosmos.implementation.RetryAnalyzer; -import com.azure.cosmos.models.CosmosContainerResponse; -import com.azure.cosmos.models.CosmosDatabaseResponse; -import com.azure.cosmos.models.CosmosItemResponse; +import com.azure.cosmos.implementation.TestConfigurations; import com.azure.cosmos.models.CosmosContainerProperties; import com.azure.cosmos.models.CosmosContainerRequestOptions; +import com.azure.cosmos.models.CosmosContainerResponse; import com.azure.cosmos.models.CosmosDatabaseProperties; import com.azure.cosmos.models.CosmosDatabaseRequestOptions; +import com.azure.cosmos.models.CosmosDatabaseResponse; import com.azure.cosmos.models.CosmosItemRequestOptions; +import com.azure.cosmos.models.CosmosItemResponse; import com.azure.cosmos.models.CosmosResponse; import com.azure.cosmos.models.IndexingMode; import com.azure.cosmos.models.IndexingPolicy; import com.azure.cosmos.models.ModelBridgeInternal; import com.azure.cosmos.models.PartitionKey; -import com.azure.cosmos.models.PartitionKeyDefinition; import com.azure.cosmos.rx.CosmosItemResponseValidator; import com.azure.cosmos.rx.TestSuiteBase; -import com.azure.cosmos.implementation.TestConfigurations; import org.testng.annotations.AfterClass; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeClass; diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerContentResponseOnWriteTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerContentResponseOnWriteTest.java index 0e529eff2e12..c9c473d7ae75 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerContentResponseOnWriteTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerContentResponseOnWriteTest.java @@ -11,14 +11,12 @@ import com.azure.cosmos.models.CosmosContainerResponse; import com.azure.cosmos.models.IndexingMode; import com.azure.cosmos.models.IndexingPolicy; -import com.azure.cosmos.models.PartitionKeyDefinition; import com.azure.cosmos.rx.TestSuiteBase; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; import org.testng.annotations.Factory; import org.testng.annotations.Test; -import java.util.ArrayList; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerTest.java index b5caf3ef2868..39c7f7f9d129 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosContainerTest.java @@ -6,17 +6,16 @@ package com.azure.cosmos; +import com.azure.cosmos.implementation.HttpConstants; import com.azure.cosmos.models.CosmosContainerProperties; import com.azure.cosmos.models.CosmosContainerRequestOptions; import com.azure.cosmos.models.CosmosContainerResponse; import com.azure.cosmos.models.CosmosQueryRequestOptions; import com.azure.cosmos.models.IndexingMode; import com.azure.cosmos.models.IndexingPolicy; -import com.azure.cosmos.models.PartitionKeyDefinition; import com.azure.cosmos.models.SqlQuerySpec; import com.azure.cosmos.models.ThroughputProperties; import com.azure.cosmos.rx.TestSuiteBase; -import com.azure.cosmos.implementation.HttpConstants; import com.azure.cosmos.util.CosmosPagedIterable; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -24,7 +23,6 @@ import org.testng.annotations.Factory; import org.testng.annotations.Test; -import java.util.ArrayList; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosItemContentResponseOnWriteTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosItemContentResponseOnWriteTest.java index 4dcb5b305a9e..e4d0b96e2174 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosItemContentResponseOnWriteTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/CosmosItemContentResponseOnWriteTest.java @@ -7,19 +7,16 @@ package com.azure.cosmos; import com.azure.cosmos.implementation.InternalObjectNode; -import com.azure.cosmos.models.CosmosContainerProperties; import com.azure.cosmos.models.CosmosItemRequestOptions; import com.azure.cosmos.models.CosmosItemResponse; import com.azure.cosmos.models.ModelBridgeInternal; import com.azure.cosmos.models.PartitionKey; -import com.azure.cosmos.models.PartitionKeyDefinition; import com.azure.cosmos.rx.TestSuiteBase; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; import org.testng.annotations.Factory; import org.testng.annotations.Test; -import java.util.ArrayList; import java.util.UUID; import static org.assertj.core.api.Assertions.assertThat; diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionQueryTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java similarity index 69% rename from sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionQueryTest.java rename to sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java index 29bb4aecd19b..110a72181987 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/CollectionQueryTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java @@ -7,15 +7,13 @@ import com.azure.cosmos.CosmosAsyncDatabase; import com.azure.cosmos.CosmosClientBuilder; import com.azure.cosmos.implementation.InternalObjectNode; -import com.azure.cosmos.implementation.RxDocumentClientImpl; import com.azure.cosmos.models.CosmosContainerProperties; +import com.azure.cosmos.models.PartitionKeyDefinition; import com.azure.cosmos.util.CosmosPagedFlux; import com.azure.cosmos.CosmosDatabaseForTest; import com.azure.cosmos.models.CosmosQueryRequestOptions; -import com.azure.cosmos.models.PartitionKeyDefinition; import com.azure.cosmos.implementation.FeedResponseListValidator; import com.azure.cosmos.implementation.FeedResponseValidator; -import com.fasterxml.jackson.databind.node.ObjectNode; import org.apache.commons.lang3.StringUtils; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -29,39 +27,39 @@ import static org.assertj.core.api.Assertions.assertThat; -public class CollectionQueryTest extends TestSuiteBase { - private final static int TIMEOUT = 300000; +public class ContainerQueryTest extends TestSuiteBase { + private final static int TIMEOUT = 30000; private final String databaseId = CosmosDatabaseForTest.generateId(); - private List createdCollections = new ArrayList<>(); + private List createdContainers = new ArrayList<>(); private CosmosAsyncClient client; private CosmosAsyncDatabase createdDatabase; @Factory(dataProvider = "clientBuilders") - public CollectionQueryTest(CosmosClientBuilder clientBuilder) { + public ContainerQueryTest(CosmosClientBuilder clientBuilder) { super(clientBuilder); this.subscriberValidationTimeout = TIMEOUT; } @Test(groups = { "simple" }, timeOut = TIMEOUT) - public void queryCollectionsWithFilter() throws Exception { + public void queryContainersWithFilter() throws Exception { - String filterCollectionId = createdCollections.get(0).getId(); - String query = String.format("SELECT * from c where c.id = '%s'", filterCollectionId); + String filterContainerId = createdContainers.get(0).getId(); + String query = String.format("SELECT * from c where c.id = '%s'", filterContainerId); CosmosQueryRequestOptions options = new CosmosQueryRequestOptions(); int maxItemCount = 2; CosmosPagedFlux queryObservable = createdDatabase.queryContainers(query, options); - List expectedCollections = createdCollections.stream() - .filter(c -> StringUtils.equals(filterCollectionId, c.getId()) ).collect(Collectors.toList()); + List expectedContainers = createdContainers.stream() + .filter(c -> StringUtils.equals(filterContainerId, c.getId()) ).collect(Collectors.toList()); - assertThat(expectedCollections).isNotEmpty(); + assertThat(expectedContainers).isNotEmpty(); - int expectedPageSize = (expectedCollections.size() + maxItemCount - 1) / maxItemCount; + int expectedPageSize = (expectedContainers.size() + maxItemCount - 1) / maxItemCount; FeedResponseListValidator validator = new FeedResponseListValidator.Builder() - .totalSize(expectedCollections.size()) - .exactlyContainsInAnyOrder(expectedCollections.stream().map(d -> d.read().block().getProperties().getResourceId()).collect(Collectors.toList())) + .totalSize(expectedContainers.size()) + .exactlyContainsInAnyOrder(expectedContainers.stream().map(d -> d.read().block().getProperties().getResourceId()).collect(Collectors.toList())) .numberOfPages(expectedPageSize) .pageSatisfy(0, new FeedResponseValidator.Builder() .requestChargeGreaterThanOrEqualTo(1.0).build()) @@ -71,7 +69,7 @@ public void queryCollectionsWithFilter() throws Exception { } @Test(groups = { "simple" }, timeOut = TIMEOUT) - public void queryAllCollections() throws Exception { + public void queryAllContainers() throws Exception { String query = "SELECT * from c"; @@ -79,15 +77,15 @@ public void queryAllCollections() throws Exception { int maxItemCount = 2; CosmosPagedFlux queryObservable = createdDatabase.queryContainers(query, options); - List expectedCollections = createdCollections; + List expectedContainers = createdContainers; - assertThat(expectedCollections).isNotEmpty(); + assertThat(expectedContainers).isNotEmpty(); - int expectedPageSize = (expectedCollections.size() + maxItemCount - 1) / maxItemCount; + int expectedPageSize = (expectedContainers.size() + maxItemCount - 1) / maxItemCount; FeedResponseListValidator validator = new FeedResponseListValidator.Builder() - .totalSize(expectedCollections.size()) - .exactlyContainsInAnyOrder(expectedCollections.stream().map(d -> d.read().block().getProperties().getResourceId()).collect(Collectors.toList())) + .totalSize(expectedContainers.size()) + .exactlyContainsInAnyOrder(expectedContainers.stream().map(d -> d.read().block().getProperties().getResourceId()).collect(Collectors.toList())) .numberOfPages(expectedPageSize) .pageSatisfy(0, new FeedResponseValidator.Builder() .requestChargeGreaterThanOrEqualTo(1.0).build()) @@ -97,7 +95,7 @@ public void queryAllCollections() throws Exception { } @Test(groups = { "simple" }, timeOut = TIMEOUT) - public void queryCollections_NoResults() throws Exception { + public void queryContainers_NoResults() throws Exception { String query = "SELECT * from root r where r.id = '2'"; CosmosQueryRequestOptions options = new CosmosQueryRequestOptions(); @@ -113,25 +111,27 @@ public void queryCollections_NoResults() throws Exception { } @Test(groups = { "simple" }, timeOut = TIMEOUT) - public void queryByRecreatCollectionWithSameName() throws Exception { + public void queryByRecreatContainerWithSameId() throws Exception { String testCollectionId = UUID.randomUUID().toString(); - CosmosContainerProperties collectionDefinition = getCollectionDefinition(testCollectionId); + CosmosContainerProperties containerProperties = getCollectionDefinition(testCollectionId); CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); String query = "SELECT * FROM r"; + CosmosAsyncDatabase database = this.getSharedCosmosDatabase(client); // step1: create container and do query - this.createdDatabase.createContainer(collectionDefinition).block(); - CosmosAsyncContainer container = this.createdDatabase.getContainer(collectionDefinition.getId()); + database.createContainer(containerProperties).block(); + CosmosAsyncContainer container = database.getContainer(containerProperties.getId()); InternalObjectNode document1 = getDocumentDefinition(); container.createItem(document1).block(); - container.queryItems(query, requestOptions, InternalObjectNode.class); + container.queryItems(query, requestOptions, InternalObjectNode.class).blockFirst(); // step2: delete the container created on step 1 safeDeleteCollection(container); - // step3: create a new collection with the same id as step 1 and do query - this.createdDatabase.createContainer(collectionDefinition).block(); - container = this.createdDatabase.getContainer(collectionDefinition.getId()); + // step3: create a new container with the same id as step 1 and do query + // This step is to check the collection cache will get refreshed + database.createContainer(containerProperties).block(); + container = database.getContainer(containerProperties.getId()); container.createItem(document1).block(); CosmosPagedFlux queryFlux = container.queryItems(query, requestOptions, InternalObjectNode.class); FeedResponseListValidator queryValidator = new FeedResponseListValidator.Builder() @@ -143,9 +143,8 @@ public void queryByRecreatCollectionWithSameName() throws Exception { safeDeleteCollection(container); } - @BeforeClass(groups = { "simple" }, timeOut = SETUP_TIMEOUT) - public void before_CollectionQueryTest() throws Exception { + public void before_ContainerQueryTest() throws Exception { client = getClientBuilder().buildAsyncClient(); createdDatabase = createDatabase(client, databaseId); @@ -154,8 +153,8 @@ public void before_CollectionQueryTest() throws Exception { paths.add("/mypk"); partitionKeyDef.setPaths(paths); - CosmosContainerProperties collection = new CosmosContainerProperties(UUID.randomUUID().toString(), partitionKeyDef); - createdCollections.add(createCollection(client, databaseId, collection)); + CosmosContainerProperties container = new CosmosContainerProperties(UUID.randomUUID().toString(), partitionKeyDef); + createdContainers.add(createCollection(client, databaseId, container)); } @AfterClass(groups = { "simple" }, timeOut = SHUTDOWN_TIMEOUT, alwaysRun = true) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java index 30805281f90c..a5e0e0682e93 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/TestSuiteBase.java @@ -521,7 +521,7 @@ static protected CosmosContainerProperties getCollectionDefinition() { static protected CosmosContainerProperties getCollectionDefinition(String collectionId) { PartitionKeyDefinition partitionKeyDef = new PartitionKeyDefinition(); - ArrayList paths = new ArrayList(); + ArrayList paths = new ArrayList<>(); paths.add("/mypk"); partitionKeyDef.setPaths(paths); From 6d609c626775c4ddf185b42ddc73427d69b46591 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 2 Jul 2020 01:22:21 -0700 Subject: [PATCH 03/15] fix --- .../src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java index 110a72181987..7dd94040a581 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java @@ -116,7 +116,7 @@ public void queryByRecreatContainerWithSameId() throws Exception { CosmosContainerProperties containerProperties = getCollectionDefinition(testCollectionId); CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); String query = "SELECT * FROM r"; - CosmosAsyncDatabase database = this.getSharedCosmosDatabase(client); + CosmosAsyncDatabase database = getSharedCosmosDatabase(client); // step1: create container and do query database.createContainer(containerProperties).block(); From 98495c2011262bcb440c5416522148a31a34dfa7 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 2 Jul 2020 09:00:37 -0700 Subject: [PATCH 04/15] clean --- .../com/azure/cosmos/implementation/BackoffRetryUtility.java | 2 +- .../com/azure/cosmos/implementation/ObservableHelper.java | 2 +- .../src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java index fa0971753f62..6a9d496ccbc8 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/BackoffRetryUtility.java @@ -45,7 +45,7 @@ static public Flux fluxExecuteRetry(Callable> callbackMethod, IRe try { return callbackMethod.call(); } catch (Exception e) { - return Mono.error(e); + return Flux.error(e); } }).retryWhen(RetryUtils.toRetryWhenFunc(retryPolicy)); } diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ObservableHelper.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ObservableHelper.java index 5be35ba27c29..2865eb351370 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ObservableHelper.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/ObservableHelper.java @@ -53,7 +53,7 @@ static public Flux fluxInlineIfPossibleAsObs(Callable> function, try { return function.call(); } catch (Exception e) { - return Mono.error(e); + return Flux.error(e); } }); diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java index 7dd94040a581..b57b77701368 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java @@ -112,8 +112,8 @@ public void queryContainers_NoResults() throws Exception { @Test(groups = { "simple" }, timeOut = TIMEOUT) public void queryByRecreatContainerWithSameId() throws Exception { - String testCollectionId = UUID.randomUUID().toString(); - CosmosContainerProperties containerProperties = getCollectionDefinition(testCollectionId); + String testContainerId = UUID.randomUUID().toString(); + CosmosContainerProperties containerProperties = getCollectionDefinition(testContainerId); CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); String query = "SELECT * FROM r"; CosmosAsyncDatabase database = getSharedCosmosDatabase(client); From 6386495ac7d95be252f56f94f3305ec9b5e92676 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 2 Jul 2020 14:47:40 -0700 Subject: [PATCH 05/15] resolve comments --- ...ContainerCreateDeleteWithSameNameTest.java | 307 ++++++++++++++++++ .../azure/cosmos/rx/ContainerQueryTest.java | 33 -- 2 files changed, 307 insertions(+), 33 deletions(-) create mode 100644 sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java new file mode 100644 index 000000000000..8e4410ff0688 --- /dev/null +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -0,0 +1,307 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.rx; + +import com.azure.cosmos.BridgeInternal; +import com.azure.cosmos.ChangeFeedProcessor; +import com.azure.cosmos.ChangeFeedProcessorBuilder; +import com.azure.cosmos.CosmosAsyncClient; +import com.azure.cosmos.CosmosAsyncContainer; +import com.azure.cosmos.CosmosAsyncDatabase; +import com.azure.cosmos.CosmosClientBuilder; +import com.azure.cosmos.implementation.FeedResponseListValidator; +import com.azure.cosmos.implementation.InternalObjectNode; +import com.azure.cosmos.models.ChangeFeedProcessorOptions; +import com.azure.cosmos.models.CosmosContainerProperties; +import com.azure.cosmos.models.CosmosContainerRequestOptions; +import com.azure.cosmos.models.CosmosItemRequestOptions; +import com.azure.cosmos.models.CosmosItemResponse; +import com.azure.cosmos.models.CosmosQueryRequestOptions; +import com.azure.cosmos.models.ModelBridgeInternal; +import com.azure.cosmos.models.PartitionKey; +import com.azure.cosmos.util.CosmosPagedFlux; +import com.fasterxml.jackson.databind.JsonNode; +import org.apache.commons.lang3.RandomStringUtils; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Factory; +import org.testng.annotations.Test; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.function.BiConsumer; +import java.util.function.Consumer; + +import static org.assertj.core.api.Assertions.assertThat; + +public class ContainerCreateDeleteWithSameNameTest extends TestSuiteBase{ + private final static int TIMEOUT = 300000; + private CosmosAsyncClient client; + private CosmosAsyncDatabase createdDatabase; + + @Factory(dataProvider = "clientBuilders") + public ContainerCreateDeleteWithSameNameTest(CosmosClientBuilder clientBuilder) { + super(clientBuilder); + this.subscriberValidationTimeout = TIMEOUT; + } + + @Test(groups = { "emulator" }, timeOut = TIMEOUT) + public void query() throws Exception { + String query = "SELECT * FROM r"; + + Consumer func = (container) -> { + InternalObjectNode docDefinition = getDocumentDefinition(); + container.createItem(docDefinition).block(); + + CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); + CosmosPagedFlux queryFlux = container.queryItems(query, requestOptions, InternalObjectNode.class); + FeedResponseListValidator queryValidator = new FeedResponseListValidator.Builder() + .totalSize(1) + .numberOfPages(1) + .build(); + validateQuerySuccess(queryFlux.byPage(10), queryValidator); + }; + + createDeleteContainerWithSameName(func); + } + + @Test(groups = { "emulator" }, timeOut = TIMEOUT) + public void readItem() throws Exception { + + Consumer func = (container) -> { + InternalObjectNode docDefinition = getDocumentDefinition(); + container.createItem(docDefinition).block(); + + Mono> responseMono = container.readItem(docDefinition.getId(), + new PartitionKey(ModelBridgeInternal.getObjectFromJsonSerializable(docDefinition, "mypk")), + new CosmosItemRequestOptions(), + InternalObjectNode.class); + + CosmosItemResponseValidator validator = + new CosmosItemResponseValidator.Builder>() + .withId(docDefinition.getId()) + .build(); + + this.validateItemSuccess(responseMono, validator); + }; + + createDeleteContainerWithSameName(func); + } + + @Test(groups = { "emulator" }, timeOut = TIMEOUT) + public void deleteItem() throws Exception { + + Consumer func = (container) -> { + InternalObjectNode docDefinition = getDocumentDefinition(); + container.createItem(docDefinition).block(); + + Mono> deleteObservable = container.deleteItem( + docDefinition.getId(), + new PartitionKey(ModelBridgeInternal.getObjectFromJsonSerializable(docDefinition, "mypk")), + new CosmosItemRequestOptions()); + + CosmosItemResponseValidator validator = + new CosmosItemResponseValidator.Builder>() + .nullResource() + .build(); + this.validateItemSuccess(deleteObservable, validator); + }; + + createDeleteContainerWithSameName(func); + } + + + @Test(groups = { "emulator" }, timeOut = TIMEOUT) + public void replaceItem() throws Exception { + + Consumer func = (container) -> { + InternalObjectNode docDefinition = getDocumentDefinition(); + docDefinition = container.createItem(docDefinition).block().getItem(); + + String newPropValue = UUID.randomUUID().toString(); + BridgeInternal.setProperty(docDefinition, "newProp", newPropValue); + + Mono> readObservable = container.upsertItem(docDefinition, new CosmosItemRequestOptions()); + + // Validate result + CosmosItemResponseValidator validator = + new CosmosItemResponseValidator.Builder>() + .withProperty("newProp", newPropValue) + .build(); + + this.validateItemSuccess(readObservable, validator); + }; + + createDeleteContainerWithSameName(func); + } + + @Test(groups = { "emulator" }, timeOut = TIMEOUT) + public void changeFeed() throws Exception { + + BiConsumer func = (feedContainer, leaseContainer) -> { + String hostName = RandomStringUtils.randomAlphabetic(6); + int CHANGE_FEED_PROCESSOR_TIMEOUT = 5000; + final int FEED_COUNT = 10; + List createdDocuments = new ArrayList<>(); + Map receivedDocuments = new ConcurrentHashMap<>(); + + setupReadFeedDocuments(createdDocuments, feedContainer, FEED_COUNT); + + ChangeFeedProcessor changeFeedProcessor = new ChangeFeedProcessorBuilder() + .hostName(hostName) + .handleChanges((docs) -> { + for (JsonNode item : docs) { + receivedDocuments.put(item.get("id").asText(), item); + } + }) + .feedContainer(feedContainer) + .leaseContainer(leaseContainer) + .options(new ChangeFeedProcessorOptions() + .setLeaseRenewInterval(Duration.ofSeconds(20)) + .setLeaseAcquireInterval(Duration.ofSeconds(10)) + .setLeaseExpirationInterval(Duration.ofSeconds(30)) + .setFeedPollDelay(Duration.ofSeconds(2)) + .setLeasePrefix("TEST") + .setMaxItemCount(10) + .setStartFromBeginning(true) + .setMaxScaleCount(0) // unlimited + ) + .buildChangeFeedProcessor(); + + try { + changeFeedProcessor.start().subscribeOn(Schedulers.elastic()) + .timeout(Duration.ofMillis(2 * CHANGE_FEED_PROCESSOR_TIMEOUT)) + .subscribe(); + } catch (Exception ex) { + throw ex; + } + + // Wait for the feed processor to receive and process the documents. + try { + Thread.sleep(2 * CHANGE_FEED_PROCESSOR_TIMEOUT); + assertThat(changeFeedProcessor.isStarted()).as("Change Feed Processor instance is running").isTrue(); + + changeFeedProcessor.stop().subscribeOn(Schedulers.elastic()).timeout(Duration.ofMillis(CHANGE_FEED_PROCESSOR_TIMEOUT)).subscribe(); + + for (InternalObjectNode item : createdDocuments) { + assertThat(receivedDocuments.containsKey(item.getId())).as("Document with getId: " + item.getId()).isTrue(); + } + + // Wait for the feed processor to shutdown. + Thread.sleep(CHANGE_FEED_PROCESSOR_TIMEOUT); + + } catch (InterruptedException e) { + throw new RuntimeException("Interrupted exception", e); + } + }; + + changeFeedCreateDeleteContainerWithSameName(func); + } + + @BeforeClass(groups = { "emulator" }, timeOut = SETUP_TIMEOUT) + public void before_ContainerCreateDeleteWithSameNameTest() throws Exception { + client = getClientBuilder().buildAsyncClient(); + createdDatabase = getSharedCosmosDatabase(client); + } + + @AfterClass(groups = { "emulator" }, timeOut = SETUP_TIMEOUT) + public void after_ContainerCreateDeleteWithSameNameTest() throws Exception { + safeDeleteAllCollections(createdDatabase); + } + + private void createDeleteContainerWithSameName(Consumer validateFunc) { + CosmosAsyncContainer container = null; + try{ + // step1: create container + String testContainerId = UUID.randomUUID().toString(); + CosmosContainerProperties containerProperties = getCollectionDefinition(testContainerId); + container = createCollection(this.createdDatabase, containerProperties, new CosmosContainerRequestOptions()); + + // Step2: execute func + validateFunc.accept(container); + + // step3: delete the container + safeDeleteCollection(container); + + // step4: recreate the container with same id as step1 + container = createCollection(this.createdDatabase, containerProperties, new CosmosContainerRequestOptions()); + + // step5: same as step2. + // This part will confirm the cache refreshed correctly + validateFunc.accept(container); + } + finally { + safeDeleteCollection(container); + } + } + + private void changeFeedCreateDeleteContainerWithSameName(BiConsumer validateFunc) { + CosmosAsyncContainer feedContainer = null; + CosmosAsyncContainer leaseContainer = null; + + try { + // step1: create feed container and lease container + String feedContainerId = UUID.randomUUID().toString(); + CosmosContainerProperties feedContainerProperties = getCollectionDefinition(feedContainerId); + feedContainer = createCollection(this.createdDatabase,feedContainerProperties, new CosmosContainerRequestOptions()); + + String leaseContainerId = UUID.randomUUID().toString(); + CosmosContainerProperties leaseContainerProperties = getCollectionDefinition(leaseContainerId); + leaseContainer = createLeaseContainer(leaseContainerProperties.getId()); + + + // Step2: execute func + validateFunc.accept(feedContainer, leaseContainer); + + // step3: delete the containers + safeDeleteCollection(feedContainer); + safeDeleteCollection(leaseContainer); + + // step4: recreate the feed container and lease container with same ids as step1 + feedContainer = createCollection(this.createdDatabase, feedContainerProperties, new CosmosContainerRequestOptions()); + leaseContainer = createLeaseContainer(leaseContainerProperties.getId()); + + // step5: same as step2. + // This part will confirm the cache refreshed correctly + validateFunc.accept(feedContainer, leaseContainer); + } + finally { + safeDeleteCollection(feedContainer); + safeDeleteCollection(leaseContainer); + } + } + + private static InternalObjectNode getDocumentDefinition() { + String uuid = UUID.randomUUID().toString(); + InternalObjectNode doc = new InternalObjectNode(String.format("{ " + + "\"id\": \"%s\", " + + "\"mypk\": \"%s\", " + + "}" + , uuid, uuid)); + return doc; + } + + private void setupReadFeedDocuments(List createdDocuments, CosmosAsyncContainer feedCollection, long count) { + List docDefList = new ArrayList<>(); + + for(int i = 0; i < count; i++) { + docDefList.add(getDocumentDefinition()); + } + + createdDocuments.addAll(bulkInsertBlocking(feedCollection, docDefList)); + waitIfNeededForReplicasToCatchUp(getClientBuilder()); + } + + private CosmosAsyncContainer createLeaseContainer(String conatinerId) { + CosmosContainerRequestOptions options = new CosmosContainerRequestOptions(); + CosmosContainerProperties collectionDefinition = new CosmosContainerProperties(conatinerId, "/id"); + return createCollection(createdDatabase, collectionDefinition, options); + } +} diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java index b57b77701368..2e2fb261377e 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerQueryTest.java @@ -110,39 +110,6 @@ public void queryContainers_NoResults() throws Exception { validateQuerySuccess(queryObservable.byPage(), validator); } - @Test(groups = { "simple" }, timeOut = TIMEOUT) - public void queryByRecreatContainerWithSameId() throws Exception { - String testContainerId = UUID.randomUUID().toString(); - CosmosContainerProperties containerProperties = getCollectionDefinition(testContainerId); - CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); - String query = "SELECT * FROM r"; - CosmosAsyncDatabase database = getSharedCosmosDatabase(client); - - // step1: create container and do query - database.createContainer(containerProperties).block(); - CosmosAsyncContainer container = database.getContainer(containerProperties.getId()); - InternalObjectNode document1 = getDocumentDefinition(); - container.createItem(document1).block(); - container.queryItems(query, requestOptions, InternalObjectNode.class).blockFirst(); - - // step2: delete the container created on step 1 - safeDeleteCollection(container); - - // step3: create a new container with the same id as step 1 and do query - // This step is to check the collection cache will get refreshed - database.createContainer(containerProperties).block(); - container = database.getContainer(containerProperties.getId()); - container.createItem(document1).block(); - CosmosPagedFlux queryFlux = container.queryItems(query, requestOptions, InternalObjectNode.class); - FeedResponseListValidator queryValidator = new FeedResponseListValidator.Builder() - .totalSize(1) - .numberOfPages(1) - .build(); - validateQuerySuccess(queryFlux.byPage(10), queryValidator); - - safeDeleteCollection(container); - } - @BeforeClass(groups = { "simple" }, timeOut = SETUP_TIMEOUT) public void before_ContainerQueryTest() throws Exception { client = getClientBuilder().buildAsyncClient(); From 88d65699caf96f1c963f87877730bc2ea82bc13c Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 2 Jul 2020 14:51:01 -0700 Subject: [PATCH 06/15] remove extra line --- .../azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index 8e4410ff0688..b63f3d93f8d1 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -119,7 +119,7 @@ public void deleteItem() throws Exception { @Test(groups = { "emulator" }, timeOut = TIMEOUT) - public void replaceItem() throws Exception { + public void upsertItem() throws Exception { Consumer func = (container) -> { InternalObjectNode docDefinition = getDocumentDefinition(); @@ -256,7 +256,6 @@ private void changeFeedCreateDeleteContainerWithSameName(BiConsumer Date: Thu, 2 Jul 2020 16:43:56 -0700 Subject: [PATCH 07/15] resolve comments --- ...ContainerCreateDeleteWithSameNameTest.java | 157 +++++++++++------- 1 file changed, 99 insertions(+), 58 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index b63f3d93f8d1..a826088a6340 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -3,7 +3,6 @@ package com.azure.cosmos.rx; -import com.azure.cosmos.BridgeInternal; import com.azure.cosmos.ChangeFeedProcessor; import com.azure.cosmos.ChangeFeedProcessorBuilder; import com.azure.cosmos.CosmosAsyncClient; @@ -11,17 +10,18 @@ import com.azure.cosmos.CosmosAsyncDatabase; import com.azure.cosmos.CosmosClientBuilder; import com.azure.cosmos.implementation.FeedResponseListValidator; -import com.azure.cosmos.implementation.InternalObjectNode; +import com.azure.cosmos.implementation.Utils; import com.azure.cosmos.models.ChangeFeedProcessorOptions; import com.azure.cosmos.models.CosmosContainerProperties; import com.azure.cosmos.models.CosmosContainerRequestOptions; import com.azure.cosmos.models.CosmosItemRequestOptions; import com.azure.cosmos.models.CosmosItemResponse; import com.azure.cosmos.models.CosmosQueryRequestOptions; -import com.azure.cosmos.models.ModelBridgeInternal; import com.azure.cosmos.models.PartitionKey; import com.azure.cosmos.util.CosmosPagedFlux; +import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; import org.apache.commons.lang3.RandomStringUtils; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -34,6 +34,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Random; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.function.BiConsumer; @@ -41,28 +42,28 @@ import static org.assertj.core.api.Assertions.assertThat; -public class ContainerCreateDeleteWithSameNameTest extends TestSuiteBase{ +public class ContainerCreateDeleteWithSameNameTest extends TestSuiteBase { private final static int TIMEOUT = 300000; private CosmosAsyncClient client; private CosmosAsyncDatabase createdDatabase; - @Factory(dataProvider = "clientBuilders") + @Factory(dataProvider = "clientBuildersWithSessionConsistency") public ContainerCreateDeleteWithSameNameTest(CosmosClientBuilder clientBuilder) { super(clientBuilder); this.subscriberValidationTimeout = TIMEOUT; } - @Test(groups = { "emulator" }, timeOut = TIMEOUT) + @Test(groups = {"emulator"}, timeOut = TIMEOUT) public void query() throws Exception { String query = "SELECT * FROM r"; Consumer func = (container) -> { - InternalObjectNode docDefinition = getDocumentDefinition(); + TestObject docDefinition = getDocumentDefinition(); container.createItem(docDefinition).block(); CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); - CosmosPagedFlux queryFlux = container.queryItems(query, requestOptions, InternalObjectNode.class); - FeedResponseListValidator queryValidator = new FeedResponseListValidator.Builder() + CosmosPagedFlux queryFlux = container.queryItems(query, requestOptions, TestObject.class); + FeedResponseListValidator queryValidator = new FeedResponseListValidator.Builder() .totalSize(1) .numberOfPages(1) .build(); @@ -72,20 +73,20 @@ public void query() throws Exception { createDeleteContainerWithSameName(func); } - @Test(groups = { "emulator" }, timeOut = TIMEOUT) + @Test(groups = {"emulator"}, timeOut = TIMEOUT) public void readItem() throws Exception { Consumer func = (container) -> { - InternalObjectNode docDefinition = getDocumentDefinition(); + TestObject docDefinition = getDocumentDefinition(); container.createItem(docDefinition).block(); - Mono> responseMono = container.readItem(docDefinition.getId(), - new PartitionKey(ModelBridgeInternal.getObjectFromJsonSerializable(docDefinition, "mypk")), + Mono> responseMono = container.readItem(docDefinition.getId(), + new PartitionKey(docDefinition.getMypk()), new CosmosItemRequestOptions(), - InternalObjectNode.class); + TestObject.class); CosmosItemResponseValidator validator = - new CosmosItemResponseValidator.Builder>() + new CosmosItemResponseValidator.Builder>() .withId(docDefinition.getId()) .build(); @@ -95,20 +96,20 @@ public void readItem() throws Exception { createDeleteContainerWithSameName(func); } - @Test(groups = { "emulator" }, timeOut = TIMEOUT) + @Test(groups = {"emulator"}, timeOut = TIMEOUT) public void deleteItem() throws Exception { Consumer func = (container) -> { - InternalObjectNode docDefinition = getDocumentDefinition(); + TestObject docDefinition = getDocumentDefinition(); container.createItem(docDefinition).block(); Mono> deleteObservable = container.deleteItem( docDefinition.getId(), - new PartitionKey(ModelBridgeInternal.getObjectFromJsonSerializable(docDefinition, "mypk")), + new PartitionKey(docDefinition.getMypk()), new CosmosItemRequestOptions()); CosmosItemResponseValidator validator = - new CosmosItemResponseValidator.Builder>() + new CosmosItemResponseValidator.Builder>() .nullResource() .build(); this.validateItemSuccess(deleteObservable, validator); @@ -117,23 +118,21 @@ public void deleteItem() throws Exception { createDeleteContainerWithSameName(func); } - - @Test(groups = { "emulator" }, timeOut = TIMEOUT) + @Test(groups = {"emulator"}, timeOut = TIMEOUT) public void upsertItem() throws Exception { Consumer func = (container) -> { - InternalObjectNode docDefinition = getDocumentDefinition(); + TestObject docDefinition = getDocumentDefinition(); docDefinition = container.createItem(docDefinition).block().getItem(); - String newPropValue = UUID.randomUUID().toString(); - BridgeInternal.setProperty(docDefinition, "newProp", newPropValue); + docDefinition.setProp(UUID.randomUUID().toString()); - Mono> readObservable = container.upsertItem(docDefinition, new CosmosItemRequestOptions()); + Mono> readObservable = container.upsertItem(docDefinition, new CosmosItemRequestOptions()); // Validate result CosmosItemResponseValidator validator = - new CosmosItemResponseValidator.Builder>() - .withProperty("newProp", newPropValue) + new CosmosItemResponseValidator.Builder>() + .withProperty("prop", String.valueOf(docDefinition.getProp())) .build(); this.validateItemSuccess(readObservable, validator); @@ -142,15 +141,16 @@ public void upsertItem() throws Exception { createDeleteContainerWithSameName(func); } - @Test(groups = { "emulator" }, timeOut = TIMEOUT) + @Test(groups = {"emulator"}, timeOut = TIMEOUT) public void changeFeed() throws Exception { + ObjectMapper objectMapper = Utils.getSimpleObjectMapper(); BiConsumer func = (feedContainer, leaseContainer) -> { String hostName = RandomStringUtils.randomAlphabetic(6); int CHANGE_FEED_PROCESSOR_TIMEOUT = 5000; final int FEED_COUNT = 10; - List createdDocuments = new ArrayList<>(); - Map receivedDocuments = new ConcurrentHashMap<>(); + List createdDocuments = new ArrayList<>(); + Map receivedDocuments = new ConcurrentHashMap<>(); setupReadFeedDocuments(createdDocuments, feedContainer, FEED_COUNT); @@ -158,7 +158,12 @@ public void changeFeed() throws Exception { .hostName(hostName) .handleChanges((docs) -> { for (JsonNode item : docs) { - receivedDocuments.put(item.get("id").asText(), item); + try { + TestObject obj = objectMapper.treeToValue(item, TestObject.class); + receivedDocuments.put(obj.getId(), obj); + } catch (JsonProcessingException e) { + e.printStackTrace(); + } } }) .feedContainer(feedContainer) @@ -190,7 +195,7 @@ public void changeFeed() throws Exception { changeFeedProcessor.stop().subscribeOn(Schedulers.elastic()).timeout(Duration.ofMillis(CHANGE_FEED_PROCESSOR_TIMEOUT)).subscribe(); - for (InternalObjectNode item : createdDocuments) { + for (TestObject item : createdDocuments) { assertThat(receivedDocuments.containsKey(item.getId())).as("Document with getId: " + item.getId()).isTrue(); } @@ -205,20 +210,20 @@ public void changeFeed() throws Exception { changeFeedCreateDeleteContainerWithSameName(func); } - @BeforeClass(groups = { "emulator" }, timeOut = SETUP_TIMEOUT) + @BeforeClass(groups = {"emulator"}, timeOut = SETUP_TIMEOUT) public void before_ContainerCreateDeleteWithSameNameTest() throws Exception { client = getClientBuilder().buildAsyncClient(); createdDatabase = getSharedCosmosDatabase(client); } - @AfterClass(groups = { "emulator" }, timeOut = SETUP_TIMEOUT) + @AfterClass(groups = {"emulator"}, timeOut = SETUP_TIMEOUT) public void after_ContainerCreateDeleteWithSameNameTest() throws Exception { safeDeleteAllCollections(createdDatabase); } - private void createDeleteContainerWithSameName(Consumer validateFunc) { + private void createDeleteContainerWithSameName(Consumer validateFunc) throws InterruptedException { CosmosAsyncContainer container = null; - try{ + try { // step1: create container String testContainerId = UUID.randomUUID().toString(); CosmosContainerProperties containerProperties = getCollectionDefinition(testContainerId); @@ -229,6 +234,7 @@ private void createDeleteContainerWithSameName(Consumer void createDeleteContainerWithSameName(Consumer void changeFeedCreateDeleteContainerWithSameName(BiConsumer validateFunc) { + private void changeFeedCreateDeleteContainerWithSameName(BiConsumer validateFunc) throws InterruptedException { CosmosAsyncContainer feedContainer = null; CosmosAsyncContainer leaseContainer = null; @@ -250,7 +255,7 @@ private void changeFeedCreateDeleteContainerWithSameName(BiConsumer void changeFeedCreateDeleteContainerWithSameName(BiConsumer createdDocuments, CosmosAsyncContainer feedCollection, long count) { - List docDefList = new ArrayList<>(); + private void setupReadFeedDocuments(List createdDocuments, CosmosAsyncContainer feedContainer, long count) { + List docDefList = new ArrayList<>(); - for(int i = 0; i < count; i++) { + for (int i = 0; i < count; i++) { docDefList.add(getDocumentDefinition()); } - createdDocuments.addAll(bulkInsertBlocking(feedCollection, docDefList)); + createdDocuments.addAll(bulkInsertBlocking(feedContainer, docDefList)); waitIfNeededForReplicasToCatchUp(getClientBuilder()); } @@ -303,4 +305,43 @@ private CosmosAsyncContainer createLeaseContainer(String conatinerId) { CosmosContainerProperties collectionDefinition = new CosmosContainerProperties(conatinerId, "/id"); return createCollection(createdDatabase, collectionDefinition, options); } + + static class TestObject { + String id; + String mypk; + String prop; + + public TestObject() { + } + + public TestObject(String id, String mypk, String prop) { + this.id = id; + this.mypk = mypk; + this.prop = prop; + } + + public String getId() { + return id; + } + + public void setId(String id) { + this.id = id; + } + + public String getMypk() { + return mypk; + } + + public void setMypk(String mypk) { + this.mypk = mypk; + } + + public String getProp() { + return prop; + } + + public void setProp(String prop) { + this.prop = prop; + } + } } From e89fed935a650380aca656f0bdedef56a322f491 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Thu, 2 Jul 2020 16:46:30 -0700 Subject: [PATCH 08/15] clean code --- .../azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index a826088a6340..1dc44a69476a 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -132,7 +132,7 @@ public void upsertItem() throws Exception { // Validate result CosmosItemResponseValidator validator = new CosmosItemResponseValidator.Builder>() - .withProperty("prop", String.valueOf(docDefinition.getProp())) + .withProperty("prop", docDefinition.getProp()) .build(); this.validateItemSuccess(readObservable, validator); From cd39ceb9a8a5a00129471f89642ff1b5014d1be9 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Mon, 6 Jul 2020 12:07:26 -0700 Subject: [PATCH 09/15] add comments --- .../cosmos/implementation/RxDocumentClientImpl.java | 3 +++ .../rx/ContainerCreateDeleteWithSameNameTest.java | 12 ++++++++---- 2 files changed, 11 insertions(+), 4 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java index 54f07f99a83b..a004f7ab68bc 100644 --- a/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java +++ b/sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/RxDocumentClientImpl.java @@ -570,6 +570,9 @@ private Flux> createQuery( UUID activityId = Utils.randomUUID(); IDocumentQueryClient queryClient = documentQueryClientImpl(RxDocumentClientImpl.this); + // Trying to put this logic as low as the query pipeline + // Since for parallelQuery, each partition will have its own request, so at this point, there will be no request associate with this retry policy. + // For default document context, it already wired up InvalidPartitionExceptionRetry, but there is no harm to wire it again here InvalidPartitionExceptionRetryPolicy invalidPartitionExceptionRetryPolicy = new InvalidPartitionExceptionRetryPolicy( this.collectionCache, null, diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index 1dc44a69476a..690ea2881513 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -22,6 +22,7 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import jdk.jfr.Timespan; import org.apache.commons.lang3.RandomStringUtils; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; @@ -44,6 +45,9 @@ public class ContainerCreateDeleteWithSameNameTest extends TestSuiteBase { private final static int TIMEOUT = 300000; + // Delete collections in emulator usually takes a long time, + // so to avoid get 500 back, we are adding delay for creating the collection with same name, since in this case we want to test 410/1000 + private final static int COLLECTION_RECREATION_TIME_DELAY = 2000; private CosmosAsyncClient client; private CosmosAsyncDatabase createdDatabase; @@ -147,8 +151,8 @@ public void changeFeed() throws Exception { ObjectMapper objectMapper = Utils.getSimpleObjectMapper(); BiConsumer func = (feedContainer, leaseContainer) -> { String hostName = RandomStringUtils.randomAlphabetic(6); - int CHANGE_FEED_PROCESSOR_TIMEOUT = 5000; - final int FEED_COUNT = 10; + int CHANGE_FEED_PROCESSOR_TIMEOUT = 10000; + final int FEED_COUNT = 5; List createdDocuments = new ArrayList<>(); Map receivedDocuments = new ConcurrentHashMap<>(); @@ -234,7 +238,7 @@ private void createDeleteContainerWithSameName(Consumer void changeFeedCreateDeleteContainerWithSameName(BiConsumer Date: Mon, 6 Jul 2020 12:16:22 -0700 Subject: [PATCH 10/15] update the comment a little --- .../azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index 690ea2881513..28d44af96914 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -45,7 +45,7 @@ public class ContainerCreateDeleteWithSameNameTest extends TestSuiteBase { private final static int TIMEOUT = 300000; - // Delete collections in emulator usually takes a long time, + // Delete collections in emulator is not instant, // so to avoid get 500 back, we are adding delay for creating the collection with same name, since in this case we want to test 410/1000 private final static int COLLECTION_RECREATION_TIME_DELAY = 2000; private CosmosAsyncClient client; From 306a1512ab53e16d710afc3c07568ffe2c9d8ad2 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Mon, 6 Jul 2020 12:27:15 -0700 Subject: [PATCH 11/15] fix compilcation --- .../azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index 28d44af96914..9ffa19935298 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -22,7 +22,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; -import jdk.jfr.Timespan; import org.apache.commons.lang3.RandomStringUtils; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; From 4dc3ccc1d9a879684823ddda06ca7a883d44dd93 Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Mon, 6 Jul 2020 22:18:26 -0700 Subject: [PATCH 12/15] update test --- .../azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index 9ffa19935298..ab696f9e8fc6 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -150,7 +150,7 @@ public void changeFeed() throws Exception { ObjectMapper objectMapper = Utils.getSimpleObjectMapper(); BiConsumer func = (feedContainer, leaseContainer) -> { String hostName = RandomStringUtils.randomAlphabetic(6); - int CHANGE_FEED_PROCESSOR_TIMEOUT = 10000; + int CHANGE_FEED_PROCESSOR_TIMEOUT = 5000; final int FEED_COUNT = 5; List createdDocuments = new ArrayList<>(); Map receivedDocuments = new ConcurrentHashMap<>(); @@ -269,6 +269,7 @@ private void changeFeedCreateDeleteContainerWithSameName(BiConsumer Date: Tue, 7 Jul 2020 00:10:13 -0700 Subject: [PATCH 13/15] update tests --- .../cosmos/rx/ContainerCreateDeleteWithSameNameTest.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index ab696f9e8fc6..0ecb1ac60230 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -198,10 +198,6 @@ public void changeFeed() throws Exception { changeFeedProcessor.stop().subscribeOn(Schedulers.elastic()).timeout(Duration.ofMillis(CHANGE_FEED_PROCESSOR_TIMEOUT)).subscribe(); - for (TestObject item : createdDocuments) { - assertThat(receivedDocuments.containsKey(item.getId())).as("Document with getId: " + item.getId()).isTrue(); - } - // Wait for the feed processor to shutdown. Thread.sleep(CHANGE_FEED_PROCESSOR_TIMEOUT); @@ -285,7 +281,6 @@ private void changeFeedCreateDeleteContainerWithSameName(BiConsumer Date: Tue, 7 Jul 2020 00:20:02 -0700 Subject: [PATCH 14/15] update tests --- .../cosmos/rx/ContainerCreateDeleteWithSameNameTest.java | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index 0ecb1ac60230..e09902c416dd 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -193,9 +193,16 @@ public void changeFeed() throws Exception { // Wait for the feed processor to receive and process the documents. try { - Thread.sleep(2 * CHANGE_FEED_PROCESSOR_TIMEOUT); assertThat(changeFeedProcessor.isStarted()).as("Change Feed Processor instance is running").isTrue(); + long remainingWork = 2 * CHANGE_FEED_PROCESSOR_TIMEOUT; + while (remainingWork > 0 && receivedDocuments.size() < createdDocuments.size()) { + remainingWork -= 100; + Thread.sleep(100); + } + + assertThat(remainingWork >= 0).as("Failed to receive all the feed documents").isTrue(); + changeFeedProcessor.stop().subscribeOn(Schedulers.elastic()).timeout(Duration.ofMillis(CHANGE_FEED_PROCESSOR_TIMEOUT)).subscribe(); // Wait for the feed processor to shutdown. From 472abc7d56115b9e1058b341f73595d31d16e42f Mon Sep 17 00:00:00 2001 From: Annie Liang Date: Tue, 7 Jul 2020 00:25:34 -0700 Subject: [PATCH 15/15] update tests --- .../azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java | 1 + 1 file changed, 1 insertion(+) diff --git a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java index e09902c416dd..24380fcdb4f3 100644 --- a/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -193,6 +193,7 @@ public void changeFeed() throws Exception { // Wait for the feed processor to receive and process the documents. try { + Thread.sleep(2 * CHANGE_FEED_PROCESSOR_TIMEOUT); assertThat(changeFeedProcessor.isStarted()).as("Change Feed Processor instance is running").isTrue(); long remainingWork = 2 * CHANGE_FEED_PROCESSOR_TIMEOUT;