ChangeFeed pull model and switching CF processor to FeedRange

eng/common/pipelines/templates/steps/cosmos-emulator.yml

@@ -9,8 +9,92 @@ steps:
       Write-Host "Target Dir: $targetDir"
       msiexec /a ${{ parameters.EmulatorMsiUrl }} TARGETDIR=$targetDir /qn | wait-process
     displayName: Download and Extract Public Cosmos DB Emulator
+  - powershell: |
+      Write-Host "Deleting Cosmos DB Emulator data"
+      if (Test-Path $Env:LOCALAPPDATA\CosmosDbEmulator) { Remove-Item -Recurse -Force $Env:LOCALAPPDATA\CosmosDbEmulator }
+    displayName: Delete Cosmos DB Emulator data
+  - powershell: |
+      Write-Host "Getting Cosmos DB Emulator Version"
+      $ProductName = "Azure Cosmos DB Emulator"
+      $Emulator = (Join-Path $env:temp (Join-Path $ProductName "Microsoft.Azure.Cosmos.Emulator.exe"))
+      $fileVersion = Get-ChildItem $Emulator
+      Write-Host $Emulator $fileVersion.VersionInfo
+    displayName: Get Cosmos DB Emulator Version
   - powershell: |
       Write-Host "Launching Cosmos DB Emulator"
-      Import-Module "$env:temp\Azure Cosmos DB Emulator\PSModules\Microsoft.Azure.CosmosDB.Emulator"
-      Start-CosmosDbEmulator -NoUI ${{ parameters.StartParameters }}
+      $ProductName = "Azure Cosmos DB Emulator"
+      $Emulator = (Join-Path $env:temp (Join-Path $ProductName "Microsoft.Azure.Cosmos.Emulator.exe"))
+      if (!(Test-Path $Emulator)) {
+        Write-Error "The emulator is not installed where expected at '$Emulator'"
+        return
+      }
+
+      $process = Start-Process $Emulator -ArgumentList "/getstatus" -PassThru -Wait
+
+      switch ($process.ExitCode) {
+        1 {
+          Write-Host "The emulator is already starting"
+          return
+        }
+        2 {
+          Write-Host "The emulator is already running"
+          return
+        }
+        3 {
+          Write-Host "The emulator is stopped"
+        }
+        default {
+          Write-Host "Unrecognized exit code $process.ExitCode"
+          return
+        }
+      }
+
+      $argumentList = ""
+      if (-not [string]::IsNullOrEmpty("${{ parameters.StartParameters }}")) {
+          $argumentList += , "${{ parameters.StartParameters }}"
+      }
+
+      Write-Host "Starting emulator process: $Emulator $argumentList"
+      $process=Start-Process $Emulator -ArgumentList $argumentList -ErrorAction Stop -PassThru
+      Write-Host "Emulator process started: $($process.Name), $($process.FileVersion)"
+      $Timeout = 600
+      $result="NotYetStarted"
+      $complete = if ($Timeout -gt 0) {
+        $start = [DateTimeOffset]::Now
+        $stop = $start.AddSeconds($Timeout)
+        {
+          $result -eq "Running" -or [DateTimeOffset]::Now -ge $stop
+        }
+      }
+      else {
+        {
+          $result -eq "Running"
+        }
+      }
+
+      do {
+        $process = Start-Process $Emulator -ArgumentList "/getstatus" -PassThru -Wait
+
+        switch ($process.ExitCode) {
+          1 {
+            Write-Host "The emulator is starting"
+          }
+          2 {
+            Write-Host "The emulator is running"
+            $result="Running"
+            return
+          }
+          3 {
+            Write-Host "The emulator is stopped"
+          }
+          default {
+            Write-Host "Unrecognized exit code $process.ExitCode"
+          }
+        }
+
+        Start-Sleep -Seconds 5
+      }
+      until ($complete.Invoke())
+
+      Write-Error "The emulator failed to reach Running status within ${Timeout} seconds"
     displayName: Start Cosmos DB Emulator

eng/pipelines/templates/stages/cosmos-sdk-client.yml

