static <T> CompletableFuture<T> timelineConsistentRead(AsyncRegionLocator locator,
    TableName tableName, Query query, byte[] row, RegionLocateType locateType,
    Function<Integer, CompletableFuture<T>> requestReplica, long rpcTimeoutNs,
    long primaryCallTimeoutNs, Timer retryTimer, Optional<MetricsConnection> metrics) {
    if (query.getConsistency() != Consistency.TIMELINE) {
      return requestReplica.apply(RegionReplicaUtil.DEFAULT_REPLICA_ID);
    }
    // user specifies a replica id explicitly, just send request to the specific replica
    if (query.getReplicaId() >= 0) {
      return requestReplica.apply(query.getReplicaId());
    }
    // Timeline consistent read, where we may send requests to other region replicas
    CompletableFuture<T> primaryFuture = requestReplica.apply(RegionReplicaUtil.DEFAULT_REPLICA_ID);
    CompletableFuture<T> future = new CompletableFuture<>();
    connect(primaryFuture, future, Optional.empty());
    long startNs = System.nanoTime();
    // after the getRegionLocations, all the locations for the replicas of this region should have
    // been cached, so it is not big deal to locate them again when actually sending requests to
    // these replicas.
    addListener(locator.getRegionLocations(tableName, row, locateType, false, rpcTimeoutNs),
      (locs, error) -> {
        if (error != null) {
          LOG.warn(
            "Failed to locate all the replicas for table={}, row='{}', locateType={}"
              + " give up timeline consistent read",
            tableName, Bytes.toStringBinary(row), locateType, error);
          return;
        }
        if (locs.size() <= 1) {
          LOG.warn(
            "There are no secondary replicas for region {}, give up timeline consistent read",
            locs.getDefaultRegionLocation().getRegion());
          return;
        }
        long delayNs = primaryCallTimeoutNs - (System.nanoTime() - startNs);
        if (delayNs <= 0) {
          sendRequestsToSecondaryReplicas(requestReplica, locs, future, metrics);
        } else {
          retryTimer.newTimeout(
            timeout -> sendRequestsToSecondaryReplicas(requestReplica, locs, future, metrics),
            delayNs, TimeUnit.NANOSECONDS);
        }
      });
    return future;
  }
Connect the two futures, if the src future is done, then mark the dst future as done. And if
the dst future is done, then cancel the src future. This is used for timeline consistent read.
<p/>
Pass empty metrics if you want to link the primary future and the dst future so we will not
increase the hedge read related metrics.
hbase-client/src/main/java/org/apache/hadoop/hbase/client/ConnectionUtils.java