@Override
  public FindResult getNextEntriesAndUpdatedToken(FindToken curFindToken, long maxTotalSizeOfEntries,
      String partitionPath) throws CosmosException {
    Timer.Context operationTimer = azureMetrics.replicationFeedQueryTime.time();
    try {
      List<CloudBlobMetadata> nextEntries = new ArrayList<>();
      CosmosChangeFeedFindToken cosmosChangeFeedFindToken = (CosmosChangeFeedFindToken) curFindToken;
      int index = cosmosChangeFeedFindToken.getIndex();
      ChangeFeedCacheEntry changeFeedCacheEntry = changeFeedCache.get(cosmosChangeFeedFindToken.getCacheSessionId());
      boolean cacheHit = true;
      if (changeFeedCacheEntry == null || !isCacheValid(partitionPath, cosmosChangeFeedFindToken,
          changeFeedCacheEntry)) {
        // the cache may not be valid. So we cannot use session id
        azureMetrics.changeFeedCacheMissRate.mark();
        cacheHit = false;
        changeFeedCacheEntry = getNextChangeFeed(partitionPath, cosmosChangeFeedFindToken.getStartContinuationToken());
        // invalidate the previous token's cache
        changeFeedCache.remove(cosmosChangeFeedFindToken.getCacheSessionId());
        index = 0;
      }

      long resultSize = 0;

      List<CloudBlobMetadata> fetchedEntries = changeFeedCacheEntry.getFetchedEntries();
      while (true) {
        if (index < fetchedEntries.size()) {
          if (cacheHit) {
            azureMetrics.changeFeedCacheHitRate.mark();
            cacheHit = false;
          }
          if (resultSize + fetchedEntries.get(index).getSize() < maxTotalSizeOfEntries || resultSize == 0) {
            nextEntries.add(fetchedEntries.get(index));
            resultSize = resultSize + fetchedEntries.get(index).getSize();
            index++;
          } else {
            break;
          }
        } else {
          // we can reuse the session id in this case, because we know that the cache ran out of new items.
          changeFeedCacheEntry = getNextChangeFeed(partitionPath, changeFeedCacheEntry.getEndContinuationToken(),
              changeFeedCacheEntry.getCacheSessionId());
          fetchedEntries = changeFeedCacheEntry.getFetchedEntries();
          if (fetchedEntries.isEmpty()) {
            // This means that either there are no new changes, or change feed made progress in continuation token
            // but no change feed results were returned (probably because this cosmos shard's change feed has other
            // other partition feed too, and they were filtered out). In either case its appropriate to break here and
            // return updated token. The source replication logic will retry replication with updated token.
            break;
          } else {
            azureMetrics.changeFeedCacheRefreshRate.mark();
          }
          index = 0;
        }
      }

      FindToken updatedToken = new CosmosChangeFeedFindToken(cosmosChangeFeedFindToken.getBytesRead() + resultSize,
          changeFeedCacheEntry.getStartContinuationToken(), changeFeedCacheEntry.getEndContinuationToken(), index,
          changeFeedCacheEntry.getFetchedEntries().size(), changeFeedCacheEntry.getCacheSessionId(),
          cosmosChangeFeedFindToken.getVersion());
      changeFeedCache.put(changeFeedCacheEntry.getCacheSessionId(), new ChangeFeedCacheEntry(changeFeedCacheEntry));
      return new FindResult(nextEntries, updatedToken);
    } finally {
      operationTimer.stop();
    }
  }
Get next set of change feed entries for the specified partition, after the {@code curFindToken}.
The number of entries is capped by maxEntriesSize.
This method creates a cache for change feed entries. If the {@code curFindToken} is not valid,
or if all the items in the cache are consumed, then it queries Cosmos for new entries.
@param curFindToken {@link FindToken} after which the next entries have to be returned.
@param maxTotalSizeOfEntries maximum size of all the blobs returned.
@param partitionPath Partition for which change feed entries have to be returned.
@return {@link FindResult} instance that contains updated {@link FindToken} object which can act as a bookmark for
subsequent requests, and {@link List} of {@link CloudBlobMetadata} entries.
@throws CosmosException if any cosmos query encounters error.
ambry-cloud/src/main/java/com/github/ambry/cloud/azure/CosmosChangeFeedBasedReplicationFeed.java