@@ -133,7 +133,7 @@ stages:
         PreRunSteps:
         - template: /eng/common/pipelines/templates/steps/cosmos-emulator.yml
           parameters:
-            StartParameters: '-EnableAadAuthentication -PartitionCount 50 -Consistency Strong -Timeout 600 -EnablePreview'
+            StartParameters: '/noexplorer /noui /enablepreview /disableratelimiting /enableaadauthentication /partitioncount=50 /consistency=Strong'
         - powershell: |
             $Key = 'C2y6yDjf5/R+ob0N8A7Cgv30VRDJIWEHLM+4QDU5DE2nQ9nDuVTqobD4b8mGGyPMbIZnqyMsEcaGQy67XIw/Jw=='
             $password = ConvertTo-SecureString -String $Key -Force -AsPlainText

sdk/cosmos/azure-cosmos-encryption/src/main/java/com/azure/cosmos/encryption/EncryptionCosmosAsyncContainer.java

@@ -11,12 +11,14 @@
 import com.azure.cosmos.implementation.CosmosPagedFluxOptions;
 import com.azure.cosmos.implementation.ItemDeserializer;
 import com.azure.cosmos.implementation.Utils;
+import com.azure.cosmos.implementation.apachecommons.lang.tuple.Pair;
 import com.azure.cosmos.implementation.encryption.CosmosResponseFactory;
 import com.azure.cosmos.implementation.encryption.CosmosResponseFactoryCore;
 import com.azure.cosmos.implementation.encryption.EncryptionProcessor;
 import com.azure.cosmos.implementation.encryption.EncryptionUtils;
 import com.azure.cosmos.implementation.guava25.base.Preconditions;
 import com.azure.cosmos.implementation.query.Transformer;
+import com.azure.cosmos.models.CosmosChangeFeedRequestOptions;
 import com.azure.cosmos.models.CosmosItemRequestOptions;
 import com.azure.cosmos.models.CosmosItemResponse;
 import com.azure.cosmos.models.CosmosQueryRequestOptions;
@@ -38,6 +40,7 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
+import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull;
 import static com.azure.cosmos.models.EncryptionModelBridgeInternal.createEncryptionItemResponse;
 
 
@@ -89,7 +92,7 @@ private Mono<CosmosItemResponse<byte[]>> createItemHelper(
                 if (decryptResponse) {
                     return setByteArrayContent(rsp,
                         EncryptionProcessor.decrypt(EncryptionModelBridgeInternal.getByteArrayContent(rsp),
-                            this.encryptor).map(pair -> pair.getLeft()).publishOn(encryptionScheduler));
+                            this.encryptor).map(Pair::getLeft).publishOn(encryptionScheduler));
                 }
 
                 return Mono.just(rsp);
@@ -131,7 +134,7 @@ private Mono<CosmosItemResponse<byte[]>> upsertItemHelper(
                 if (decryptResponse) {
                     return setByteArrayContent(rsp,
                         EncryptionProcessor.decrypt(EncryptionModelBridgeInternal.getByteArrayContent(rsp),
-                            this.encryptor).map(pair -> pair.getLeft()).publishOn(encryptionScheduler));
+                            this.encryptor).map(Pair::getLeft).publishOn(encryptionScheduler));
                 }
 
                 return Mono.just(rsp);
@@ -176,7 +179,7 @@ private Mono<CosmosItemResponse<byte[]>> replaceItemHelper(
                 if (decryptResponse) {
                     return setByteArrayContent(rsp,
                         EncryptionProcessor.decrypt(EncryptionModelBridgeInternal.getByteArrayContent(rsp),
-                            this.encryptor).map(pair -> pair.getLeft()).publishOn(encryptionScheduler));
+                            this.encryptor).map(Pair::getLeft).publishOn(encryptionScheduler));
                 }
 
                 return Mono.just(rsp);
