HBASE-28839: Handle all types of exceptions during retrieval of bucket-cache from persistence.

In certain scenarios, where there is discrepancy between the number of chunks persisted
in the file and the number of chunks stored in the persistence file. This is because the bucket
cache may be operated upon in parallel.

During the retrievel of bucket-cache from persistence, it was observed that, if an exception,
other than the IOException occurs, the exception is not logged and also the retrieval thread
exits leaving the bucket cache in an uninitialised state, leaving it unusable.

With this change, the retrieval code does not rely on the metadata information (number of chunks)
and instead, it reads from the file-stream as long as the data is available to be read.

This change, enables the retrieval thread to print all types of exceptions and also reinitialises
the bucket cache and makes it reusable.

Change-Id: I81b7f5fe06945702bbc59df96d054f95f03de499

Author: jhungund

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

@@ -390,17 +390,17 @@ private void startPersistenceRetriever(int[] bucketSizes, long capacity) {
       try {
         retrieveFromFile(bucketSizes);
         LOG.info("Persistent bucket cache recovery from {} is complete.", persistencePath);
-      } catch (IOException ioex) {
-        LOG.error("Can't restore from file[{}] because of ", persistencePath, ioex);
+      } catch (Throwable ex) {
+        LOG.error("Can't restore from file[{}] because of ", persistencePath, ex);
         backingMap.clear();
         fullyCachedFiles.clear();
         backingMapValidated.set(true);
+        regionCachedSize.clear();
         try {
           bucketAllocator = new BucketAllocator(capacity, bucketSizes);
-        } catch (BucketAllocatorException ex) {
-          LOG.error("Exception during Bucket Allocation", ex);
+        } catch (BucketAllocatorException allocatorException) {
+          LOG.error("Exception during Bucket Allocation", allocatorException);
         }
-        regionCachedSize.clear();
       } finally {
         this.cacheState = CacheState.ENABLED;
         startWriterThreads();
@@ -951,7 +951,8 @@ public void logStats() {
         : (StringUtils.formatPercent(cacheStats.getHitCachingRatio(), 2) + ", "))
       + "evictions=" + cacheStats.getEvictionCount() + ", " + "evicted="
       + cacheStats.getEvictedCount() + ", " + "evictedPerRun=" + cacheStats.evictedPerEviction()
-      + ", " + "allocationFailCount=" + cacheStats.getAllocationFailCount());
+      + ", " + "allocationFailCount=" + cacheStats.getAllocationFailCount() + ", blocksCount="
+      + backingMap.size());
     cacheStats.reset();
 
     bucketAllocator.logDebugStatistics();
@@ -1496,7 +1497,7 @@ private void retrieveFromFile(int[] bucketSizes) throws IOException {
       } else if (Arrays.equals(pbuf, BucketProtoUtils.PB_MAGIC_V2)) {
         // The new persistence format of chunked persistence.
         LOG.info("Reading new chunked format of persistence.");
-        retrieveChunkedBackingMap(in, bucketSizes);
+        retrieveChunkedBackingMap(in);
       } else {
         // In 3.0 we have enough flexibility to dump the old cache data.
         // TODO: In 2.x line, this might need to be filled in to support reading the old format
@@ -1626,52 +1627,33 @@ private void parsePB(BucketCacheProtos.BucketCacheEntry proto) throws IOExceptio
   }
 
   private void persistChunkedBackingMap(FileOutputStream fos) throws IOException {
-    long numChunks = backingMap.size() / persistenceChunkSize;
-    if (backingMap.size() % persistenceChunkSize != 0) {
-      numChunks += 1;
-    }
-
     LOG.debug(
       "persistToFile: before persisting backing map size: {}, "
         + "fullycachedFiles size: {}, chunkSize: {}, numberofChunks: {}",
-      backingMap.size(), fullyCachedFiles.size(), persistenceChunkSize, numChunks);
+      backingMap.size(), fullyCachedFiles.size(), persistenceChunkSize);
 
-    BucketProtoUtils.serializeAsPB(this, fos, persistenceChunkSize, numChunks);
+    BucketProtoUtils.serializeAsPB(this, fos, persistenceChunkSize);
 
     LOG.debug(
-      "persistToFile: after persisting backing map size: {}, "
-        + "fullycachedFiles size: {}, numChunksPersisteed: {}",
-      backingMap.size(), fullyCachedFiles.size(), numChunks);
+      "persistToFile: after persisting backing map size: {}, " + "fullycachedFiles size: {}",
+      backingMap.size(), fullyCachedFiles.size());
   }
 
