[Bug] CFP Skipping Changes

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/CosmosContainerChangeFeedTest.java

@@ -323,6 +323,31 @@ public void asyncChangeFeed_fromBeginning_incremental_forLogicalPartition() thro
         }
     }
 
+    @Test(groups = { "emulator" }, timeOut = TIMEOUT)
+    public void asyncChangeFeedPrefetching() throws Exception {
+        this.createContainer(
+            (cp) -> cp.setChangeFeedPolicy(ChangeFeedPolicy.createLatestVersionPolicy())
+        );
+        insertDocuments(1, 20);
+
+        CosmosChangeFeedRequestOptions options = CosmosChangeFeedRequestOptions
+            .createForProcessingFromBeginning(FeedRange.forFullRange()).setMaxItemCount(10);
+        AtomicInteger count = new AtomicInteger(0);
+        // Will keep grabbing pages
+        createdContainer.asyncContainer.queryChangeFeed(options, ObjectNode.class).handle((r) ->
+            count.incrementAndGet()).byPage().subscribe();
+
+        Thread.sleep(3000);
+        assertThat(count.get()).isNotEqualTo(2);
+
+        count.set(0);
+        // should only get two pages
+        createdContainer.asyncContainer.queryChangeFeed(options, ObjectNode.class).handle((r) ->
+            count.incrementAndGet()).byPage().take(2, true).subscribe();
+        Thread.sleep(3000);
+        assertThat(count.get()).isEqualTo(2);
+    }
+
     @Test(groups = { "emulator" }, timeOut = TIMEOUT)
     public void asyncChangeFeed_fromBeginning_incremental_forEPK() throws Exception {
         this.createContainer(

sdk/cosmos/azure-cosmos-tests/src/test/java/com/azure/cosmos/rx/changefeed/pkversion/IncrementalChangeFeedProcessorTest.java

@@ -77,8 +77,6 @@ public class IncrementalChangeFeedProcessorTest extends TestSuiteBase {
     private static final ObjectMapper OBJECT_MAPPER = Utils.getSimpleObjectMapper();
 
     private CosmosAsyncDatabase createdDatabase;
-//    private final String databaseId = "testdb1";
-//    private final String hostName = "TestHost1";
     private final String hostName = RandomStringUtils.randomAlphabetic(6);
     private final int FEED_COUNT = 10;
     private final int CHANGE_FEED_PROCESSOR_TIMEOUT = 5000;

sdk/cosmos/azure-cosmos/CHANGELOG.md

@@ -7,6 +7,7 @@
 #### Breaking Changes
 
 #### Bugs Fixed
+* Fixes an issue in change feed processor where records are skipped and excessive requests are prefetched. - See [PR 43788](https://github.com/Azure/azure-sdk-for-java/pull/43788)
 
 #### Other Changes
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/CosmosChangeFeedRequestOptionsImpl.java

@@ -11,6 +11,7 @@
 import com.azure.cosmos.implementation.changefeed.common.ChangeFeedMode;
 import com.azure.cosmos.implementation.changefeed.common.ChangeFeedStartFromInternal;
 import com.azure.cosmos.implementation.changefeed.common.ChangeFeedState;
+import com.azure.cosmos.implementation.changefeed.common.ChangeFeedStateV1;
 import com.azure.cosmos.implementation.feedranges.FeedRangeInternal;
 import com.azure.cosmos.implementation.spark.OperationContextAndListenerTuple;
 import com.azure.cosmos.models.CosmosRequestOptions;
@@ -52,7 +53,11 @@ public final class CosmosChangeFeedRequestOptionsImpl implements OverridableRequ
     private Long endLSN;
 
     public CosmosChangeFeedRequestOptionsImpl(CosmosChangeFeedRequestOptionsImpl toBeCloned) {
-        this.continuationState = toBeCloned.continuationState;
+        if (toBeCloned.continuationState != null) {
+            this.continuationState = new ChangeFeedStateV1((ChangeFeedStateV1) toBeCloned.continuationState);
+        } else {
+            this.continuationState = null;
+        }
         this.feedRangeInternal = toBeCloned.feedRangeInternal;
         this.properties = toBeCloned.properties;
         this.maxItemCount = toBeCloned.maxItemCount;
@@ -93,7 +98,12 @@ public CosmosChangeFeedRequestOptionsImpl(
         this.maxPrefetchPageCount = DEFAULT_MAX_PREFETCH_PAGE_COUNT;
         this.feedRangeInternal = feedRange;
         this.startFromInternal = startFromInternal;
-        this.continuationState = continuationState;
+        if (continuationState != null) {
+            this.continuationState = new ChangeFeedStateV1((ChangeFeedStateV1) continuationState);
+        } else {
+            this.continuationState = null;
+        }
+
 
         if (mode != ChangeFeedMode.INCREMENTAL && mode != ChangeFeedMode.FULL_FIDELITY) {
             throw new IllegalArgumentException(

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/common/ChangeFeedContextClientImpl.java

@@ -7,12 +7,8 @@
 import com.azure.cosmos.CosmosAsyncDatabase;
 import com.azure.cosmos.CosmosBridgeInternal;
 import com.azure.cosmos.implementation.AsyncDocumentClient;
-import com.azure.cosmos.implementation.ChangeFeedOperationState;
-import com.azure.cosmos.implementation.Document;
 import com.azure.cosmos.implementation.ImplementationBridgeHelpers;
-import com.azure.cosmos.implementation.OperationType;
 import com.azure.cosmos.implementation.PartitionKeyRange;
-import com.azure.cosmos.implementation.ResourceType;
 import com.azure.cosmos.implementation.changefeed.ChangeFeedContextClient;
 import com.azure.cosmos.implementation.routing.Range;
 import com.azure.cosmos.models.CosmosBulkOperationResponse;
@@ -42,7 +38,6 @@
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.stream.Collectors;
 
 import static com.azure.cosmos.CosmosBridgeInternal.getContextClient;
 import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull;
@@ -151,7 +146,7 @@ public  <T> Flux<FeedResponse<T>> createDocumentChangeFeedQuery(CosmosAsyncConta
         }
         return collectionLink
             .queryChangeFeed(changeFeedRequestOptions, klass)
-            .byPage()
+            .byPage().take(1, true)
             .publishOn(this.scheduler);
     }
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/common/ChangeFeedStateV1.java

@@ -11,6 +11,8 @@
 import com.azure.cosmos.implementation.feedranges.FeedRangeInternal;
 import com.azure.cosmos.implementation.query.CompositeContinuationToken;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 
 import static com.azure.cosmos.implementation.guava25.base.Preconditions.checkNotNull;
@@ -38,6 +40,20 @@ public ChangeFeedStateV1(
         this.mode = mode;
     }
 
+    public ChangeFeedStateV1(ChangeFeedStateV1 toBeCloned) {
+        this.containerRid = toBeCloned.containerRid;
+        this.feedRange = toBeCloned.feedRange;
+        this.startFromSettings = toBeCloned.startFromSettings;
+        if (toBeCloned.continuation != null) {
+            List<CompositeContinuationToken> compositeContinuationTokens = new ArrayList<>();
+           compositeContinuationTokens.addAll(toBeCloned.continuation.getCompositeContinuationTokens());
+            this.continuation = FeedRangeContinuation.create(toBeCloned.continuation.getContainerRid(), toBeCloned.continuation.getFeedRange(), compositeContinuationTokens);
+        } else {
+            this.continuation = null;
+        }
+        this.mode = toBeCloned.mode;
+    }
+
     @Override
     public FeedRangeContinuation getContinuation() {
         return this.continuation;

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/epkversion/ChangeFeedProcessorImplBase.java

@@ -232,7 +232,6 @@ public Mono<List<ChangeFeedProcessorState>> getCurrentState() {
 
                             return this.feedContextClient
                                     .createDocumentChangeFeedQuery(this.feedContextClient.getContainerClient(), options, ChangeFeedProcessorItem.class, false)
-                                    .take(1)
                                     .map(feedResponse -> {
                                         ChangeFeedProcessorState changeFeedProcessorState = new ChangeFeedProcessorState()
                                                 .setHostName(lease.getOwner())

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/epkversion/PartitionProcessorImpl.java

@@ -119,8 +119,7 @@ public Mono<Void> run(CancellationToken cancellationToken) {
                 return this.documentClient.createDocumentChangeFeedQuery(
                         this.settings.getCollectionSelfLink(),
                         this.options,
-                        itemType)
-                    .limitRequest(1);
+                        itemType);
             })
             .flatMap(documentFeedResponse -> {
                 if (cancellationToken.isCancellationRequested()) return Flux.error(new TaskCancelledException());

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/changefeed/pkversion/PartitionProcessorImpl.java

@@ -121,7 +121,7 @@ public Mono<Void> run(CancellationToken cancellationToken) {
                 return this.documentClient.createDocumentChangeFeedQuery(
                     this.settings.getCollectionSelfLink(),
                     this.options,
-                    JsonNode.class).limitRequest(1);
+                    JsonNode.class);
             })
             .flatMap(documentFeedResponse -> {
                 if (cancellationToken.isCancellationRequested()) return Flux.error(new TaskCancelledException());