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..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 @@ -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 Flux.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..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 @@ -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 @@ -56,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 { @@ -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..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 @@ -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 Flux.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..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 @@ -560,18 +560,42 @@ 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); + + // 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, + 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/test/java/com/azure/cosmos/AzureKeyCredentialTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/AzureKeyCredentialTest.java index 9e161fe0662e..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; @@ -65,17 +64,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..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; @@ -102,17 +100,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..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; @@ -53,19 +51,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..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; @@ -106,17 +103,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/ContainerCreateDeleteWithSameNameTest.java b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java new file mode 100644 index 000000000000..24380fcdb4f3 --- /dev/null +++ b/sdk/cosmos/azure-cosmos/src/test/java/com/azure/cosmos/rx/ContainerCreateDeleteWithSameNameTest.java @@ -0,0 +1,354 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.azure.cosmos.rx; + +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.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.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; +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.Random; +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; + // 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; + private CosmosAsyncDatabase createdDatabase; + + @Factory(dataProvider = "clientBuildersWithSessionConsistency") + 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) -> { + TestObject docDefinition = getDocumentDefinition(); + container.createItem(docDefinition).block(); + + CosmosQueryRequestOptions requestOptions = new CosmosQueryRequestOptions(); + CosmosPagedFlux queryFlux = container.queryItems(query, requestOptions, TestObject.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) -> { + TestObject docDefinition = getDocumentDefinition(); + container.createItem(docDefinition).block(); + + Mono> responseMono = container.readItem(docDefinition.getId(), + new PartitionKey(docDefinition.getMypk()), + new CosmosItemRequestOptions(), + TestObject.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) -> { + TestObject docDefinition = getDocumentDefinition(); + container.createItem(docDefinition).block(); + + Mono> deleteObservable = container.deleteItem( + docDefinition.getId(), + new PartitionKey(docDefinition.getMypk()), + new CosmosItemRequestOptions()); + + CosmosItemResponseValidator validator = + new CosmosItemResponseValidator.Builder>() + .nullResource() + .build(); + this.validateItemSuccess(deleteObservable, validator); + }; + + createDeleteContainerWithSameName(func); + } + + @Test(groups = {"emulator"}, timeOut = TIMEOUT) + public void upsertItem() throws Exception { + + Consumer func = (container) -> { + TestObject docDefinition = getDocumentDefinition(); + docDefinition = container.createItem(docDefinition).block().getItem(); + + docDefinition.setProp(UUID.randomUUID().toString()); + + Mono> readObservable = container.upsertItem(docDefinition, new CosmosItemRequestOptions()); + + // Validate result + CosmosItemResponseValidator validator = + new CosmosItemResponseValidator.Builder>() + .withProperty("prop", docDefinition.getProp()) + .build(); + + this.validateItemSuccess(readObservable, validator); + }; + + createDeleteContainerWithSameName(func); + } + + @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 = 5; + 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) { + try { + TestObject obj = objectMapper.treeToValue(item, TestObject.class); + receivedDocuments.put(obj.getId(), obj); + } catch (JsonProcessingException e) { + e.printStackTrace(); + } + } + }) + .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(); + + 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. + 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) throws InterruptedException { + 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); + Thread.sleep(COLLECTION_RECREATION_TIME_DELAY); + + // 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) throws InterruptedException { + 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 lease container + safeDeleteCollection(leaseContainer); + truncateCollection(feedContainer); + Thread.sleep(COLLECTION_RECREATION_TIME_DELAY); + + // step4: recreate the lease container and lease container with same ids as step1 + 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 TestObject getDocumentDefinition() { + return new TestObject( + UUID.randomUUID().toString(), + UUID.randomUUID().toString(), + UUID.randomUUID().toString() + ); + } + + private void setupReadFeedDocuments(List createdDocuments, CosmosAsyncContainer feedContainer, long count) { + List docDefList = new ArrayList<>(); + + for (int i = 0; i < count; i++) { + docDefList.add(getDocumentDefinition()); + } + + createdDocuments.addAll(bulkInsertBlocking(feedContainer, docDefList)); + waitIfNeededForReplicasToCatchUp(getClientBuilder()); + } + + private CosmosAsyncContainer createLeaseContainer(String conatinerId) { + CosmosContainerRequestOptions options = new CosmosContainerRequestOptions(); + 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; + } + } +} 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 67% 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 1349df5d905b..2e2fb261377e 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 @@ -6,11 +6,12 @@ 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.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 org.apache.commons.lang3.StringUtils; @@ -26,39 +27,39 @@ import static org.assertj.core.api.Assertions.assertThat; -public class CollectionQueryTest extends TestSuiteBase { +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()) @@ -68,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"; @@ -76,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()) @@ -94,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(); @@ -110,7 +111,7 @@ public void queryCollections_NoResults() throws Exception { } @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); @@ -119,8 +120,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) @@ -128,4 +129,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..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 @@ -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(); + 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; }