@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 testPageDataFileCorrupted() throws Exception
  {
    int pages = 10;
    int fileSize = mPageSize * pages;
    byte[] testData = BufferUtils.getIncreasingByteArray(fileSize);
    ByteArrayCacheManager manager = new ByteArrayCacheManager();
    //by default local cache fallback is not enabled, the read should fail for any error
    LocalCacheFileInStream streamWithOutFallback = setupWithSingleFile(testData, manager);

    sConf.set(PropertyKey.USER_CLIENT_CACHE_FALLBACK_ENABLED, true);
    LocalCacheFileInStream streamWithFallback = setupWithSingleFile(testData, manager);
    Assert.assertEquals(100, streamWithFallback.positionedRead(0, new byte[10], 100, 100));
    Assert.assertEquals(1,
        MetricsSystem.counter(MetricKey.CLIENT_CACHE_POSITION_READ_FALLBACK.getName()).getCount());
  }