Fix cacheMissPercentage metric

core/common/src/main/java/alluxio/worker/block/io/LocalFileBlockReader.java

@@ -102,9 +102,7 @@ public ByteBuffer read(long offset, long length) throws IOException {
 
   @Override
   public int transferTo(ByteBuf buf) throws IOException {
-    int bytesReadFromCache = buf.writeBytes(mLocalFileChannel, buf.writableBytes());
-    MetricsSystem.counter(MetricKey.WORKER_BYTES_READ_CACHE.getName()).inc(bytesReadFromCache);
-    return bytesReadFromCache;
+    return buf.writeBytes(mLocalFileChannel, buf.writableBytes());
   }
 
   @Override

core/server/worker/src/main/java/alluxio/worker/block/TieredBlockStore.java

@@ -29,6 +29,7 @@
 import alluxio.worker.block.annotator.BlockOrder;
 import alluxio.worker.block.io.BlockReader;
 import alluxio.worker.block.io.BlockWriter;
+import alluxio.worker.block.io.DeStoreBlockReader;
 import alluxio.worker.block.io.DelegatingBlockReader;
 import alluxio.worker.block.io.StoreBlockReader;
 import alluxio.worker.block.io.StoreBlockWriter;
@@ -222,9 +223,10 @@ public BlockReader createBlockReader(long sessionId, long blockId, long offset)
 
     try {
       BlockReader reader = new StoreBlockReader(sessionId, block);
+      BlockReader dereader = new DeStoreBlockReader(reader);
       ((FileChannel) reader.getChannel()).position(offset);
       accessBlock(sessionId, blockId);
-      return new DelegatingBlockReader(reader, blockLock);
+      return new DelegatingBlockReader(dereader, blockLock);
     } catch (Exception e) {
       blockLock.close();
       throw new IOException(format("Failed to get local block reader, sessionId=%d, "

core/server/worker/src/main/java/alluxio/worker/block/io/DeStoreBlockReader.java

@@ -0,0 +1,62 @@
+package alluxio.worker.block.io;
+
+import alluxio.metrics.MetricKey;
+import alluxio.metrics.MetricsSystem;
+
+import io.netty.buffer.ByteBuf;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.ReadableByteChannel;
+
+/**
+ * An reader class with metrics.
+ */
+public class DeStoreBlockReader extends BlockReader {
+  private final BlockReader mDeBlockReader;
+
+  /**
+   * A decorator of BlockReader.
+   * @param deblockReader block reader
+   */
+  public DeStoreBlockReader(BlockReader deblockReader) {
+    mDeBlockReader = deblockReader;
+  }
+
+  @Override
+  public ByteBuffer read(long offset, long length) throws IOException {
+    return mDeBlockReader.read(offset, length);
+  }
+
+  @Override
+  public long getLength() {
+    return mDeBlockReader.getLength();
+  }
+
+  @Override
+  public ReadableByteChannel getChannel() {
+    return mDeBlockReader.getChannel();
+  }
+
+  @Override
+  public int transferTo(ByteBuf buf) throws IOException {
+    int bytesReadFromCache = mDeBlockReader.transferTo(buf);
+    MetricsSystem.counter(MetricKey.WORKER_BYTES_READ_CACHE.getName()).inc(bytesReadFromCache);
+    return bytesReadFromCache;
+  }
+
+  @Override
+  public boolean isClosed() {
+    return mDeBlockReader.isClosed();
+  }
+
+  @Override
+  public String getLocation() {
+    return mDeBlockReader.getLocation();
+  }
+
+  @Override
+  public String toString() {
+    return mDeBlockReader.toString();
+  }
+}