From cae55e9bd58fc21e0e7834e1ecde7e48d7459088 Mon Sep 17 00:00:00 2001 From: Duo Zhang Date: Sat, 10 Jan 2026 22:04:34 +0800 Subject: [PATCH 1/2] HBASE-29783 Fix flaky TestVerifyBucketCacheFile.testRetrieveFromFile test --- .../hbase/io/hfile/bucket/BucketCache.java | 34 +++++++++---------- .../bucket/TestVerifyBucketCacheFile.java | 6 ++-- 2 files changed, 19 insertions(+), 21 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 60ba9f32cd7e..abf7e075c7a0 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 @@ -57,7 +57,6 @@ import java.util.function.Consumer; import java.util.function.Function; import org.apache.commons.io.IOUtils; -import org.apache.commons.lang3.mutable.MutableInt; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HBaseConfiguration; @@ -229,8 +228,6 @@ protected enum CacheState { /** Cache access count (sequential ID) */ private final AtomicLong accessCount = new AtomicLong(); - private static final int DEFAULT_CACHE_WAIT_TIME = 50; - private final BucketCacheStats cacheStats; private final String persistencePath; static AtomicBoolean isCacheInconsistent = new AtomicBoolean(false); @@ -2385,7 +2382,7 @@ public void notifyFileCachingCompleted(Path fileName, int totalBlockCount, int d LOG.debug("Notifying caching completed for file {}, with total blocks {}, and data blocks {}", fileName, totalBlockCount, dataBlockCount); try { - final MutableInt count = new MutableInt(); + int count = 0; LOG.debug("iterating over {} entries in the backing map", backingMap.size()); Set result = getAllCacheKeysForFile(fileName.getName(), 0, Long.MAX_VALUE); if (result.isEmpty() && StoreFileInfo.isReference(fileName)) { @@ -2393,18 +2390,18 @@ public void notifyFileCachingCompleted(Path fileName, int totalBlockCount, int d StoreFileInfo.getReferredToRegionAndFile(fileName.getName()).getSecond(), 0, Long.MAX_VALUE); } - result.stream().forEach(entry -> { + for (BlockCacheKey entry : result) { LOG.debug("found block for file {} in the backing map. Acquiring read lock for offset {}", fileName.getName(), entry.getOffset()); ReentrantReadWriteLock lock = offsetLock.getLock(entry.getOffset()); lock.readLock().lock(); locks.add(lock); if (backingMap.containsKey(entry) && entry.getBlockType().isData()) { - count.increment(); + count++; } - }); + } // BucketCache would only have data blocks - if (dataBlockCount == count.getValue()) { + if (dataBlockCount == count) { LOG.debug("File {} has now been fully cached.", fileName); fileCacheCompleted(fileName, size); } else { @@ -2412,7 +2409,7 @@ public void notifyFileCachingCompleted(Path fileName, int totalBlockCount, int d "Prefetch executor completed for {}, but only {} data blocks were cached. " + "Total data blocks for file: {}. " + "Checking for blocks pending cache in cache writer queue.", - fileName, count.getValue(), dataBlockCount); + fileName, count, dataBlockCount); if (ramCache.hasBlocksForFile(fileName.getName())) { for (ReentrantReadWriteLock lock : locks) { lock.readLock().unlock(); @@ -2507,16 +2504,19 @@ boolean isCacheInitialized(String api) { @Override public boolean waitForCacheInitialization(long timeout) { - try { - while (cacheState == CacheState.INITIALIZING) { - if (timeout <= 0) { - break; - } + while (cacheState == CacheState.INITIALIZING) { + if (timeout <= 0) { + break; + } + try { Thread.sleep(100); - timeout -= 100; + } catch (InterruptedException e) { + LOG.warn("Interrupted while waint for cache initialization", e); + Thread.currentThread().interrupt(); + break; } - } finally { - return isCacheEnabled(); + timeout -= 100; } + return isCacheEnabled(); } } 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 679b9098607f..5ea88c0208a8 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 @@ -46,7 +46,6 @@ import org.apache.hadoop.hbase.io.hfile.Cacheable; import org.apache.hadoop.hbase.testclassification.SmallTests; import org.apache.hadoop.hbase.util.EnvironmentEdgeManager; -import org.apache.hadoop.hbase.util.Pair; import org.junit.ClassRule; import org.junit.Test; import org.junit.experimental.categories.Category; @@ -138,7 +137,8 @@ public void testRetrieveFromFile() throws Exception { recoveredBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize, constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence"); - waitPersistentCacheValidation(conf, bucketCache); + assertTrue(recoveredBucketCache.waitForCacheInitialization(10000)); + waitPersistentCacheValidation(conf, recoveredBucketCache); assertEquals(0, recoveredBucketCache.getAllocator().getUsedSize()); assertEquals(0, recoveredBucketCache.backingMap.size()); BlockCacheKey[] newKeys = CacheTestUtils.regenerateKeys(blocks, names); @@ -308,8 +308,6 @@ public void testModifiedBucketCacheFileTime() throws Exception { long usedSize = bucketCache.getAllocator().getUsedSize(); assertEquals(0, usedSize); - Pair myPair = new Pair<>(); - CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1); // Add blocks From 2713bcfe9606517e8921eea3ebe09435bc6f7e54 Mon Sep 17 00:00:00 2001 From: Duo Zhang Date: Mon, 12 Jan 2026 16:20:01 +0800 Subject: [PATCH 2/2] Update hbase-server/src/main/java/org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java Co-authored-by: Aman Poonia --- .../org/apache/hadoop/hbase/io/hfile/bucket/BucketCache.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 abf7e075c7a0..ba302d7c42ce 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 @@ -2511,7 +2511,7 @@ public boolean waitForCacheInitialization(long timeout) { try { Thread.sleep(100); } catch (InterruptedException e) { - LOG.warn("Interrupted while waint for cache initialization", e); + LOG.warn("Interrupted while waiting for cache initialization", e); Thread.currentThread().interrupt(); break; }