@@ -529,7 +532,36 @@ public <T> CosmosPagedFlux<T> queryItems(SqlQuerySpec query, CosmosQueryRequestO
             new Transformer<T>() {
                 @Override
                 public Function<CosmosPagedFluxOptions, Flux<FeedResponse<T>>> transform(Function<CosmosPagedFluxOptions, Flux<FeedResponse<JsonNode>>> func) {
-                    return queryDecryptionTransformer(classType, func);
+                    return queryDecryptionTransformer(classType, func, false);
+                }
+            });
+    }
+
+    /**
+     * Query for items in the change feed of the current container using the {@link CosmosChangeFeedRequestOptions}.
+     * <p>
+     * After subscription the operation will be performed. The {@link Flux} will
+     * contain one or several feed response of the obtained items. In case of
+     * failure the {@link CosmosPagedFlux} will error.
+     *
+     * @param <T> the type parameter.
+     * @param options the change feed request options.
+     * @param classType the class type.
+     * @return a {@link CosmosPagedFlux} containing one or several feed response pages of the obtained
+     * items or an error.
+     */
+    public <T> CosmosPagedFlux<T> queryChangeFeed(
+        CosmosChangeFeedRequestOptions options,
+        Class<T> classType) {
+
+        checkNotNull(options, "Argument 'options' must not be null.");
+        checkNotNull(classType, "Argument 'classType' must not be null.");
+
+        return CosmosBridgeInternal.queryChangeFeedInternal(container, options,
+            new Transformer<T>() {
+                @Override
+                public Function<CosmosPagedFluxOptions, Flux<FeedResponse<T>>> transform(Function<CosmosPagedFluxOptions, Flux<FeedResponse<JsonNode>>> func) {
+                    return queryDecryptionTransformer(classType, func, true);
                 }
             });
     }
@@ -565,8 +597,11 @@ private Mono<CosmosItemResponse<byte[]>> setByteArrayContent(CosmosItemResponse<
         );
     }
 
-    private <T> Function<CosmosPagedFluxOptions, Flux<FeedResponse<T>>> queryDecryptionTransformer(Class<T> classType,
-                                                                                                   Function<CosmosPagedFluxOptions, Flux<FeedResponse<JsonNode>>> func) {
+    private <T> Function<CosmosPagedFluxOptions, Flux<FeedResponse<T>>> queryDecryptionTransformer(
+        Class<T> classType,
+        Function<CosmosPagedFluxOptions, Flux<FeedResponse<JsonNode>>> func,
+        boolean useEtagAsContinuation) {
+
         return func.andThen(flux ->
             flux.publishOn(encryptionScheduler)
                 .flatMap(
@@ -589,7 +624,9 @@ private <T> Function<CosmosPagedFluxOptions, Flux<FeedResponse<T>>> queryDecrypt
                             return Mono.just(ModelBridgeInternal.createFeedResponseWithQueryMetrics(itemList,
                                 page.getResponseHeaders(),
                                 BridgeInternal.queryMetricsFromFeedResponse(page),
-                                ModelBridgeInternal.getQueryPlanDiagnosticsContext(page)));
+                                ModelBridgeInternal.getQueryPlanDiagnosticsContext(page),
+                                useEtagAsContinuation,
+                                ModelBridgeInternal.noChanges(page)));
                         } else {
                             List<Mono<byte[]>> byteArrayMonoList =
                                 byteArrayList.stream().map(bytes -> decryptResponse(bytes)).collect(Collectors.toList());
@@ -601,7 +638,9 @@ private <T> Function<CosmosPagedFluxOptions, Flux<FeedResponse<T>>> queryDecrypt
                                 ModelBridgeInternal.createFeedResponseWithQueryMetrics(itemList,
                                     page.getResponseHeaders(),
                                     BridgeInternal.queryMetricsFromFeedResponse(page),
-                                    ModelBridgeInternal.getQueryPlanDiagnosticsContext(page))
+                                    ModelBridgeInternal.getQueryPlanDiagnosticsContext(page),
+                                    useEtagAsContinuation,
+                                    ModelBridgeInternal.noChanges(page))
                             );
                         }
                     }

sdk/cosmos/azure-cosmos-encryption/src/main/java/com/azure/cosmos/models/EncryptionFeedResponse.java

