@Test
  public void testFindEntriesSinceUsingChangeFeed() throws Exception {
    long chunkSize = 110000;
    long maxTotalSize = 1000000; // between 9 and 10 chunks
    long startTime = System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1);
    int totalBlobs = 20;

    // create metadata list where total size > maxTotalSize
    List<String> blobIdList = new ArrayList<>();
    List<CloudBlobMetadata> cloudBlobMetadataList = new ArrayList<>();
    for (int j = 0; j < totalBlobs; j++) {
      BlobId blobId = generateBlobId();
      blobIdList.add(blobId.getID());
      CloudBlobMetadata inputMetadata = new CloudBlobMetadata(blobId, creationTime, Utils.Infinite_Time, chunkSize,
          CloudBlobMetadata.EncryptionOrigin.NONE);
      inputMetadata.setUploadTime(startTime + j);
      cloudBlobMetadataList.add(inputMetadata);
    }

    MockChangeFeedQuery mockChangeFeedQuery = new MockChangeFeedQuery();
    AzureReplicationFeed azureReplicationFeed = null;
    try {
      azureReplicationFeed =
          new CosmosChangeFeedBasedReplicationFeed(mockChangeFeedQuery, azureMetrics, azureDest.getQueryBatchSize());
      FieldSetter.setField(azureDest, azureDest.getClass().getDeclaredField("azureReplicationFeed"),
          azureReplicationFeed);
      cloudBlobMetadataList.stream().forEach(doc -> mockChangeFeedQuery.add(doc));
      CosmosChangeFeedFindToken findToken = new CosmosChangeFeedFindToken();
      // Run the query
      FindResult findResult = azureDest.findEntriesSince(blobId.getPartition().toPathString(), findToken, maxTotalSize);
      List<CloudBlobMetadata> firstResult = findResult.getMetadataList();
      findToken = (CosmosChangeFeedFindToken) findResult.getUpdatedFindToken();
      assertEquals("Did not get expected doc count", maxTotalSize / chunkSize, firstResult.size());

      assertEquals("Find token has wrong end continuation token", (findToken).getIndex(), firstResult.size());
      assertEquals("Find token has wrong totalItems count", (findToken).getTotalItems(),
          Math.min(blobIdList.size(), azureDest.getQueryBatchSize()));
      assertEquals("Unexpected change feed cache miss count", 1, azureMetrics.changeFeedCacheMissRate.getCount());
      assertEquals("Unexpected change feed cache refresh count", 0, azureMetrics.changeFeedCacheRefreshRate.getCount());
      assertEquals("Unexpected change feed cache hit count", 0, azureMetrics.changeFeedCacheHitRate.getCount());

      cloudBlobMetadataList = cloudBlobMetadataList.subList(firstResult.size(), cloudBlobMetadataList.size());

      findResult = azureDest.findEntriesSince(blobId.getPartition().toPathString(), findToken, maxTotalSize);
      List<CloudBlobMetadata> secondResult = findResult.getMetadataList();
      findToken = (CosmosChangeFeedFindToken) findResult.getUpdatedFindToken();

      assertEquals("Unexpected doc count", maxTotalSize / chunkSize, secondResult.size());
      assertEquals("Unexpected first blobId", blobIdList.get(firstResult.size()), secondResult.get(0).getId());

      assertEquals("Find token has wrong totalItems count", (findToken).getTotalItems(),
          Math.min(blobIdList.size(), azureDest.getQueryBatchSize()));

      assertEquals("Unexpected change feed cache miss count", 1, azureMetrics.changeFeedCacheMissRate.getCount());
      assertEquals("Unexpected change feed cache refresh count", 0, azureMetrics.changeFeedCacheRefreshRate.getCount());
      assertEquals("Unexpected change feed cache hit count", 1, azureMetrics.changeFeedCacheHitRate.getCount());

      // Rerun with max size below blob size, and make sure it returns one result
      findResult = azureDest.findEntriesSince(blobId.getPartition().toPathString(), findToken, chunkSize - 1);
      List<CloudBlobMetadata> thirdResult = findResult.getMetadataList();
      assertEquals("Expected one result", 1, thirdResult.size());
      findToken = (CosmosChangeFeedFindToken) findResult.getUpdatedFindToken();

      assertEquals("Unexpected change feed cache miss count", 1, azureMetrics.changeFeedCacheMissRate.getCount());
      assertEquals("Unexpected change feed cache refresh count", 0, azureMetrics.changeFeedCacheRefreshRate.getCount());
      assertEquals("Unexpected change feed cache hit count", 2, azureMetrics.changeFeedCacheHitRate.getCount());

      // Add more than AzureCloudConfig.cosmosQueryBatchSize blobs and test for correct change feed cache hits and misses.
      AzureCloudConfig azureConfig = new AzureCloudConfig(new VerifiableProperties(configProps));
      for (int j = 0; j < azureConfig.cosmosQueryBatchSize + 5; j++) {
        BlobId blobId = generateBlobId();
        blobIdList.add(blobId.getID());
        CloudBlobMetadata inputMetadata = new CloudBlobMetadata(blobId, creationTime, Utils.Infinite_Time, 10,
            CloudBlobMetadata.EncryptionOrigin.NONE);
        inputMetadata.setUploadTime(startTime + j);
        cloudBlobMetadataList.add(inputMetadata);
      }
      cloudBlobMetadataList.stream().forEach(doc -> mockChangeFeedQuery.add(doc));

      // Final correct query to drain out all the blobs and trigger a cache refresh.
      String prevEndToken = findToken.getEndContinuationToken();
      findResult = azureDest.findEntriesSince(blobId.getPartition().toPathString(), findToken, 1000000);
      findToken = (CosmosChangeFeedFindToken) findResult.getUpdatedFindToken();

      assertEquals("Unexpected change feed cache miss count", 1, azureMetrics.changeFeedCacheMissRate.getCount());
      assertEquals("Unexpected change feed cache refresh count", 1, azureMetrics.changeFeedCacheRefreshRate.getCount());
      assertEquals("Unexpected change feed cache hit count", 3, azureMetrics.changeFeedCacheHitRate.getCount());
      assertEquals("Since this would have triggered refresh, start token should have been previous token's end token",
          prevEndToken, findToken.getStartContinuationToken());

      // Query changefeed with invalid token and check for cache miss
      testFindEntriesSinceUsingChangeFeedWithInvalidToken(findToken);
    } finally {
      if (azureReplicationFeed != null) {
        azureReplicationFeed.close();
      }
    }
  }
Test findEntriesSince when cloud destination uses change feed based token.
ambry-cloud/src/test/java/com/github/ambry/cloud/azure/AzureCloudDestinationTest.java