HBASE-8547. Fix java.lang.RuntimeException: Cached an already cached block (Addendum2 to check cached buffers for equality)

git-svn-id: https://svn.apache.org/repos/asf/hbase/trunk@1483476 13f79535-47bb-0310-9956-ffa450edef68

hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/LruBlockCache.java

@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.io.hfile;
 
 import java.io.IOException;
 import java.lang.ref.WeakReference;
+import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.EnumMap;
@@ -283,19 +284,26 @@ public class LruBlockCache implements BlockCache, HeapSize {
   /**
    * Cache the block with the specified name and buffer.
    * <p>
-   * It is assumed this will NEVER be called on an already cached block.  If
-   * that is done, an exception will be thrown.
+   * It is assumed this will NOT be called on an already cached block. In rare cases (HBASE-8547)
+   * this can happen, for which we compare the buffer contents.
    * @param cacheKey block's cache key
    * @param buf block buffer
    * @param inMemory if block is in-memory
    */
+  @Override
   public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory) {
     CachedBlock cb = map.get(cacheKey);
     if(cb != null) {
+      // compare the contents, if they are not equal, we are in big trouble
+      if (compare(buf, cb.getBuffer()) != 0) {
+        throw new RuntimeException("Cached block contents differ, which should not have happened."
+          + "cacheKey:" + cacheKey);
+      }
       String msg = "Cached an already cached block: " + cacheKey + " cb:" + cb.getCacheKey();
       msg += ". This is harmless and can happen in rare cases (see HBASE-8547)";
       LOG.warn(msg);
       assert false : msg;
+      return;
     }
     cb = new CachedBlock(cacheKey, buf, count.incrementAndGet(), inMemory);
     long newSize = updateSizeMetrics(cb, false);
@@ -306,6 +314,15 @@ public class LruBlockCache implements BlockCache, HeapSize {
     }
   }
 
+  private int compare(Cacheable left, Cacheable right) {
+    ByteBuffer l = ByteBuffer.allocate(left.getSerializedLength());
+    left.serialize(l);
+    ByteBuffer r = ByteBuffer.allocate(right.getSerializedLength());
+    right.serialize(r);
+    return Bytes.compareTo(l.array(), l.arrayOffset(), l.limit(),
+      r.array(), r.arrayOffset(), r.limit());
+  }
+
   /**
    * Cache the block with the specified name and buffer.
    * <p>