public CompletableFuture<Integer> read(ByteBuffer buf, long offset, long len, FileId fileId,
      String ufsPath, UfsReadOptions options) {
    Objects.requireNonNull(buf);
    if (offset < 0 || len < 0 || len > buf.remaining()) {
      throw new OutOfRangeRuntimeException(String.format(
          "offset is negative, len is negative, or len is greater than buf remaining. "
              + "offset: %s, len: %s, buf remaining: %s", offset, len, buf.remaining()));
    }
    if (mReadQueue.size() >= READ_CAPACITY) {
      throw new ResourceExhaustedRuntimeException("UFS read at capacity", true);
    }
    CompletableFuture<Integer> future = new CompletableFuture<>();
    if (len == 0) {
      future.complete(0);
      return future;
    }
    Meter meter = mUfsBytesReadThroughputMetrics.computeIfAbsent(mUfsClient.getUfsMountPointUri(),
        uri -> MetricsSystem.meterWithTags(MetricKey.WORKER_BYTES_READ_UFS_THROUGHPUT.getName(),
            MetricKey.WORKER_BYTES_READ_UFS_THROUGHPUT.isClusterAggregated(), MetricInfo.TAG_UFS,
            MetricsSystem.escape(mUfsClient.getUfsMountPointUri()), MetricInfo.TAG_USER,
            options.getTag()));

    mReadQueue.add(new ReadTask(buf, ufsPath, fileId, offset,
        len, options, future, meter));
    return future;
  }
@Test
  public void readFullBlock() throws Exception {
    mUfsIOManager.read(TEST_BUF, 0, TEST_BLOCK_SIZE, FIRST_BLOCK_ID, mTestFilePath,
        UfsReadOptions.getDefaultInstance()).get();
    assertTrue(checkBuf(0, (int) TEST_BLOCK_SIZE, TEST_BUF));
    TEST_BUF.clear();
  }