diff --git a/core/client/fs/src/test/java/alluxio/client/file/cache/store/LocalPageStoreTest.java b/core/client/fs/src/test/java/alluxio/client/file/cache/store/LocalPageStoreTest.java
index d07c492ca4a3..3adaeda7f3ad 100644
--- a/core/client/fs/src/test/java/alluxio/client/file/cache/store/LocalPageStoreTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/file/cache/store/LocalPageStoreTest.java
@@ -15,10 +15,12 @@
 import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
 import static org.junit.Assert.assertTrue;
 
 import alluxio.client.file.cache.PageId;
 import alluxio.client.file.cache.PageStore;
+import alluxio.exception.PageCorruptedException;
 import alluxio.file.ByteArrayTargetBuffer;
 
 import org.junit.Before;
@@ -171,12 +173,26 @@ public void cleanFileAndDirectory() throws Exception {
     assertFalse(Files.exists(p.getParent()));
   }
 
+  @Test
+  public void testCorruptedPages() throws Exception {
+    mOptions.setFileBuckets(1);
+    LocalPageStore pageStore = new LocalPageStore(mOptions);
+    byte[] buf = new byte[1000];
+    PageId id = new PageId("1", 0);
+    pageStore.put(id, "corrupted".getBytes());
+    assertThrows(PageCorruptedException.class, () -> {
+      //the bytes caller want to read is larger than the page file, mostly means the page corrupted
+      pageStore.get(id, 0, 100, new ByteArrayTargetBuffer(buf, 0));
+    });
+  }
+
   private void helloWorldTest(PageStore store) throws Exception {
     String msg = "Hello, World!";
     PageId id = new PageId("0", 0);
     store.put(id, msg.getBytes());
     byte[] buf = new byte[1024];
-    assertEquals(msg.getBytes().length, store.get(id, new ByteArrayTargetBuffer(buf, 0)));
+    assertEquals(msg.getBytes().length, store.get(id, 0, msg.length(),
+        new ByteArrayTargetBuffer(buf, 0)));
     assertArrayEquals(msg.getBytes(), Arrays.copyOfRange(buf, 0, msg.getBytes().length));
   }
 }
diff --git a/core/client/fs/src/test/java/alluxio/client/file/cache/store/MemoryPageStoreTest.java b/core/client/fs/src/test/java/alluxio/client/file/cache/store/MemoryPageStoreTest.java
index ed6f4b2aacb5..e20f11f50e5f 100644
--- a/core/client/fs/src/test/java/alluxio/client/file/cache/store/MemoryPageStoreTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/file/cache/store/MemoryPageStoreTest.java
@@ -43,7 +43,8 @@ private void helloWorldTest(PageStore store) throws Exception {
     PageId id = new PageId("0", 0);
     store.put(id, msg.getBytes());
     byte[] buf = new byte[PAGE_SIZE];
-    assertEquals(msg.getBytes().length, store.get(id, new ByteArrayTargetBuffer(buf, 0)));
+    assertEquals(msg.getBytes().length,
+        store.get(id, 0, msg.length(), new ByteArrayTargetBuffer(buf, 0)));
     assertArrayEquals(msg.getBytes(), Arrays.copyOfRange(buf, 0, msg.getBytes().length));
   }
 }
diff --git a/core/client/fs/src/test/java/alluxio/client/file/cache/store/PageStoreDirTest.java b/core/client/fs/src/test/java/alluxio/client/file/cache/store/PageStoreDirTest.java
index 3c6e768f5903..17e16c486c3b 100644
--- a/core/client/fs/src/test/java/alluxio/client/file/cache/store/PageStoreDirTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/file/cache/store/PageStoreDirTest.java
@@ -11,11 +11,14 @@
 
 package alluxio.client.file.cache.store;
 
+import static alluxio.client.file.cache.CacheUsage.PartitionDescriptor.file;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 
 import alluxio.ProjectConstants;
 import alluxio.client.file.cache.CacheManagerOptions;
+import alluxio.client.file.cache.CacheUsage;
+import alluxio.client.file.cache.CacheUsageView;
 import alluxio.client.file.cache.PageId;
 import alluxio.client.file.cache.PageInfo;
 import alluxio.conf.AlluxioConfiguration;
