001/* 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018package org.apache.hadoop.hbase.io.hfile.bucket; 019 020import static org.apache.hadoop.hbase.io.hfile.CacheConfig.BUCKETCACHE_PERSIST_INTERVAL_KEY; 021import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.ACCEPT_FACTOR_CONFIG_NAME; 022import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.DEFAULT_ERROR_TOLERATION_DURATION; 023import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.EXTRA_FREE_FACTOR_CONFIG_NAME; 024import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.MIN_FACTOR_CONFIG_NAME; 025import static org.awaitility.Awaitility.await; 026import static org.junit.jupiter.api.Assertions.assertEquals; 027import static org.junit.jupiter.api.Assertions.assertNull; 028import static org.junit.jupiter.api.Assertions.assertTrue; 029 030import java.time.Duration; 031import org.apache.hadoop.conf.Configuration; 032import org.apache.hadoop.fs.Path; 033import org.apache.hadoop.hbase.HBaseConfiguration; 034import org.apache.hadoop.hbase.HBaseTestingUtil; 035import org.apache.hadoop.hbase.Waiter; 036import org.apache.hadoop.hbase.io.hfile.BlockCacheKey; 037import org.apache.hadoop.hbase.io.hfile.CacheTestUtils; 038import org.apache.hadoop.hbase.io.hfile.Cacheable; 039import org.apache.hadoop.hbase.testclassification.RegionServerTests; 040import org.apache.hadoop.hbase.testclassification.SmallTests; 041import org.junit.jupiter.api.Tag; 042import org.junit.jupiter.api.Test; 043 044/** 045 * Basic test for check file's integrity before start BucketCache in fileIOEngine 046 */ 047@Tag(SmallTests.TAG) 048@Tag(RegionServerTests.TAG) 049public class TestRecoveryPersistentBucketCache { 050 051 final long capacitySize = 32 * 1024 * 1024; 052 final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS; 053 final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS; 054 055 @Test 056 public void testBucketCacheRecovery() throws Exception { 057 HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 058 Path testDir = TEST_UTIL.getDataTestDir(); 059 TEST_UTIL.getTestFileSystem().mkdirs(testDir); 060 Configuration conf = HBaseConfiguration.create(); 061 // Disables the persister thread by setting its interval to MAX_VALUE 062 conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE); 063 int[] bucketSizes = new int[] { 8 * 1024 + 1024 }; 064 BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, 065 8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence", 066 DEFAULT_ERROR_TOLERATION_DURATION, conf); 067 assertTrue(bucketCache.waitForCacheInitialization(1000)); 068 assertTrue( 069 bucketCache.isCacheInitialized("testBucketCacheRecovery") && bucketCache.isCacheEnabled()); 070 071 CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 4); 072 String[] names = CacheTestUtils.getHFileNames(blocks); 073 074 CacheTestUtils.HFileBlockPair[] smallerBlocks = CacheTestUtils.generateHFileBlocks(4096, 1); 075 String[] smallerNames = CacheTestUtils.getHFileNames(smallerBlocks); 076 // Add four blocks 077 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock()); 078 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock()); 079 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName(), blocks[2].getBlock()); 080 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[3].getBlockName(), blocks[3].getBlock()); 081 // saves the current state of the cache 082 bucketCache.persistToFile(); 083 // evicts the 4th block 084 bucketCache.evictBlock(blocks[3].getBlockName()); 085 // now adds a 5th block to bucket cache. This block is half the size of the previous 086 // blocks, and it will be added in the same offset of the previous evicted block. 087 // This overwrites part of the 4th block. Because we persisted only up to the 088 // 4th block addition, recovery would try to read the whole 4th block, but the cached time 089 // validation will fail, and we'll recover only the first three blocks 090 cacheAndWaitUntilFlushedToBucket(bucketCache, smallerBlocks[0].getBlockName(), 091 smallerBlocks[0].getBlock()); 092 093 // Creates new bucket cache instance without persisting to file after evicting 4th block 094 // and caching 5th block. Here the cache file has the first three blocks, followed by the 095 // 5th block and the second half of 4th block (we evicted 4th block, freeing up its 096 // offset in the cache, then added 5th block which is half the size of other blocks, so it's 097 // going to override the first half of the 4th block in the cache). That's fine because 098 // the in-memory backing map has the right blocks and related offsets. However, the 099 // persistent map file only has information about the first four blocks. We validate the 100 // cache time recorded in the back map against the block data in the cache. This is recorded 101 // in the cache as the first 8 bytes of a block, so the 4th block had its first 8 blocks 102 // now overridden by the 5th block, causing this check to fail and removal of 103 // the 4th block from the backing map. 104 BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, 105 8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence", 106 DEFAULT_ERROR_TOLERATION_DURATION, conf); 107 assertTrue(newBucketCache.waitForCacheInitialization(1000)); 108 BlockCacheKey[] newKeys = CacheTestUtils.regenerateKeys(blocks, names); 109 BlockCacheKey[] newKeysSmaller = CacheTestUtils.regenerateKeys(smallerBlocks, smallerNames); 110 // The new bucket cache would have only the first three blocks. Although we have persisted the 111 // the cache state when it had the first four blocks, the 4th block was evicted and then we 112 // added a 5th block, which overrides part of the 4th block in the cache. This would cause a 113 // checksum failure for this block offset, when we try to read from the cache, and we would 114 // consider that block as invalid and its offset available in the cache. 115 assertNull(newBucketCache.getBlock(newKeys[3], false, false, false)); 116 assertNull(newBucketCache.getBlock(newKeysSmaller[0], false, false, false)); 117 assertEquals(blocks[0].getBlock(), newBucketCache.getBlock(newKeys[0], false, false, false)); 118 assertEquals(blocks[1].getBlock(), newBucketCache.getBlock(newKeys[1], false, false, false)); 119 assertEquals(blocks[2].getBlock(), newBucketCache.getBlock(newKeys[2], false, false, false)); 120 TEST_UTIL.cleanupTestDir(); 121 } 122 123 @Test 124 public void testBucketCacheEvictByHFileAfterRecovery() throws Exception { 125 HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 126 Path testDir = TEST_UTIL.getDataTestDir(); 127 TEST_UTIL.getTestFileSystem().mkdirs(testDir); 128 Configuration conf = HBaseConfiguration.create(); 129 // Disables the persister thread by setting its interval to MAX_VALUE 130 conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE); 131 int[] bucketSizes = new int[] { 8 * 1024 + 1024 }; 132 BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, 133 8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence", 134 DEFAULT_ERROR_TOLERATION_DURATION, conf); 135 assertTrue(bucketCache.waitForCacheInitialization(10000)); 136 137 CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 4); 138 139 // Add four blocks 140 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock()); 141 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock()); 142 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName(), blocks[2].getBlock()); 143 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[3].getBlockName(), blocks[3].getBlock()); 144 145 String firstFileName = blocks[0].getBlockName().getHfileName(); 146 147 // saves the current state of the cache 148 bucketCache.persistToFile(); 149 150 BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, 151 8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence", 152 DEFAULT_ERROR_TOLERATION_DURATION, conf); 153 assertTrue(newBucketCache.waitForCacheInitialization(10000)); 154 assertEquals(4, newBucketCache.backingMap.size()); 155 156 newBucketCache.evictBlocksByHfileName(firstFileName); 157 assertEquals(3, newBucketCache.backingMap.size()); 158 TEST_UTIL.cleanupTestDir(); 159 } 160 161 @Test 162 public void testValidateCacheInitialization() throws Exception { 163 HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 164 Path testDir = TEST_UTIL.getDataTestDir(); 165 TEST_UTIL.getTestFileSystem().mkdirs(testDir); 166 Configuration conf = HBaseConfiguration.create(); 167 // Disables the persister thread by setting its interval to MAX_VALUE 168 conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE); 169 int[] bucketSizes = new int[] { 8 * 1024 + 1024 }; 170 BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, 171 8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence", 172 DEFAULT_ERROR_TOLERATION_DURATION, conf); 173 assertTrue(bucketCache.waitForCacheInitialization(10000)); 174 175 CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 4); 176 177 // Add four blocks 178 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock()); 179 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock()); 180 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName(), blocks[2].getBlock()); 181 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[3].getBlockName(), blocks[3].getBlock()); 182 // saves the current state of the cache 183 bucketCache.persistToFile(); 184 185 BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, 186 8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence", 187 DEFAULT_ERROR_TOLERATION_DURATION, conf); 188 assertTrue(newBucketCache.waitForCacheInitialization(10000)); 189 190 // Set the state of bucket cache to INITIALIZING 191 newBucketCache.setCacheState(BucketCache.CacheState.INITIALIZING); 192 193 // Validate that zero values are returned for the cache being initialized. 194 assertEquals(0, newBucketCache.acceptableSize()); 195 assertEquals(0, newBucketCache.getPartitionSize(1)); 196 assertEquals(0, newBucketCache.getFreeSize()); 197 assertEquals(0, newBucketCache.getCurrentSize()); 198 assertEquals(false, newBucketCache.blockFitsIntoTheCache(blocks[0].getBlock()).get()); 199 200 newBucketCache.setCacheState(BucketCache.CacheState.ENABLED); 201 202 // Validate that non-zero values are returned for enabled cache 203 assertTrue(newBucketCache.acceptableSize() > 0); 204 assertTrue(newBucketCache.getPartitionSize(1) > 0); 205 assertTrue(newBucketCache.getFreeSize() > 0); 206 assertTrue(newBucketCache.getCurrentSize() > 0); 207 assertTrue(newBucketCache.blockFitsIntoTheCache(blocks[0].getBlock()).get()); 208 209 TEST_UTIL.cleanupTestDir(); 210 } 211 212 @Test 213 public void testBucketCacheRecoveryWithAllocationInconsistencies() throws Exception { 214 HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil(); 215 Path testDir = TEST_UTIL.getDataTestDir(); 216 TEST_UTIL.getTestFileSystem().mkdirs(testDir); 217 Configuration conf = HBaseConfiguration.create(); 218 // Disables the persister thread by setting its interval to MAX_VALUE 219 conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE); 220 conf.setDouble(MIN_FACTOR_CONFIG_NAME, 0.99); 221 conf.setDouble(ACCEPT_FACTOR_CONFIG_NAME, 1); 222 conf.setDouble(EXTRA_FREE_FACTOR_CONFIG_NAME, 0.01); 223 int[] bucketSizes = new int[] { 8 * 1024 + 1024 }; 224 BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", 36 * 1024, 8192, 225 bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence", 226 DEFAULT_ERROR_TOLERATION_DURATION, conf); 227 assertTrue(bucketCache.waitForCacheInitialization(1000)); 228 assertTrue( 229 bucketCache.isCacheInitialized("testBucketCacheRecovery") && bucketCache.isCacheEnabled()); 230 231 CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 5); 232 String[] names = CacheTestUtils.getHFileNames(blocks); 233 234 // Add four blocks 235 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock()); 236 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock()); 237 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName(), blocks[2].getBlock()); 238 cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[3].getBlockName(), blocks[3].getBlock()); 239 240 // creates a entry for a 5th block with the same cache offset of the 1st block. Just add it 241 // straight to the backingMap, bypassing caching, in order to fabricate an inconsistency 242 BucketEntry bucketEntry = 243 new BucketEntry(bucketCache.backingMap.get(blocks[0].getBlockName()).offset(), 244 blocks[4].getBlock().getSerializedLength(), blocks[4].getBlock().getOnDiskSizeWithHeader(), 245 0, false, bucketCache::createRecycler, blocks[4].getBlock().getByteBuffAllocator()); 246 bucketEntry.setDeserializerReference(blocks[4].getBlock().getDeserializer()); 247 bucketCache.getBackingMap().put(blocks[4].getBlockName(), bucketEntry); 248 249 // saves the current state of the cache: 5 blocks in the map, but we only have cached 4. The 250 // 5th block has same cache offset as the first 251 bucketCache.persistToFile(); 252 253 BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", 36 * 1024, 254 8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence", 255 DEFAULT_ERROR_TOLERATION_DURATION, conf); 256 // backingMapValidated is set inside retrieveFromFile(), but isCacheEnabled() (which 257 // getBlock() checks) is set to true only in the finally block that follows. Waiting on 258 // backingMapValidated alone leaves a race window; await both conditions together. 259 await().atMost(Duration.ofSeconds(30)).pollInterval(Duration.ofMillis(10)).until( 260 () -> newBucketCache.getBackingMapValidated().get() && newBucketCache.isCacheEnabled()); 261 262 BlockCacheKey[] newKeys = CacheTestUtils.regenerateKeys(blocks, names); 263 264 assertNull(newBucketCache.getBlock(newKeys[4], false, false, false)); 265 // The backing map entry with key blocks[0].getBlockName() for the may point to a valid entry 266 // or null based on different ordering of the keys in the backing map. 267 // Hence, skipping the check for that key. 268 assertEquals(blocks[1].getBlock(), newBucketCache.getBlock(newKeys[1], false, false, false)); 269 assertEquals(blocks[2].getBlock(), newBucketCache.getBlock(newKeys[2], false, false, false)); 270 assertEquals(blocks[3].getBlock(), newBucketCache.getBlock(newKeys[3], false, false, false)); 271 assertEquals(4, newBucketCache.backingMap.size()); 272 TEST_UTIL.cleanupTestDir(); 273 } 274 275 private void waitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey) 276 throws InterruptedException { 277 Waiter.waitFor(HBaseConfiguration.create(), 12000, 278 () -> (cache.backingMap.containsKey(cacheKey) && !cache.ramCache.containsKey(cacheKey))); 279 } 280 281 // BucketCache.cacheBlock is async, it first adds block to ramCache and writeQueue, then writer 282 // threads will flush it to the bucket and put reference entry in backingMap. 283 private void cacheAndWaitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey, 284 Cacheable block) throws InterruptedException { 285 cache.cacheBlock(cacheKey, block); 286 waitUntilFlushedToBucket(cache, cacheKey); 287 } 288 289}