From c5275f0cf9d18946d5bc52cd90cd4ba2deef2a1b Mon Sep 17 00:00:00 2001 From: Janardhan Hungund Date: Mon, 16 Sep 2024 10:30:03 +0530 Subject: [PATCH 1/3] 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 --- .../hbase/io/hfile/bucket/BucketCache.java | 52 ++++++------------- .../io/hfile/bucket/BucketProtoUtils.java | 21 ++++---- 2 files changed, 26 insertions(+), 47 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java index c236886f5f91..0ce637b5067b 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java +++ b/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,39 +1627,19 @@ 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 = @@ -1666,12 +1647,13 @@ private void retrieveChunkedBackingMap(FileInputStream in, int[] bucketSizes) th 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()); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java index 4acf724edd75..17e8cc9c0e35 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java +++ b/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 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); From d0744db8d9f5af53dc84a25d27bd2c9dfe1fe8e8 Mon Sep 17 00:00:00 2001 From: Janardhan Hungund Date: Fri, 20 Sep 2024 10:18:15 +0530 Subject: [PATCH 2/3] HBASE-28839: Address review comments Change-Id: Icf6cdcf829e7d4bd16f50f48fd02059b415f2d09 --- .../hbase/io/hfile/bucket/BucketCache.java | 5 ++- .../io/hfile/bucket/BucketProtoUtils.java | 44 +++++++++++++------ 2 files changed, 33 insertions(+), 16 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java index 0ce637b5067b..86f947ca2143 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java @@ -391,7 +391,8 @@ private void startPersistenceRetriever(int[] bucketSizes, long capacity) { retrieveFromFile(bucketSizes); LOG.info("Persistent bucket cache recovery from {} is complete.", persistencePath); } catch (Throwable ex) { - LOG.error("Can't restore from file[{}] because of ", persistencePath, 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); @@ -1629,7 +1630,7 @@ private void parsePB(BucketCacheProtos.BucketCacheEntry proto) throws IOExceptio private void persistChunkedBackingMap(FileOutputStream fos) throws IOException { LOG.debug( "persistToFile: before persisting backing map size: {}, " - + "fullycachedFiles size: {}, chunkSize: {}, numberofChunks: {}", + + "fullycachedFiles size: {}, chunkSize: {}", backingMap.size(), fullyCachedFiles.size(), persistenceChunkSize); BucketProtoUtils.serializeAsPB(this, fos, persistenceChunkSize); diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java index 17e8cc9c0e35..957b0694137b 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.util.Comparator; import java.util.HashMap; +import java.util.Iterator; import java.util.Map; import java.util.NavigableSet; import java.util.concurrent.ConcurrentHashMap; @@ -69,25 +70,31 @@ public static void serializeAsPB(BucketCache cache, FileOutputStream fos, long c int blockCount = 0; int backingMapSize = cache.backingMap.size(); BucketCacheProtos.BackingMap.Builder builder = BucketCacheProtos.BackingMap.newBuilder(); - boolean firstChunkPersisted = false; BucketCacheProtos.BackingMapEntry.Builder entryBuilder = BucketCacheProtos.BackingMapEntry.newBuilder(); + Iterator> entrySetIter = + cache.backingMap.entrySet().iterator(); - for (Map.Entry entry : cache.backingMap.entrySet()) { + // Create the first chunk and persist all details along with it. + while (entrySetIter.hasNext()) { blockCount++; - entryBuilder.clear(); - entryBuilder.setKey(BucketProtoUtils.toPB(entry.getKey())); - entryBuilder.setValue(BucketProtoUtils.toPB(entry.getValue())); - builder.addEntry(entryBuilder.build()); + Map.Entry entry = entrySetIter.next(); + addToBuilder(entry, entryBuilder, builder); if (blockCount % chunkSize == 0 || (blockCount == backingMapSize)) { - 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); - } + BucketProtoUtils.toPB(cache, builder.build()).writeDelimitedTo(fos); + break; + } + } + builder.clear(); + + // Persist the remaining chunks. + // These chunks only have the backing map entries. + while (entrySetIter.hasNext()) { + blockCount++; + Map.Entry entry = entrySetIter.next(); + addToBuilder(entry, entryBuilder, builder); + if (blockCount % chunkSize == 0 || (blockCount == backingMapSize)) { + builder.build().writeDelimitedTo(fos); if (blockCount < backingMapSize) { builder.clear(); } @@ -95,6 +102,15 @@ public static void serializeAsPB(BucketCache cache, FileOutputStream fos, long c } } + private static void addToBuilder(Map.Entry 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) { return BucketCacheProtos.BlockCacheKey.newBuilder().setHfilename(key.getHfileName()) .setOffset(key.getOffset()).setPrimaryReplicaBlock(key.isPrimary()) From 409554d3fa2a161780e0e209acacc8297ac8a24b Mon Sep 17 00:00:00 2001 From: Janardhan Hungund Date: Fri, 20 Sep 2024 19:20:10 +0530 Subject: [PATCH 3/3] HBASE-28839: Address further review comments Change-Id: I97936a683673ff89e04a15bc66542fb93a32fe8a --- .../hbase/io/hfile/bucket/BucketCache.java | 34 +++++++------- .../io/hfile/bucket/BucketProtoUtils.java | 44 +++++++------------ .../bucket/TestVerifyBucketCacheFile.java | 9 +++- 3 files changed, 39 insertions(+), 48 deletions(-) diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java index 86f947ca2143..eb612635dda2 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java @@ -1592,17 +1592,7 @@ private void verifyFileIntegrity(BucketCacheProtos.BucketCacheEntry proto) { } } - private void parseFirstChunk(BucketCacheProtos.BucketCacheEntry firstChunk) throws IOException { - fullyCachedFiles.clear(); - Pair, NavigableSet> 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 deserializer) throws IOException { Pair, NavigableSet> pair2 = BucketProtoUtils.fromPB(deserializer, chunk, this::createRecycler); @@ -1643,21 +1633,27 @@ private void persistChunkedBackingMap(FileOutputStream fos) throws IOException { 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. + 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) { - parseChunkPB(BucketCacheProtos.BackingMap.parseDelimitedFrom(in), - firstChunk.getDeserializersMap()); + updateCacheIndex(BucketCacheProtos.BackingMap.parseDelimitedFrom(in), + cacheEntry.getDeserializersMap()); numChunks++; } + LOG.info("Retrieved {} of chunks with blockCount = {}.", numChunks, backingMap.size()); - verifyFileIntegrity(firstChunk); - verifyCapacityAndClasses(firstChunk.getCacheCapacity(), firstChunk.getIoClass(), - firstChunk.getMapClass()); + verifyFileIntegrity(cacheEntry); + verifyCapacityAndClasses(cacheEntry.getCacheCapacity(), cacheEntry.getIoClass(), + cacheEntry.getMapClass()); updateRegionSizeMapWhileRetrievingFromFile(); } diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java index 957b0694137b..eb9c2cb5de88 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java @@ -21,7 +21,6 @@ import java.io.IOException; import java.util.Comparator; import java.util.HashMap; -import java.util.Iterator; import java.util.Map; import java.util.NavigableSet; import java.util.concurrent.ConcurrentHashMap; @@ -51,12 +50,13 @@ 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(); @@ -67,42 +67,30 @@ public static void serializeAsPB(BucketCache cache, FileOutputStream fos, long c // Write the new version of magic number. fos.write(PB_MAGIC_V2); - int blockCount = 0; - int backingMapSize = cache.backingMap.size(); BucketCacheProtos.BackingMap.Builder builder = BucketCacheProtos.BackingMap.newBuilder(); BucketCacheProtos.BackingMapEntry.Builder entryBuilder = BucketCacheProtos.BackingMapEntry.newBuilder(); - Iterator> entrySetIter = - cache.backingMap.entrySet().iterator(); - // Create the first chunk and persist all details along with it. - while (entrySetIter.hasNext()) { - blockCount++; - Map.Entry entry = entrySetIter.next(); - addToBuilder(entry, entryBuilder, builder); - if (blockCount % chunkSize == 0 || (blockCount == backingMapSize)) { - BucketProtoUtils.toPB(cache, builder.build()).writeDelimitedTo(fos); - break; - } - } - builder.clear(); + // Persist the metadata first. + toPB(cache, builder).writeDelimitedTo(fos); - // Persist the remaining chunks. - // These chunks only have the backing map entries. - while (entrySetIter.hasNext()) { + int blockCount = 0; + // Persist backing map entries in chunks of size 'chunkSize'. + for (Map.Entry entry : cache.backingMap.entrySet()) { blockCount++; - Map.Entry entry = entrySetIter.next(); - addToBuilder(entry, entryBuilder, builder); - if (blockCount % chunkSize == 0 || (blockCount == backingMapSize)) { + addEntryToBuilder(entry, entryBuilder, builder); + if (blockCount % chunkSize == 0) { builder.build().writeDelimitedTo(fos); - if (blockCount < backingMapSize) { - builder.clear(); - } + builder.clear(); } } + // Persist the last chunk. + if (builder.getEntryList().size() > 0) { + builder.build().writeDelimitedTo(fos); + } } - private static void addToBuilder(Map.Entry entry, + private static void addEntryToBuilder(Map.Entry entry, BucketCacheProtos.BackingMapEntry.Builder entryBuilder, BucketCacheProtos.BackingMap.Builder builder) { entryBuilder.clear(); diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestVerifyBucketCacheFile.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestVerifyBucketCacheFile.java index 21b71fa79b8d..4d4784167c61 100644 --- a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestVerifyBucketCacheFile.java +++ b/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();