@Override
  public FindResult getNextEntriesAndUpdatedToken(FindToken curfindToken, long maxTotalSizeOfEntries,
      String partitionPath) throws CosmosException {
    Timer.Context operationTimer = azureMetrics.replicationFeedQueryTime.time();
    try {
      CosmosUpdateTimeFindToken findToken = (CosmosUpdateTimeFindToken) curfindToken;

      SqlQuerySpec sqlQuerySpec =
          new SqlQuerySpec(ENTRIES_SINCE_QUERY_TEMPLATE, new SqlParameter(LIMIT_PARAM, queryBatchSize),
              new SqlParameter(TIME_SINCE_PARAM, findToken.getLastUpdateTime()));

      List<CloudBlobMetadata> queryResults =
          cosmosDataAccessor.queryMetadataAsync(partitionPath, sqlQuerySpec, azureMetrics.findSinceQueryTime).join();
      if (queryResults.isEmpty()) {
        return new FindResult(new ArrayList<>(), findToken);
      }
      if (queryResults.get(0).getLastUpdateTime() == findToken.getLastUpdateTime()) {
        filterOutLastReadBlobs(queryResults, findToken.getLastUpdateTimeReadBlobIds(), findToken.getLastUpdateTime());
      }
      List<CloudBlobMetadata> cappedResults =
          CloudBlobMetadata.capMetadataListBySize(queryResults, maxTotalSizeOfEntries);
      return new FindResult(cappedResults, CosmosUpdateTimeFindToken.getUpdatedToken(findToken, cappedResults));
    } catch (Exception ex) {
      ex = Utils.extractFutureExceptionCause(ex);
      if (ex instanceof CosmosException) {
        throw (CosmosException) ex;
      } else {
        throw new RuntimeException(ex);
      }
    } finally {
      operationTimer.stop();
    }
  }
Constructor for {@link CosmosUpdateTimeBasedReplicationFeed} object.
@param cosmosDataAccessor {@link CosmosDataAccessor} object to run Cosmos change feed queries.
@param azureMetrics {@link AzureMetrics} object.
@param queryBatchSize batch size for each find since query.
ambry-cloud/src/main/java/com/github/ambry/cloud/azure/CosmosUpdateTimeBasedReplicationFeed.java