Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wchevreuil merged 3 commits into from
Sep 23, 2024
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
c5275f0
HBASE-28839: Handle all types of exceptions during retrieval of bucke…
jhungund Sep 16, 2024
d0744db
HBASE-28839: Address review comments
jhungund Sep 20, 2024
409554d
HBASE-28839: Address further review comments
jhungund Sep 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 36 additions & 57 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down Expand Up @@ -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();
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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);
Expand All @@ -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();
}

Expand Down
59 changes: 30 additions & 29 deletions hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketProtoUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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())
Copy link
Contributor Author

@jhungund jhungund Sep 23, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need to pass an empty backingmap here otherwise we see an exception of an incomplete object. Hence, we just pass an empty backing map along with the metadata. Subsequently, we persist all entries of the backing map in chunks.

Copy link
Contributor

@wchevreuil wchevreuil Sep 23, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We don't need a builder, just pass an empty map. Or, since we don't persist any map entries within the BucketCacheEntry proto object, just remove the map from the protobuf message. We already changed the persistent file format on HBASE-28805, as long as this can land on all related branches whilst HBASE-28805 has not made into any release, we are free to change the format.

Copy link
Contributor Author

@jhungund jhungund Sep 23, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @wchevreuil, we will need to retain the old format of protobuf message to maintain the backward compatibility. Hence, the we cannot change the protobuf message. We can reuse this protobuf message by persisting an empty backing map instead of introducing a new version.

.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) {
Expand Down
9 changes: 8 additions & 1 deletion ...rver/src/test/java/org/apache/hadoop/hbase/io/hfile/bucket/TestVerifyBucketCacheFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down