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

Signed-off-by: Wellington Chevreuil <[email protected]>

Author: jhungund

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

@@ -390,17 +390,18 @@ 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.warn("Can't restore from file[{}]. The bucket cache will be reset and rebuilt."
+          + " Exception seen: ", 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 +952,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 +1498,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
@@ -1590,17 +1592,7 @@ private void verifyFileIntegrity(BucketCacheProtos.BucketCacheEntry proto) {
     }
   }
 
-  private void parseFirstChunk(BucketCacheProtos.BucketCacheEntry firstChunk) throws IOException {
-    fullyCachedFiles.clear();
-    Pair<ConcurrentHashMap<BlockCacheKey, BucketEntry>, NavigableSet<BlockCacheKey>> pair =
-      BucketProtoUtils.fromPB(firstChunk.getDeserializersMap(), firstChunk.getBackingMap(),
-        this::createRecycler);
-    backingMap.putAll(pair.getFirst());
-    blocksByHFile.addAll(pair.getSecond());
-    fullyCachedFiles.putAll(BucketProtoUtils.fromPB(firstChunk.getCachedFilesMap()));
-  }
-
-  private void parseChunkPB(BucketCacheProtos.BackingMap chunk,
+  private void updateCacheIndex(BucketCacheProtos.BackingMap chunk,
     java.util.Map<java.lang.Integer, java.lang.String> deserializer) throws IOException {
     Pair<ConcurrentHashMap<BlockCacheKey, BucketEntry>, NavigableSet<BlockCacheKey>> pair2 =
       BucketProtoUtils.fromPB(deserializer, chunk, this::createRecycler);
@@ -1626,55 +1618,42 @@ 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);
+        + "fullycachedFiles size: {}, chunkSize: {}",
+      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 cacheEntry =
       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);
-      parseChunkPB(BucketCacheProtos.BackingMap.parseDelimitedFrom(in),
-        firstChunk.getDeserializersMap());
-      LOG.info("Retrieved chunk: {}", i + 1);
-    }
-    verifyFileIntegrity(firstChunk);
-    verifyCapacityAndClasses(firstChunk.getCacheCapacity(), firstChunk.getIoClass(),
-      firstChunk.getMapClass());
+
+    fullyCachedFiles.clear();
+    fullyCachedFiles.putAll(BucketProtoUtils.fromPB(cacheEntry.getCachedFilesMap()));
+
+    backingMap.clear();
+    blocksByHFile.clear();
+
+    // Read the backing map entries in batches.
+    int numChunks = 0;
+    while (in.available() > 0) {
+      updateCacheIndex(BucketCacheProtos.BackingMap.parseDelimitedFrom(in),
+        cacheEntry.getDeserializersMap());
+      numChunks++;
+    }
+
+    LOG.info("Retrieved {} of chunks with blockCount = {}.", numChunks, backingMap.size());
+    verifyFileIntegrity(cacheEntry);
+    verifyCapacityAndClasses(cacheEntry.getCacheCapacity(), cacheEntry.getIoClass(),
+      cacheEntry.getMapClass());
     updateRegionSizeMapWhileRetrievingFromFile();
   }
 

Diff for: 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;
 
@@ -51,51 +50,53 @@ private BucketProtoUtils() {
   }
 
   static BucketCacheProtos.BucketCacheEntry toPB(BucketCache cache,
-    BucketCacheProtos.BackingMap backingMap) {
+    BucketCacheProtos.BackingMap.Builder backingMapBuilder) {
     return BucketCacheProtos.BucketCacheEntry.newBuilder().setCacheCapacity(cache.getMaxSize())
       .setIoClass(cache.ioEngine.getClass().getName())
       .setMapClass(cache.backingMap.getClass().getName())
       .putAllDeserializers(CacheableDeserializerIdManager.save())
-      .putAllCachedFiles(toCachedPB(cache.fullyCachedFiles)).setBackingMap(backingMap)
+      .putAllCachedFiles(toCachedPB(cache.fullyCachedFiles))
+      .setBackingMap(backingMapBuilder.build())
       .setChecksum(ByteString
         .copyFrom(((PersistentIOEngine) cache.ioEngine).calculateChecksum(cache.getAlgorithm())))
       .build();
   }
 
-  public static void serializeAsPB(BucketCache cache, FileOutputStream fos, long chunkSize,
-    long numChunks) throws IOException {
-    int blockCount = 0;
-    int chunkCount = 0;
-    int backingMapSize = cache.backingMap.size();
-    BucketCacheProtos.BackingMap.Builder builder = BucketCacheProtos.BackingMap.newBuilder();
-
+  public static void serializeAsPB(BucketCache cache, FileOutputStream fos, long chunkSize)
+    throws IOException {
+    // Write the new version of magic number.
     fos.write(PB_MAGIC_V2);
-    fos.write(Bytes.toBytes(chunkSize));
-    fos.write(Bytes.toBytes(numChunks));
 
+    BucketCacheProtos.BackingMap.Builder builder = BucketCacheProtos.BackingMap.newBuilder();
     BucketCacheProtos.BackingMapEntry.Builder entryBuilder =
       BucketCacheProtos.BackingMapEntry.newBuilder();
+
+    // Persist the metadata first.
+    toPB(cache, builder).writeDelimitedTo(fos);
+
+    int blockCount = 0;
+    // Persist backing map entries in chunks of size 'chunkSize'.
     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) {
-          // Persist all details along with the first chunk into BucketCacheEntry
-          BucketProtoUtils.toPB(cache, builder.build()).writeDelimitedTo(fos);
-        } else {
-          // Directly persist subsequent backing-map chunks.
-          builder.build().writeDelimitedTo(fos);
-        }
-        if (blockCount < backingMapSize) {
-          builder.clear();
-        }
+      addEntryToBuilder(entry, entryBuilder, builder);
+      if (blockCount % chunkSize == 0) {
+        builder.build().writeDelimitedTo(fos);
+        builder.clear();
       }
     }
+    // Persist the last chunk.
+    if (builder.getEntryList().size() > 0) {
+      builder.build().writeDelimitedTo(fos);
+    }
+  }
+
+  private static void addEntryToBuilder(Map.Entry<BlockCacheKey, BucketEntry> entry,
+    BucketCacheProtos.BackingMapEntry.Builder entryBuilder,
+    BucketCacheProtos.BackingMap.Builder builder) {
+    entryBuilder.clear();
+    entryBuilder.setKey(BucketProtoUtils.toPB(entry.getKey()));
+    entryBuilder.setValue(BucketProtoUtils.toPB(entry.getValue()));
+    builder.addEntry(entryBuilder.build());
   }
 
   private static BucketCacheProtos.BlockCacheKey toPB(BlockCacheKey key) {

Diff for: hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestVerifyBucketCacheFile.java

@@ -366,10 +366,17 @@ public void testSingleChunk() throws Exception {
   }
 
   @Test
-  public void testMultipleChunks() throws Exception {
+  public void testCompletelyFilledChunks() throws Exception {
+    // Test where the all the chunks are complete with chunkSize entries
     testChunkedBackingMapRecovery(5, 10);
   }
 
+  @Test
+  public void testPartiallyFilledChunks() throws Exception {
+    // Test where the last chunk is not completely filled.
+    testChunkedBackingMapRecovery(5, 13);
+  }
+
   private void testChunkedBackingMapRecovery(int chunkSize, int numBlocks) throws Exception {
     HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
     Path testDir = TEST_UTIL.getDataTestDir();