code: private void triggerAsyncCaching(BlockInStream stream) throws IOException {
      boolean cache = ReadType.fromProto(mOptions.getOptions().getReadType()).isCache();
      boolean overReplicated = mStatus.getReplicationMax() > 0
          && mStatus.getFileBlockInfos().get((int) (getPos() / mBlockSize))
          .getBlockInfo().getLocations().size() >= mStatus.getReplicationMax();
      cache = cache && !overReplicated;
      // Get relevant information from the stream.
      WorkerNetAddress dataSource = stream.getAddress();
      long blockId = stream.getId();
      if (cache && (mLastBlockIdCached != blockId)) {
        WorkerNetAddress worker;
        if (mPassiveCachingEnabled && mContext.hasLocalWorker()) { // send request to local worker
          worker = mContext.getLocalWorker();
        } else { // send request to data source
          worker = dataSource;
        }
        try {
          // Construct the async cache request
          long blockLength = mOptions.getBlockInfo(blockId).getLength();
          AsyncCacheRequest request =
              AsyncCacheRequest.newBuilder().setBlockId(blockId).setLength(blockLength)
                  .setOpenUfsBlockOptions(mOptions.getOpenUfsBlockOptions(blockId))
                  .setSourceHost(dataSource.getHost()).setSourcePort(dataSource.getDataPort())
                  .build();
          BlockWorkerClient blockWorker = mContext.acquireBlockWorkerClient(worker);
          try {
            blockWorker.asyncCache(request);
            mLastBlockIdCached = blockId;
          } finally {
            mContext.releaseBlockWorkerClient(worker, blockWorker);
          }
        } catch (Exception e) {
          LOG.warn("Failed to complete async cache request for block {} at worker {}: {}", blockId,
              worker, e.getMessage());
        }
      }
    }
docs: Send an async cache request to a worker based on read type and passive cache options.