-  private void retrieveChunkedBackingMap(FileInputStream in, int[] bucketSizes) throws IOException {
-    byte[] bytes = new byte[Long.BYTES];
-    int readSize = in.read(bytes);
-    if (readSize != Long.BYTES) {
-      throw new IOException("Invalid size of chunk-size read from persistence: " + readSize);
-    }
-    long batchSize = Bytes.toLong(bytes, 0);
-
-    readSize = in.read(bytes);
-    if (readSize != Long.BYTES) {
-      throw new IOException("Invalid size for number of chunks read from persistence: " + readSize);
-    }
-    long numChunks = Bytes.toLong(bytes, 0);
-
-    LOG.info("Number of chunks: {}, chunk size: {}", numChunks, batchSize);
+  private void retrieveChunkedBackingMap(FileInputStream in) throws IOException {
 
     // Read the first chunk that has all the details.
     BucketCacheProtos.BucketCacheEntry firstChunk =
       BucketCacheProtos.BucketCacheEntry.parseDelimitedFrom(in);
     parseFirstChunk(firstChunk);
 
     // Subsequent chunks have the backingMap entries.
-    for (int i = 1; i < numChunks; i++) {
-      LOG.info("Reading chunk no: {}", i + 1);
+    int numChunks = 0;
+    while (in.available() > 0) {
       parseChunkPB(BucketCacheProtos.BackingMap.parseDelimitedFrom(in),
         firstChunk.getDeserializersMap());
-      LOG.info("Retrieved chunk: {}", i + 1);
+      numChunks++;
     }
+    LOG.info("Retrieved {} of chunks with blockCount = {}.", numChunks, backingMap.size());
     verifyFileIntegrity(firstChunk);
     verifyCapacityAndClasses(firstChunk.getCacheCapacity(), firstChunk.getIoClass(),
       firstChunk.getMapClass());

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

@@ -33,7 +33,6 @@
 import org.apache.hadoop.hbase.io.hfile.BlockType;
 import org.apache.hadoop.hbase.io.hfile.CacheableDeserializerIdManager;
 import org.apache.hadoop.hbase.io.hfile.HFileBlock;
-import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -62,31 +61,29 @@ static BucketCacheProtos.BucketCacheEntry toPB(BucketCache cache,
       .build();
   }
 
-  public static void serializeAsPB(BucketCache cache, FileOutputStream fos, long chunkSize,
-    long numChunks) throws IOException {
+  public static void serializeAsPB(BucketCache cache, FileOutputStream fos, long chunkSize)
+    throws IOException {
+    // Write the new version of magic number.
+    fos.write(PB_MAGIC_V2);
+
     int blockCount = 0;
-    int chunkCount = 0;
     int backingMapSize = cache.backingMap.size();
     BucketCacheProtos.BackingMap.Builder builder = BucketCacheProtos.BackingMap.newBuilder();
-
-    fos.write(PB_MAGIC_V2);
-    fos.write(Bytes.toBytes(chunkSize));
-    fos.write(Bytes.toBytes(numChunks));
-
+    boolean firstChunkPersisted = false;
     BucketCacheProtos.BackingMapEntry.Builder entryBuilder =
       BucketCacheProtos.BackingMapEntry.newBuilder();
+
     for (Map.Entry<BlockCacheKey, BucketEntry> entry : cache.backingMap.entrySet()) {
       blockCount++;
       entryBuilder.clear();
       entryBuilder.setKey(BucketProtoUtils.toPB(entry.getKey()));
       entryBuilder.setValue(BucketProtoUtils.toPB(entry.getValue()));
       builder.addEntry(entryBuilder.build());
-
       if (blockCount % chunkSize == 0 || (blockCount == backingMapSize)) {
-        chunkCount++;
-        if (chunkCount == 1) {
+        if (firstChunkPersisted == false) {
           // Persist all details along with the first chunk into BucketCacheEntry
           BucketProtoUtils.toPB(cache, builder.build()).writeDelimitedTo(fos);
+          firstChunkPersisted = true;
         } else {
           // Directly persist subsequent backing-map chunks.
           builder.build().writeDelimitedTo(fos);