@Override
  public int positionedRead(long pos, byte[] b, int off, int len) throws IOException {
    if (!CachePerThreadContext.get().getCacheEnabled()) {
      MetricsSystem.meter(MetricKey.CLIENT_CACHE_BYTES_REQUESTED_EXTERNAL.getName())
          .mark(len);
      MetricsSystem.counter(MetricKey.CLIENT_CACHE_EXTERNAL_REQUESTS.getName()).inc();
      len = getExternalFileInStream().positionedRead(pos, b, off, len);
      MultiDimensionalMetricsSystem.EXTERNAL_DATA_READ.inc(len);
      return len;
    }
    try {
      return readInternal(new ByteArrayTargetBuffer(b, off), off, len,
          ReadType.READ_INTO_BYTE_ARRAY, pos, true);
    } catch (IOException | RuntimeException e) {
      LOG.warn("Failed to read from Alluxio's page cache.", e);
      if (mFallbackEnabled) {
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_POSITION_READ_FALLBACK.getName()).inc();
        len = getExternalFileInStream().positionedRead(pos, b, off, len);
        MultiDimensionalMetricsSystem.EXTERNAL_DATA_READ.inc(len);
        return len;
      }
      throw e;
    }
  }
@Test
  public void positionReadOversizedBuffer() throws Exception {
    int fileSize = mPageSize;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    LocalCacheFileInStream stream = setupWithSingleFile(testData, manager);

    // cache miss
    byte[] cacheMiss = new byte[fileSize * 2];
    Assert.assertEquals(fileSize - 1,
        stream.positionedRead(1, cacheMiss, 2, fileSize * 2));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, 1, fileSize - 1),
        Arrays.copyOfRange(cacheMiss, 2, fileSize));
    Assert.assertEquals(0, manager.mPagesServed);
    Assert.assertEquals(1, manager.mPagesCached);

    // cache hit
    byte[] cacheHit = new byte[fileSize * 2];
    Assert.assertEquals(fileSize - 1,
        stream.positionedRead(1, cacheHit, 2, fileSize * 2));
    Assert.assertArrayEquals(
        Arrays.copyOfRange(testData, 1, fileSize - 1),
        Arrays.copyOfRange(cacheHit, 2, fileSize));
    Assert.assertEquals(1, manager.mPagesServed);
  }