@@ -34,17 +37,19 @@
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashSet;
+import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 
 @RunWith(Parameterized.class)
 public class PageStoreDirTest {
+  public static final long CACHE_CAPACITY = 65536;
+  public static final long PAGE_SIZE = 1024;
   private final AlluxioConfiguration mConf = Configuration.global();
 
-  @Parameterized.Parameters
+  @Parameterized.Parameters(name = "{index}-{0}")
   public static Collection<Object[]> data() {
     return Arrays.asList(new Object[][] {
-        {PageStoreType.ROCKS},
         {PageStoreType.LOCAL},
         {PageStoreType.MEM}
     });
@@ -65,8 +70,8 @@ public void before() throws Exception {
     CacheManagerOptions cacheManagerOptions = CacheManagerOptions.create(mConf);
     mOptions = cacheManagerOptions.getPageStoreOptions().get(0);
     mOptions.setStoreType(mPageStoreType);
-    mOptions.setPageSize(1024);
-    mOptions.setCacheSize(65536);
+    mOptions.setPageSize(PAGE_SIZE);
+    mOptions.setCacheSize(CACHE_CAPACITY);
     mOptions.setAlluxioVersion(ProjectConstants.VERSION);
     mOptions.setRootDir(Paths.get(mTemp.getRoot().getAbsolutePath()));
 
@@ -120,4 +125,26 @@ public void getPagesUUID() throws Exception {
       assertEquals(pages, restored);
     }
   }
+
+  @Test
+  public void cacheUsage() throws Exception {
+    int len = 32;
+    int count = 16;
+    byte[] data = BufferUtils.getIncreasingByteArray(len);
+    for (int i = 0; i < count; i++) {
+      PageId id = new PageId("0", i);
+      mPageStoreDir.getPageStore().put(id, data);
+      mPageStoreDir.putPage(new PageInfo(id, data.length, mPageStoreDir));
+    }
+    Optional<CacheUsage> usage = mPageStoreDir.getUsage();
+    assertEquals(Optional.of(mPageStoreDir.getCapacityBytes()),
+        usage.map(CacheUsageView::capacity));
+    assertEquals(Optional.of((long) len * count), usage.map(CacheUsageView::used));
+    assertEquals(Optional.of(mPageStoreDir.getCapacityBytes() - (long) len * count),
+        usage.map(CacheUsageView::available));
+    // cache dir currently does not support get file level usage stat
+    Optional<CacheUsage> fileUsage = mPageStoreDir.getUsage()
+        .flatMap(usage1 -> usage1.partitionedBy(file("0")));
+    assertEquals(Optional.empty(), fileUsage);
+  }
 }
diff --git a/core/client/fs/src/test/java/alluxio/client/file/cache/store/PageStoreTest.java b/core/client/fs/src/test/java/alluxio/client/file/cache/store/PageStoreTest.java
index 40cfe243ce38..3c382d5b4f22 100644
--- a/core/client/fs/src/test/java/alluxio/client/file/cache/store/PageStoreTest.java
+++ b/core/client/fs/src/test/java/alluxio/client/file/cache/store/PageStoreTest.java
@@ -19,6 +19,7 @@
 import alluxio.ProjectConstants;
 import alluxio.client.file.cache.PageId;
 import alluxio.client.file.cache.PageStore;
+import alluxio.exception.PageCorruptedException;
 import alluxio.exception.PageNotFoundException;
 import alluxio.file.ByteArrayTargetBuffer;
 import alluxio.util.io.BufferUtils;
@@ -79,7 +80,8 @@ public void helloWorldTest() throws Exception {
     PageId id = new PageId("0", 0);
     mPageStore.put(id, msgBytes);
     byte[] buf = new byte[1024];
-    assertEquals(msgBytes.length, mPageStore.get(id, new ByteArrayTargetBuffer(buf, 0)));
+    assertEquals(msgBytes.length,
+        mPageStore.get(id, 0, msgBytes.length, new ByteArrayTargetBuffer(buf, 0)));
     assertArrayEquals(msgBytes, Arrays.copyOfRange(buf, 0, msgBytes.length));
     mPageStore.delete(id);
     try {
@@ -97,7 +99,8 @@ public void getOffset() throws Exception {
     mPageStore.put(id, BufferUtils.getIncreasingByteArray(len));
     byte[] buf = new byte[len];
     for (int offset = 1; offset < len; offset++) {
-      int bytesRead = mPageStore.get(id, offset, len, new ByteArrayTargetBuffer(buf, 0), false);
+      int bytesRead = mPageStore.get(id, offset, len - offset,
+          new ByteArrayTargetBuffer(buf, 0), false);
       assertEquals(len - offset, bytesRead);
       assertArrayEquals(BufferUtils.getIncreasingByteArray(offset, len - offset),
           Arrays.copyOfRange(buf, 0, bytesRead));
@@ -111,7 +114,7 @@ public void getOffsetOverflow() throws Exception {
     PageId id = new PageId("0", 0);
     mPageStore.put(id, BufferUtils.getIncreasingByteArray(len));
     byte[] buf = new byte[1024];
-    assertThrows(IllegalArgumentException.class, () ->
+    assertThrows(PageCorruptedException.class, () ->
         mPageStore.get(id, offset, len, new ByteArrayTargetBuffer(buf, 0)));
   }