@@ -10,7 +10,13 @@
 import java.util.concurrent.ConcurrentMap;
 
 class EncryptionFeedResponse<T> extends FeedResponse<T> {
-    EncryptionFeedResponse(List<T> results, Map<String, String> headers, ConcurrentMap<String, QueryMetrics> queryMetricsMap) {
-        super(results, headers, queryMetricsMap);
+    EncryptionFeedResponse(
+        List<T> results,
+        Map<String, String> headers,
+        ConcurrentMap<String, QueryMetrics> queryMetricsMap,
+        boolean useEtagAsContinuation,
+        boolean isNoChanges) {
+
+        super(results, headers, queryMetricsMap, useEtagAsContinuation, isNoChanges);
     }
 }

sdk/cosmos/azure-cosmos-encryption/src/test/java/com/azure/cosmos/implementation/encryption/EncryptionTests.java

@@ -35,9 +35,11 @@
 import com.azure.cosmos.encryption.EncryptionOptions;
 import com.azure.cosmos.implementation.directconnectivity.Protocol;
 import com.azure.cosmos.implementation.guava25.collect.ImmutableList;
+import com.azure.cosmos.models.CosmosChangeFeedRequestOptions;
 import com.azure.cosmos.models.CosmosItemRequestOptions;
 import com.azure.cosmos.models.CosmosItemResponse;
 import com.azure.cosmos.models.CosmosQueryRequestOptions;
+import com.azure.cosmos.models.FeedRange;
 import com.azure.cosmos.models.FeedResponse;
 import com.azure.cosmos.models.PartitionKey;
 import com.azure.cosmos.models.SqlParameter;
@@ -481,6 +483,35 @@ public void encryptionDecryptQueryResultMultipleDocs() {
         EncryptionTests.validateQueryResultsMultipleDocuments(EncryptionTests.encryptionContainer, testDoc1, testDoc2, query);
     }
 
+    @Test(groups = { "encryption" }, timeOut = TIMEOUT)
+    public void encryptionDecryptChangeFeedResultMultipleDocs() {
+        String partitionKey = UUID.randomUUID().toString();
+        TestDoc testDoc1 =  EncryptionTests
+            .createItem(
+                EncryptionTests.encryptionContainer,
+                EncryptionTests.dekId,
+                TestDoc.PathsToEncrypt,
+                partitionKey)
+            .getItem();
+        TestDoc testDoc2 =  EncryptionTests
+            .createItem(
+                EncryptionTests.encryptionContainer,
+                EncryptionTests.dekId,
+                TestDoc.PathsToEncrypt,
+                partitionKey)
+            .getItem();
+
+        CosmosChangeFeedRequestOptions options =
+            CosmosChangeFeedRequestOptions.createForProcessingFromBeginning(
+                FeedRange.forLogicalPartition(new PartitionKey(partitionKey)));
+
+        EncryptionTests.validateChangeFeedResultsMultipleDocuments(
+            EncryptionTests.encryptionContainer,
+            testDoc1,
+            testDoc2,
+            options);
+    }
+
     @Test(groups = { "encryption" }, timeOut = TIMEOUT)
     public void encryptionDecryptQueryResultMultipleEncryptedProperties() {
         List<String> pathsEncrypted = ImmutableList.of("/Sensitive", "/NonSensitive");
@@ -1047,6 +1078,29 @@ private static void validateQueryResultsMultipleDocuments(
         validateQueryResultsMultipleDocuments(container, testDoc1, testDoc2, query, null);
     }
 
+    private static void validateChangeFeedResultsMultipleDocuments(
+        EncryptionCosmosAsyncContainer container,
+        TestDoc testDoc1,
+        TestDoc testDoc2,
+        CosmosChangeFeedRequestOptions requestOptions) {
+        CosmosPagedFlux<TestDoc> pageFlux = container.queryChangeFeed(requestOptions, TestDoc.class);
+        List<TestDoc> readDocs = pageFlux.collectList().block();
+
+        assertThat(readDocs.size()).isEqualTo(2);
+        assertThat(readDocs).containsExactlyInAnyOrder(testDoc1, testDoc2);
+
+
+        CosmosPagedFlux<DecryptableItem> lazyDecryptablePageFlux = container.queryChangeFeed(requestOptions, DecryptableItem.class);
+        List<DecryptableItem> lazyDecryptableItems = lazyDecryptablePageFlux.collectList().block();
+
+        assertThat(readDocs.size()).isEqualTo(2);
+        assertThat(readDocs).containsExactlyInAnyOrder(testDoc1, testDoc2);
+
+
+        assertThat(lazyDecryptableItems.size()).isEqualTo(2);
+        assertThat(lazyDecryptableItems.stream().map(ldi -> ldi.getDecryptionResult(TestDoc.class).block().getDecryptedItem())).containsExactlyInAnyOrder(testDoc1, testDoc2);
+    }
+
     private static void validateQueryResultsMultipleDocuments(
         EncryptionCosmosAsyncContainer container,
         TestDoc testDoc1,
@@ -1060,15 +1114,15 @@ private static void validateQueryResultsMultipleDocuments(
         assertThat(readDocs).containsExactlyInAnyOrder(testDoc1, testDoc2);
 
 
-        CosmosPagedFlux<DecryptableItem> lazyDecraptablePageFlux = container.queryItems(new SqlQuerySpec(query), requestOptions, DecryptableItem.class);
-        List<DecryptableItem> lazyDecreptableItems = lazyDecraptablePageFlux.collectList().block();
+        CosmosPagedFlux<DecryptableItem> lazyDecryptablePageFlux = container.queryItems(new SqlQuerySpec(query), requestOptions, DecryptableItem.class);
+        List<DecryptableItem> lazyDecryptableItems = lazyDecryptablePageFlux.collectList().block();
 
         assertThat(readDocs.size()).isEqualTo(2);
         assertThat(readDocs).containsExactlyInAnyOrder(testDoc1, testDoc2);
 
 
-        assertThat(lazyDecreptableItems.size()).isEqualTo(2);
-        assertThat(lazyDecreptableItems.stream().map(ldi -> ldi.getDecryptionResult(TestDoc.class).block().getDecryptedItem())).containsExactlyInAnyOrder(testDoc1, testDoc2);
+        assertThat(lazyDecryptableItems.size()).isEqualTo(2);
+        assertThat(lazyDecryptableItems.stream().map(ldi -> ldi.getDecryptionResult(TestDoc.class).block().getDecryptedItem())).containsExactlyInAnyOrder(testDoc1, testDoc2);
     }
 
     private static <T> void validateQueryResponse(EncryptionCosmosAsyncContainer container,

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/BridgeInternal.java

@@ -147,11 +147,21 @@ public static <T> FeedResponse<T> createFeedResponse(List<T> results,
     }
 
     @Warning(value = INTERNAL_USE_ONLY_WARNING)
-    public static <T> FeedResponse<T> createFeedResponseWithQueryMetrics(List<T> results,
-                                                                         Map<String, String> headers,
-                                                                         ConcurrentMap<String, QueryMetrics> queryMetricsMap,
-                                                                         QueryInfo.QueryPlanDiagnosticsContext diagnosticsContext) {
-        return ModelBridgeInternal.createFeedResponseWithQueryMetrics(results, headers, queryMetricsMap, diagnosticsContext);
+    public static <T> FeedResponse<T> createFeedResponseWithQueryMetrics(
+        List<T> results,
+        Map<String, String> headers,
+        ConcurrentMap<String, QueryMetrics> queryMetricsMap,
+        QueryInfo.QueryPlanDiagnosticsContext diagnosticsContext,
+        boolean useEtagAsContinuation,
+        boolean isNoChangesResponse) {
+
+        return ModelBridgeInternal.createFeedResponseWithQueryMetrics(
+            results,
+            headers,
+            queryMetricsMap,
+            diagnosticsContext,
+            useEtagAsContinuation,
+            isNoChangesResponse);
     }
 
     @Warning(value = INTERNAL_USE_ONLY_WARNING)