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.BACKING_MAP_PERSISTENCE_CHUNK_SIZE;
023import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.BLOCK_ORPHAN_GRACE_PERIOD;
024import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.DEFAULT_ERROR_TOLERATION_DURATION;
025import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.DEFAULT_MIN_FACTOR;
026import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.DEFAULT_SINGLE_FACTOR;
027import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.EXTRA_FREE_FACTOR_CONFIG_NAME;
028import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.MEMORY_FACTOR_CONFIG_NAME;
029import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.MIN_FACTOR_CONFIG_NAME;
030import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.MULTI_FACTOR_CONFIG_NAME;
031import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.QUEUE_ADDITION_WAIT_TIME;
032import static org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.SINGLE_FACTOR_CONFIG_NAME;
033import static org.junit.jupiter.api.Assertions.assertEquals;
034import static org.junit.jupiter.api.Assertions.assertFalse;
035import static org.junit.jupiter.api.Assertions.assertNotEquals;
036import static org.junit.jupiter.api.Assertions.assertNotNull;
037import static org.junit.jupiter.api.Assertions.assertNull;
038import static org.junit.jupiter.api.Assertions.assertSame;
039import static org.junit.jupiter.api.Assertions.assertTrue;
040import static org.junit.jupiter.api.Assertions.fail;
041import static org.mockito.Mockito.mock;
042import static org.mockito.Mockito.when;
043
044import java.io.File;
045import java.io.IOException;
046import java.lang.reflect.Field;
047import java.nio.ByteBuffer;
048import java.util.ArrayList;
049import java.util.Arrays;
050import java.util.Collection;
051import java.util.HashMap;
052import java.util.List;
053import java.util.Map;
054import java.util.Set;
055import java.util.concurrent.ThreadLocalRandom;
056import java.util.concurrent.atomic.LongAdder;
057import java.util.concurrent.locks.ReentrantReadWriteLock;
058import java.util.stream.Stream;
059import org.apache.hadoop.conf.Configuration;
060import org.apache.hadoop.fs.Path;
061import org.apache.hadoop.hbase.HBaseConfiguration;
062import org.apache.hadoop.hbase.HBaseParameterizedTestTemplate;
063import org.apache.hadoop.hbase.HBaseTestingUtil;
064import org.apache.hadoop.hbase.HConstants;
065import org.apache.hadoop.hbase.Waiter;
066import org.apache.hadoop.hbase.io.ByteBuffAllocator;
067import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
068import org.apache.hadoop.hbase.io.hfile.BlockPriority;
069import org.apache.hadoop.hbase.io.hfile.BlockType;
070import org.apache.hadoop.hbase.io.hfile.CacheStats;
071import org.apache.hadoop.hbase.io.hfile.CacheTestUtils;
072import org.apache.hadoop.hbase.io.hfile.CacheTestUtils.HFileBlockPair;
073import org.apache.hadoop.hbase.io.hfile.Cacheable;
074import org.apache.hadoop.hbase.io.hfile.HFileBlock;
075import org.apache.hadoop.hbase.io.hfile.HFileContext;
076import org.apache.hadoop.hbase.io.hfile.HFileContextBuilder;
077import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.BucketSizeInfo;
078import org.apache.hadoop.hbase.io.hfile.bucket.BucketAllocator.IndexStatistics;
079import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.RAMCache;
080import org.apache.hadoop.hbase.io.hfile.bucket.BucketCache.RAMQueueEntry;
081import org.apache.hadoop.hbase.nio.ByteBuff;
082import org.apache.hadoop.hbase.regionserver.HRegion;
083import org.apache.hadoop.hbase.regionserver.HStore;
084import org.apache.hadoop.hbase.regionserver.HStoreFile;
085import org.apache.hadoop.hbase.testclassification.IOTests;
086import org.apache.hadoop.hbase.testclassification.LargeTests;
087import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
088import org.apache.hadoop.hbase.util.Pair;
089import org.apache.hadoop.hbase.util.Threads;
090import org.junit.jupiter.api.AfterEach;
091import org.junit.jupiter.api.BeforeEach;
092import org.junit.jupiter.api.Tag;
093import org.junit.jupiter.api.TestTemplate;
094import org.junit.jupiter.params.provider.Arguments;
095import org.mockito.Mockito;
096import org.slf4j.Logger;
097import org.slf4j.LoggerFactory;
098
099import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap;
100
101/**
102 * Basic test of BucketCache.Puts and gets.
103 * <p>
104 * Tests will ensure that blocks' data correctness under several threads concurrency
105 */
106@Tag(IOTests.TAG)
107@Tag(LargeTests.TAG)
108@HBaseParameterizedTestTemplate(name = "{index}: blockSize={0}, bucketSizes={1}")
109public class TestBucketCache {
110
111  private static final Logger LOG = LoggerFactory.getLogger(TestBucketCache.class);
112
113  public static Stream<Arguments> parameters() {
114    // TODO: why is 8k the default blocksize for these tests?
115    return Stream.of(Arguments.of(8192, null),
116      Arguments.of(16 * 1024,
117        new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024, 8 * 1024 + 1024, 16 * 1024 + 1024,
118          28 * 1024 + 1024, 32 * 1024 + 1024, 64 * 1024 + 1024, 96 * 1024 + 1024,
119          128 * 1024 + 1024 }));
120  }
121
122  private final int constructedBlockSize;
123  private final int[] constructedBlockSizes;
124
125  public TestBucketCache(int constructedBlockSize, int[] constructedBlockSizes) {
126    this.constructedBlockSize = constructedBlockSize;
127    this.constructedBlockSizes = constructedBlockSizes;
128  }
129
130  BucketCache cache;
131  final int CACHE_SIZE = 1000000;
132  final int NUM_BLOCKS = 100;
133  final int BLOCK_SIZE = CACHE_SIZE / NUM_BLOCKS;
134  final int NUM_THREADS = 100;
135  final int NUM_QUERIES = 10000;
136
137  final long capacitySize = 32 * 1024 * 1024;
138  final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS;
139  final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS;
140  private String ioEngineName = "offheap";
141
142  private static final HBaseTestingUtil HBASE_TESTING_UTILITY = new HBaseTestingUtil();
143
144  private static class MockedBucketCache extends BucketCache {
145
146    public MockedBucketCache(String ioEngineName, long capacity, int blockSize, int[] bucketSizes,
147      int writerThreads, int writerQLen, String persistencePath) throws IOException {
148      super(ioEngineName, capacity, blockSize, bucketSizes, writerThreads, writerQLen,
149        persistencePath);
150    }
151
152    @Override
153    public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf, boolean inMemory) {
154      super.cacheBlock(cacheKey, buf, inMemory);
155    }
156
157    @Override
158    public void cacheBlock(BlockCacheKey cacheKey, Cacheable buf) {
159      super.cacheBlock(cacheKey, buf);
160    }
161  }
162
163  @BeforeEach
164  public void setup() throws IOException {
165    cache = new MockedBucketCache(ioEngineName, capacitySize, constructedBlockSize,
166      constructedBlockSizes, writeThreads, writerQLen, null);
167  }
168
169  @AfterEach
170  public void tearDown() {
171    cache.shutdown();
172  }
173
174  /**
175   * Test Utility to create test dir and return name
176   * @return return name of created dir
177   * @throws IOException throws IOException
178   */
179  private Path createAndGetTestDir() throws IOException {
180    final Path testDir = HBASE_TESTING_UTILITY.getDataTestDir();
181    HBASE_TESTING_UTILITY.getTestFileSystem().mkdirs(testDir);
182    return testDir;
183  }
184
185  /**
186   * Return a random element from {@code a}.
187   */
188  private static <T> T randFrom(List<T> a) {
189    return a.get(ThreadLocalRandom.current().nextInt(a.size()));
190  }
191
192  @TestTemplate
193  public void testBucketAllocator() throws BucketAllocatorException {
194    BucketAllocator mAllocator = cache.getAllocator();
195    /*
196     * Test the allocator first
197     */
198    final List<Integer> BLOCKSIZES = Arrays.asList(4 * 1024, 8 * 1024, 64 * 1024, 96 * 1024);
199
200    boolean full = false;
201    ArrayList<Pair<Long, Integer>> allocations = new ArrayList<>();
202    // Fill the allocated extents by choosing a random blocksize. Continues selecting blocks until
203    // the cache is completely filled.
204    List<Integer> tmp = new ArrayList<>(BLOCKSIZES);
205    while (!full) {
206      Integer blockSize = null;
207      try {
208        blockSize = randFrom(tmp);
209        allocations.add(new Pair<>(mAllocator.allocateBlock(blockSize), blockSize));
210      } catch (CacheFullException cfe) {
211        tmp.remove(blockSize);
212        if (tmp.isEmpty()) full = true;
213      }
214    }
215
216    for (Integer blockSize : BLOCKSIZES) {
217      BucketSizeInfo bucketSizeInfo = mAllocator.roundUpToBucketSizeInfo(blockSize);
218      IndexStatistics indexStatistics = bucketSizeInfo.statistics();
219      assertEquals(0, indexStatistics.freeCount(), "unexpected freeCount for " + bucketSizeInfo);
220
221      // we know the block sizes above are multiples of 1024, but default bucket sizes give an
222      // additional 1024 on top of that so this counts towards fragmentation in our test
223      // real life may have worse fragmentation because blocks may not be perfectly sized to block
224      // size, given encoding/compression and large rows
225      assertEquals(1024 * indexStatistics.totalCount(), indexStatistics.fragmentationBytes());
226    }
227
228    mAllocator.logDebugStatistics();
229
230    for (Pair<Long, Integer> allocation : allocations) {
231      assertEquals(mAllocator.sizeOfAllocation(allocation.getFirst()),
232        mAllocator.freeBlock(allocation.getFirst(), allocation.getSecond()));
233    }
234    assertEquals(0, mAllocator.getUsedSize());
235  }
236
237  @TestTemplate
238  public void testCacheSimple() throws Exception {
239    CacheTestUtils.testCacheSimple(cache, BLOCK_SIZE, NUM_QUERIES);
240  }
241
242  @TestTemplate
243  public void testCacheMultiThreadedSingleKey() throws Exception {
244    CacheTestUtils.hammerSingleKey(cache, 2 * NUM_THREADS, 2 * NUM_QUERIES);
245  }
246
247  @TestTemplate
248  public void testHeapSizeChanges() throws Exception {
249    cache.stopWriterThreads();
250    CacheTestUtils.testHeapSizeChanges(cache, BLOCK_SIZE);
251  }
252
253  public static void waitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey)
254    throws InterruptedException {
255    Waiter.waitFor(HBaseConfiguration.create(), 10000,
256      () -> (cache.backingMap.containsKey(cacheKey) && !cache.ramCache.containsKey(cacheKey)));
257  }
258
259  public static void waitUntilAllFlushedToBucket(BucketCache cache) throws InterruptedException {
260    while (!cache.ramCache.isEmpty()) {
261      Thread.sleep(100);
262    }
263    Thread.sleep(1000);
264  }
265
266  // BucketCache.cacheBlock is async, it first adds block to ramCache and writeQueue, then writer
267  // threads will flush it to the bucket and put reference entry in backingMap.
268  private void cacheAndWaitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey,
269    Cacheable block, boolean waitWhenCache) throws InterruptedException {
270    cache.cacheBlock(cacheKey, block, false, waitWhenCache);
271    waitUntilFlushedToBucket(cache, cacheKey);
272  }
273
274  @TestTemplate
275  public void testMemoryLeak() throws Exception {
276    final BlockCacheKey cacheKey = new BlockCacheKey("dummy", 1L);
277    cacheAndWaitUntilFlushedToBucket(cache, cacheKey,
278      new CacheTestUtils.ByteArrayCacheable(new byte[10]), true);
279    BucketEntry oldEntry = cache.backingMap.get(cacheKey);
280    long lockId = oldEntry.offset();
281    ReentrantReadWriteLock lock = cache.offsetLock.getLock(lockId);
282    Thread evictThread = new Thread("evict-block") {
283      @Override
284      public void run() {
285        cache.evictBlock(cacheKey);
286      }
287    };
288    BucketEntry replacementEntry;
289    lock.writeLock().lock();
290    try {
291      evictThread.start();
292      cache.offsetLock.waitForWaiters(lockId, 1);
293      assertTrue(cache.backingMap.remove(cacheKey, oldEntry));
294      cache.blockEvicted(cacheKey, oldEntry, true, true);
295      assertEquals(0, cache.getBlockCount());
296      cache.cacheBlock(cacheKey, new CacheTestUtils.ByteArrayCacheable(new byte[10]), false, true);
297      // The replacement is published before taking its offset lock. Full flush must wait until the
298      // old entry's lock is released.
299      Waiter.waitFor(HBaseConfiguration.create(), 10000, () -> {
300        BucketEntry currentEntry = cache.backingMap.get(cacheKey);
301        return currentEntry != null && currentEntry != oldEntry;
302      });
303      replacementEntry = cache.backingMap.get(cacheKey);
304    } finally {
305      lock.writeLock().unlock();
306    }
307    waitUntilFlushedToBucket(cache, cacheKey);
308    assertEquals(1, cache.getBlockCount());
309    evictThread.join();
310    /**
311     * <pre>
312     * The asserts here before HBASE-21957 are:
313     * assertEquals(1L, cache.getBlockCount());
314     * assertTrue(cache.getCurrentSize() > 0L);
315     * assertTrue("We should have a block!", cache.iterator().hasNext());
316     *
317     * The asserts here after HBASE-21957 are:
318     * assertEquals(0, cache.getBlockCount());
319     * assertEquals(cache.getCurrentSize(), 0L);
320     *
321     * I think the asserts before HBASE-21957 is more reasonable,because
322     * {@link BucketCache#evictBlock} should only evict the {@link BucketEntry}
323     * it had seen, and newly added Block after the {@link BucketEntry}
324     * it had seen should not be evicted.
325     * </pre>
326     */
327    assertSame(replacementEntry, cache.backingMap.get(cacheKey));
328    assertTrue(cache.blocksByHFile.contains(cacheKey));
329    assertEquals(1L, cache.getBlockCount());
330    assertTrue(cache.getCurrentSize() > 0L);
331    assertTrue(cache.iterator().hasNext(), "We should have a block!");
332  }
333
334  @TestTemplate
335  public void testRetrieveFromFile() throws Exception {
336    Path testDir = createAndGetTestDir();
337    String ioEngineName = "file:" + testDir + "/bucket.cache";
338    testRetrievalUtils(testDir, ioEngineName);
339    int[] smallBucketSizes = new int[] { 3 * 1024, 5 * 1024 };
340    String persistencePath = testDir + "/bucket.persistence";
341    BucketCache bucketCache = null;
342    try {
343      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
344        smallBucketSizes, writeThreads, writerQLen, persistencePath);
345      assertTrue(bucketCache.waitForCacheInitialization(10000));
346      assertFalse(new File(persistencePath).exists());
347      assertEquals(0, bucketCache.getAllocator().getUsedSize());
348      assertEquals(0, bucketCache.backingMap.size());
349    } finally {
350      bucketCache.shutdown();
351      HBASE_TESTING_UTILITY.cleanupTestDir();
352    }
353  }
354
355  @TestTemplate
356  public void testRetrieveFromMMap() throws Exception {
357    final Path testDir = createAndGetTestDir();
358    final String ioEngineName = "mmap:" + testDir + "/bucket.cache";
359    testRetrievalUtils(testDir, ioEngineName);
360  }
361
362  @TestTemplate
363  public void testRetrieveFromPMem() throws Exception {
364    final Path testDir = createAndGetTestDir();
365    final String ioEngineName = "pmem:" + testDir + "/bucket.cache";
366    testRetrievalUtils(testDir, ioEngineName);
367    int[] smallBucketSizes = new int[] { 3 * 1024, 5 * 1024 };
368    String persistencePath = testDir + "/bucket.persistence" + EnvironmentEdgeManager.currentTime();
369    BucketCache bucketCache = null;
370    try {
371      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
372        smallBucketSizes, writeThreads, writerQLen, persistencePath);
373      assertTrue(bucketCache.waitForCacheInitialization(10000));
374      assertFalse(new File(persistencePath).exists());
375      assertEquals(0, bucketCache.getAllocator().getUsedSize());
376      assertEquals(0, bucketCache.backingMap.size());
377    } finally {
378      bucketCache.shutdown();
379      HBASE_TESTING_UTILITY.cleanupTestDir();
380    }
381  }
382
383  private void testRetrievalUtils(Path testDir, String ioEngineName)
384    throws IOException, InterruptedException {
385    final String persistencePath =
386      testDir + "/bucket.persistence" + EnvironmentEdgeManager.currentTime();
387    BucketCache bucketCache = null;
388    try {
389      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
390        constructedBlockSizes, writeThreads, writerQLen, persistencePath);
391      assertTrue(bucketCache.waitForCacheInitialization(10000));
392      long usedSize = bucketCache.getAllocator().getUsedSize();
393      assertEquals(0, usedSize);
394      HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1);
395      for (HFileBlockPair block : blocks) {
396        bucketCache.cacheBlock(block.getBlockName(), block.getBlock());
397      }
398      for (HFileBlockPair block : blocks) {
399        cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock(),
400          false);
401      }
402      usedSize = bucketCache.getAllocator().getUsedSize();
403      assertNotEquals(0, usedSize);
404      BucketEntry persistedEntry = bucketCache.backingMap.values().iterator().next();
405      assertEquals(1, persistedEntry.refCnt());
406      bucketCache.shutdown();
407      assertEquals(0, persistedEntry.refCnt());
408      assertTrue(bucketCache.backingMap.isEmpty());
409      assertTrue(new File(persistencePath).exists());
410      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
411        constructedBlockSizes, writeThreads, writerQLen, persistencePath);
412      assertTrue(bucketCache.waitForCacheInitialization(10000));
413
414      assertEquals(usedSize, bucketCache.getAllocator().getUsedSize());
415    } finally {
416      if (bucketCache != null) {
417        bucketCache.shutdown();
418      }
419    }
420    assertTrue(new File(persistencePath).exists());
421  }
422
423  @TestTemplate
424  public void testRetrieveUnsupportedIOE() throws Exception {
425    try {
426      final Path testDir = createAndGetTestDir();
427      final String ioEngineName = testDir + "/bucket.cache";
428      testRetrievalUtils(testDir, ioEngineName);
429      fail("Should have thrown IllegalArgumentException because of unsupported IOEngine!!");
430    } catch (IllegalArgumentException e) {
431      assertEquals("Don't understand io engine name for cache- prefix with file:, "
432        + "files:, mmap: or offheap", e.getMessage());
433    }
434  }
435
436  @TestTemplate
437  public void testRetrieveFromMultipleFiles() throws Exception {
438    final Path testDirInitial = createAndGetTestDir();
439    final Path newTestDir = new HBaseTestingUtil().getDataTestDir();
440    HBASE_TESTING_UTILITY.getTestFileSystem().mkdirs(newTestDir);
441    String ioEngineName =
442      new StringBuilder("files:").append(testDirInitial).append("/bucket1.cache")
443        .append(FileIOEngine.FILE_DELIMITER).append(newTestDir).append("/bucket2.cache").toString();
444    testRetrievalUtils(testDirInitial, ioEngineName);
445    int[] smallBucketSizes = new int[] { 3 * 1024, 5 * 1024 };
446    String persistencePath = testDirInitial + "/bucket.persistence";
447    BucketCache bucketCache = null;
448    try {
449      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
450        smallBucketSizes, writeThreads, writerQLen, persistencePath);
451      assertTrue(bucketCache.waitForCacheInitialization(10000));
452      assertFalse(new File(persistencePath).exists());
453      assertEquals(0, bucketCache.getAllocator().getUsedSize());
454      assertEquals(0, bucketCache.backingMap.size());
455    } finally {
456      bucketCache.shutdown();
457      HBASE_TESTING_UTILITY.cleanupTestDir();
458    }
459  }
460
461  @TestTemplate
462  public void testRetrieveFromFileWithoutPersistence() throws Exception {
463    BucketCache bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
464      constructedBlockSizes, writeThreads, writerQLen, null);
465    assertTrue(bucketCache.waitForCacheInitialization(10000));
466    try {
467      final Path testDir = createAndGetTestDir();
468      String ioEngineName = "file:" + testDir + "/bucket.cache";
469      long usedSize = bucketCache.getAllocator().getUsedSize();
470      assertEquals(0, usedSize);
471      HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1);
472      for (HFileBlockPair block : blocks) {
473        bucketCache.cacheBlock(block.getBlockName(), block.getBlock());
474      }
475      for (HFileBlockPair block : blocks) {
476        cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock(),
477          false);
478      }
479      usedSize = bucketCache.getAllocator().getUsedSize();
480      assertNotEquals(0, usedSize);
481      bucketCache.shutdown();
482      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
483        constructedBlockSizes, writeThreads, writerQLen, null);
484      assertTrue(bucketCache.waitForCacheInitialization(10000));
485      assertEquals(0, bucketCache.getAllocator().getUsedSize());
486    } finally {
487      bucketCache.shutdown();
488      HBASE_TESTING_UTILITY.cleanupTestDir();
489    }
490  }
491
492  @TestTemplate
493  public void testBucketAllocatorLargeBuckets() throws BucketAllocatorException {
494    long availableSpace = 20 * 1024L * 1024 * 1024;
495    int[] bucketSizes = new int[] { 1024, 1024 * 1024, 1024 * 1024 * 1024 };
496    BucketAllocator allocator = new BucketAllocator(availableSpace, bucketSizes);
497    assertTrue(allocator.getBuckets().length > 0);
498  }
499
500  @TestTemplate
501  public void testGetPartitionSize() throws IOException {
502    // Test default values
503    validateGetPartitionSize(cache, DEFAULT_SINGLE_FACTOR, DEFAULT_MIN_FACTOR);
504
505    Configuration conf = HBaseConfiguration.create();
506    conf.setFloat(MIN_FACTOR_CONFIG_NAME, 0.5f);
507    conf.setFloat(SINGLE_FACTOR_CONFIG_NAME, 0.1f);
508    conf.setFloat(MULTI_FACTOR_CONFIG_NAME, 0.7f);
509    conf.setFloat(MEMORY_FACTOR_CONFIG_NAME, 0.2f);
510
511    BucketCache cache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
512      constructedBlockSizes, writeThreads, writerQLen, null, 100, conf);
513    assertTrue(cache.waitForCacheInitialization(10000));
514
515    validateGetPartitionSize(cache, 0.1f, 0.5f);
516    validateGetPartitionSize(cache, 0.7f, 0.5f);
517    validateGetPartitionSize(cache, 0.2f, 0.5f);
518  }
519
520  @TestTemplate
521  public void testCacheSizeCapacity() throws IOException {
522    // Test cache capacity (capacity / blockSize) < Integer.MAX_VALUE
523    validateGetPartitionSize(cache, DEFAULT_SINGLE_FACTOR, DEFAULT_MIN_FACTOR);
524    Configuration conf = HBaseConfiguration.create();
525    conf.setFloat(BucketCache.MIN_FACTOR_CONFIG_NAME, 0.5f);
526    conf.setFloat(SINGLE_FACTOR_CONFIG_NAME, 0.1f);
527    conf.setFloat(MULTI_FACTOR_CONFIG_NAME, 0.7f);
528    conf.setFloat(MEMORY_FACTOR_CONFIG_NAME, 0.2f);
529    try {
530      new BucketCache(ioEngineName, Long.MAX_VALUE, 1, constructedBlockSizes, writeThreads,
531        writerQLen, null, 100, conf);
532      fail("Should have thrown IllegalArgumentException because of large cache capacity!");
533    } catch (IllegalArgumentException e) {
534      assertEquals("Cache capacity is too large, only support 32TB now", e.getMessage());
535    }
536  }
537
538  @TestTemplate
539  public void testValidBucketCacheConfigs() throws IOException {
540    Configuration conf = HBaseConfiguration.create();
541    conf.setFloat(ACCEPT_FACTOR_CONFIG_NAME, 0.9f);
542    conf.setFloat(MIN_FACTOR_CONFIG_NAME, 0.5f);
543    conf.setFloat(EXTRA_FREE_FACTOR_CONFIG_NAME, 0.5f);
544    conf.setFloat(SINGLE_FACTOR_CONFIG_NAME, 0.1f);
545    conf.setFloat(MULTI_FACTOR_CONFIG_NAME, 0.7f);
546    conf.setFloat(MEMORY_FACTOR_CONFIG_NAME, 0.2f);
547
548    BucketCache cache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
549      constructedBlockSizes, writeThreads, writerQLen, null, 100, conf);
550    assertTrue(cache.waitForCacheInitialization(10000));
551
552    assertEquals(0.9f, cache.getAcceptableFactor(), 0,
553      ACCEPT_FACTOR_CONFIG_NAME + " failed to propagate.");
554    assertEquals(0.5f, cache.getMinFactor(), 0, MIN_FACTOR_CONFIG_NAME + " failed to propagate.");
555    assertEquals(0.5f, cache.getExtraFreeFactor(), 0,
556      EXTRA_FREE_FACTOR_CONFIG_NAME + " failed to propagate.");
557    assertEquals(0.1f, cache.getSingleFactor(), 0,
558      SINGLE_FACTOR_CONFIG_NAME + " failed to propagate.");
559    assertEquals(0.7f, cache.getMultiFactor(), 0,
560      MULTI_FACTOR_CONFIG_NAME + " failed to propagate.");
561    assertEquals(0.2f, cache.getMemoryFactor(), 0,
562      MEMORY_FACTOR_CONFIG_NAME + " failed to propagate.");
563  }
564
565  @TestTemplate
566  public void testInvalidAcceptFactorConfig() throws IOException {
567    float[] configValues = { -1f, 0.2f, 0.86f, 1.05f };
568    boolean[] expectedOutcomes = { false, false, true, false };
569    Map<String, float[]> configMappings = ImmutableMap.of(ACCEPT_FACTOR_CONFIG_NAME, configValues);
570    Configuration conf = HBaseConfiguration.create();
571    checkConfigValues(conf, configMappings, expectedOutcomes);
572  }
573
574  @TestTemplate
575  public void testInvalidMinFactorConfig() throws IOException {
576    float[] configValues = { -1f, 0f, 0.96f, 1.05f };
577    // throws due to <0, in expected range, minFactor > acceptableFactor, > 1.0
578    boolean[] expectedOutcomes = { false, true, false, false };
579    Map<String, float[]> configMappings = ImmutableMap.of(MIN_FACTOR_CONFIG_NAME, configValues);
580    Configuration conf = HBaseConfiguration.create();
581    checkConfigValues(conf, configMappings, expectedOutcomes);
582  }
583
584  @TestTemplate
585  public void testInvalidExtraFreeFactorConfig() throws IOException {
586    float[] configValues = { -1f, 0f, 0.2f, 1.05f };
587    // throws due to <0, in expected range, in expected range, config can be > 1.0
588    boolean[] expectedOutcomes = { false, true, true, true };
589    Map<String, float[]> configMappings =
590      ImmutableMap.of(EXTRA_FREE_FACTOR_CONFIG_NAME, configValues);
591    Configuration conf = HBaseConfiguration.create();
592    checkConfigValues(conf, configMappings, expectedOutcomes);
593  }
594
595  @TestTemplate
596  public void testInvalidCacheSplitFactorConfig() throws IOException {
597    float[] singleFactorConfigValues = { 0.2f, 0f, -0.2f, 1f };
598    float[] multiFactorConfigValues = { 0.4f, 0f, 1f, .05f };
599    float[] memoryFactorConfigValues = { 0.4f, 0f, 0.2f, .5f };
600    // All configs add up to 1.0 and are between 0 and 1.0, configs don't add to 1.0, configs can't
601    // be negative, configs don't add to 1.0
602    boolean[] expectedOutcomes = { true, false, false, false };
603    Map<String,
604      float[]> configMappings = ImmutableMap.of(SINGLE_FACTOR_CONFIG_NAME, singleFactorConfigValues,
605        MULTI_FACTOR_CONFIG_NAME, multiFactorConfigValues, MEMORY_FACTOR_CONFIG_NAME,
606        memoryFactorConfigValues);
607    Configuration conf = HBaseConfiguration.create();
608    checkConfigValues(conf, configMappings, expectedOutcomes);
609  }
610
611  private void checkConfigValues(Configuration conf, Map<String, float[]> configMap,
612    boolean[] expectSuccess) throws IOException {
613    Set<String> configNames = configMap.keySet();
614    for (int i = 0; i < expectSuccess.length; i++) {
615      try {
616        for (String configName : configNames) {
617          conf.setFloat(configName, configMap.get(configName)[i]);
618        }
619        BucketCache cache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
620          constructedBlockSizes, writeThreads, writerQLen, null, 100, conf);
621        assertTrue(cache.waitForCacheInitialization(10000));
622        assertTrue(expectSuccess[i], "Created BucketCache and expected it to succeed: "
623          + expectSuccess[i] + ", but it actually was: " + !expectSuccess[i]);
624      } catch (IllegalArgumentException e) {
625        assertFalse(expectSuccess[i], "Created BucketCache and expected it to succeed: "
626          + expectSuccess[i] + ", but it actually was: " + !expectSuccess[i]);
627      }
628    }
629  }
630
631  private void validateGetPartitionSize(BucketCache bucketCache, float partitionFactor,
632    float minFactor) {
633    long expectedOutput =
634      (long) Math.floor(bucketCache.getAllocator().getTotalSize() * partitionFactor * minFactor);
635    assertEquals(expectedOutput, bucketCache.getPartitionSize(partitionFactor));
636  }
637
638  @TestTemplate
639  public void testOffsetProducesPositiveOutput() {
640    // This number is picked because it produces negative output if the values isn't ensured to be
641    // positive. See HBASE-18757 for more information.
642    long testValue = 549888460800L;
643    BucketEntry bucketEntry = new BucketEntry(testValue, 10, 10, 10L, true, (entry) -> {
644      return ByteBuffAllocator.NONE;
645    }, ByteBuffAllocator.HEAP);
646    assertEquals(testValue, bucketEntry.offset());
647  }
648
649  @TestTemplate
650  public void testEvictionCount() throws InterruptedException {
651    int size = 100;
652    int length = HConstants.HFILEBLOCK_HEADER_SIZE + size;
653    ByteBuffer buf1 = ByteBuffer.allocate(size), buf2 = ByteBuffer.allocate(size);
654    HFileContext meta = new HFileContextBuilder().build();
655    ByteBuffAllocator allocator = ByteBuffAllocator.HEAP;
656    HFileBlock blockWithNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1,
657      ByteBuff.wrap(buf1), HFileBlock.FILL_HEADER, -1, 52, -1, meta, allocator);
658    HFileBlock blockWithoutNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1,
659      ByteBuff.wrap(buf2), HFileBlock.FILL_HEADER, -1, -1, -1, meta, allocator);
660
661    BlockCacheKey key = new BlockCacheKey("testEvictionCount", 0);
662    ByteBuffer actualBuffer = ByteBuffer.allocate(length);
663    ByteBuffer block1Buffer = ByteBuffer.allocate(length);
664    ByteBuffer block2Buffer = ByteBuffer.allocate(length);
665    blockWithNextBlockMetadata.serialize(block1Buffer, true);
666    blockWithoutNextBlockMetadata.serialize(block2Buffer, true);
667
668    // Add blockWithNextBlockMetadata, expect blockWithNextBlockMetadata back.
669    CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer,
670      block1Buffer);
671
672    waitUntilFlushedToBucket(cache, key);
673
674    assertEquals(0, cache.getStats().getEvictionCount());
675
676    // evict call should return 1, but then eviction count be 0
677    assertEquals(1, cache.evictBlocksByHfileName("testEvictionCount"));
678    assertEquals(0, cache.getStats().getEvictionCount());
679
680    // add back
681    key = new BlockCacheKey("testEvictionCount", 0);
682    CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer,
683      block1Buffer);
684    waitUntilFlushedToBucket(cache, key);
685
686    // should not increment
687    assertTrue(cache.evictBlock(key));
688    assertEquals(0, cache.getStats().getEvictionCount());
689
690    // add back
691    CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer,
692      block1Buffer);
693    waitUntilFlushedToBucket(cache, key);
694
695    // should finally increment eviction count
696    cache.freeSpace("testing");
697    assertEquals(1, cache.getStats().getEvictionCount());
698  }
699
700  @TestTemplate
701  public void testCacheBlockNextBlockMetadataMissing() throws Exception {
702    int size = 100;
703    int length = HConstants.HFILEBLOCK_HEADER_SIZE + size;
704    ByteBuffer buf1 = ByteBuffer.allocate(size), buf2 = ByteBuffer.allocate(size);
705    HFileContext meta = new HFileContextBuilder().build();
706    ByteBuffAllocator allocator = ByteBuffAllocator.HEAP;
707    HFileBlock blockWithNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1,
708      ByteBuff.wrap(buf1), HFileBlock.FILL_HEADER, -1, 52, -1, meta, allocator);
709    HFileBlock blockWithoutNextBlockMetadata = new HFileBlock(BlockType.DATA, size, size, -1,
710      ByteBuff.wrap(buf2), HFileBlock.FILL_HEADER, -1, -1, -1, meta, allocator);
711
712    BlockCacheKey key = new BlockCacheKey("testCacheBlockNextBlockMetadataMissing", 0);
713    ByteBuffer actualBuffer = ByteBuffer.allocate(length);
714    ByteBuffer block1Buffer = ByteBuffer.allocate(length);
715    ByteBuffer block2Buffer = ByteBuffer.allocate(length);
716    blockWithNextBlockMetadata.serialize(block1Buffer, true);
717    blockWithoutNextBlockMetadata.serialize(block2Buffer, true);
718
719    // Add blockWithNextBlockMetadata, expect blockWithNextBlockMetadata back.
720    CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer,
721      block1Buffer);
722
723    waitUntilFlushedToBucket(cache, key);
724    assertNotNull(cache.backingMap.get(key));
725    assertEquals(1, cache.backingMap.get(key).refCnt());
726    assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt());
727    assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt());
728
729    // Add blockWithoutNextBlockMetada, expect blockWithNextBlockMetadata back.
730    CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithoutNextBlockMetadata, actualBuffer,
731      block1Buffer);
732    assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt());
733    assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt());
734    assertEquals(1, cache.backingMap.get(key).refCnt());
735
736    // Clear and add blockWithoutNextBlockMetadata
737    assertTrue(cache.evictBlock(key));
738    assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt());
739    assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt());
740
741    assertNull(cache.getBlock(key, false, false, false));
742    CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithoutNextBlockMetadata, actualBuffer,
743      block2Buffer);
744
745    waitUntilFlushedToBucket(cache, key);
746    assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt());
747    assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt());
748
749    // Add blockWithNextBlockMetadata, expect blockWithNextBlockMetadata to replace.
750    CacheTestUtils.getBlockAndAssertEquals(cache, key, blockWithNextBlockMetadata, actualBuffer,
751      block1Buffer);
752
753    waitUntilFlushedToBucket(cache, key);
754    assertEquals(1, blockWithNextBlockMetadata.getBufferReadOnly().refCnt());
755    assertEquals(1, blockWithoutNextBlockMetadata.getBufferReadOnly().refCnt());
756  }
757
758  @TestTemplate
759  public void testRAMCache() {
760    int size = 100;
761    int length = HConstants.HFILEBLOCK_HEADER_SIZE + size;
762    byte[] byteArr = new byte[length];
763    ByteBuffer buf = ByteBuffer.wrap(byteArr, 0, size);
764    HFileContext meta = new HFileContextBuilder().build();
765
766    RAMCache cache = new RAMCache();
767    BlockCacheKey key1 = new BlockCacheKey("file-1", 1);
768    BlockCacheKey key2 = new BlockCacheKey("file-2", 2);
769    HFileBlock blk1 = new HFileBlock(BlockType.DATA, size, size, -1, ByteBuff.wrap(buf),
770      HFileBlock.FILL_HEADER, -1, 52, -1, meta, ByteBuffAllocator.HEAP);
771    HFileBlock blk2 = new HFileBlock(BlockType.DATA, size, size, -1, ByteBuff.wrap(buf),
772      HFileBlock.FILL_HEADER, -1, -1, -1, meta, ByteBuffAllocator.HEAP);
773    RAMQueueEntry re1 = new RAMQueueEntry(key1, blk1, 1, false, false, false);
774    RAMQueueEntry re2 = new RAMQueueEntry(key1, blk2, 1, false, false, false);
775
776    assertFalse(cache.containsKey(key1));
777    assertNull(cache.putIfAbsent(key1, re1));
778    assertEquals(2, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt());
779
780    assertNotNull(cache.putIfAbsent(key1, re2));
781    assertEquals(2, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt());
782    assertEquals(1, ((HFileBlock) re2.getData()).getBufferReadOnly().refCnt());
783
784    assertNull(cache.putIfAbsent(key2, re2));
785    assertEquals(2, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt());
786    assertEquals(2, ((HFileBlock) re2.getData()).getBufferReadOnly().refCnt());
787
788    cache.remove(key1);
789    assertEquals(1, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt());
790    assertEquals(2, ((HFileBlock) re2.getData()).getBufferReadOnly().refCnt());
791
792    cache.clear();
793    assertEquals(1, ((HFileBlock) re1.getData()).getBufferReadOnly().refCnt());
794    assertEquals(1, ((HFileBlock) re2.getData()).getBufferReadOnly().refCnt());
795  }
796
797  @TestTemplate
798  public void testFreeBlockWhenIOEngineWriteFailure() throws IOException {
799    // initialize an block.
800    int size = 100, offset = 20;
801    int length = HConstants.HFILEBLOCK_HEADER_SIZE + size;
802    ByteBuffer buf = ByteBuffer.allocate(length);
803    HFileContext meta = new HFileContextBuilder().build();
804    HFileBlock block = new HFileBlock(BlockType.DATA, size, size, -1, ByteBuff.wrap(buf),
805      HFileBlock.FILL_HEADER, offset, 52, -1, meta, ByteBuffAllocator.HEAP);
806
807    // initialize an mocked ioengine.
808    IOEngine ioEngine = Mockito.mock(IOEngine.class);
809    when(ioEngine.usesSharedMemory()).thenReturn(false);
810    // Mockito.doNothing().when(ioEngine).write(Mockito.any(ByteBuffer.class), Mockito.anyLong());
811    Mockito.doThrow(RuntimeException.class).when(ioEngine).write(Mockito.any(ByteBuffer.class),
812      Mockito.anyLong());
813    Mockito.doThrow(RuntimeException.class).when(ioEngine).write(Mockito.any(ByteBuff.class),
814      Mockito.anyLong());
815
816    // create an bucket allocator.
817    long availableSpace = 1024 * 1024 * 1024L;
818    BucketAllocator allocator = new BucketAllocator(availableSpace, null);
819
820    BlockCacheKey key = new BlockCacheKey("dummy", 1L);
821    RAMQueueEntry re = new RAMQueueEntry(key, block, 1, true, false, false);
822
823    assertEquals(0, allocator.getUsedSize());
824    try {
825      re.writeToCache(ioEngine, allocator, null, null,
826        ByteBuffer.allocate(HFileBlock.BLOCK_METADATA_SPACE), Long.MAX_VALUE);
827      fail();
828    } catch (Exception e) {
829    }
830    assertEquals(0, allocator.getUsedSize());
831  }
832
833  /**
834   * This test is for HBASE-26295, {@link BucketEntry} which is restored from a persistence file
835   * could not be freed even if corresponding {@link HFileBlock} is evicted from
836   * {@link BucketCache}.
837   */
838  @TestTemplate
839  public void testFreeBucketEntryRestoredFromFile() throws Exception {
840    BucketCache bucketCache = null;
841    try {
842      final Path dataTestDir = createAndGetTestDir();
843
844      String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache";
845      String persistencePath = dataTestDir + "/bucketNoRecycler.persistence";
846
847      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
848        constructedBlockSizes, writeThreads, writerQLen, persistencePath);
849      assertTrue(bucketCache.waitForCacheInitialization(10000));
850      long usedByteSize = bucketCache.getAllocator().getUsedSize();
851      assertEquals(0, usedByteSize);
852
853      HFileBlockPair[] hfileBlockPairs =
854        CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1);
855      // Add blocks
856      for (HFileBlockPair hfileBlockPair : hfileBlockPairs) {
857        bucketCache.cacheBlock(hfileBlockPair.getBlockName(), hfileBlockPair.getBlock());
858      }
859
860      for (HFileBlockPair hfileBlockPair : hfileBlockPairs) {
861        cacheAndWaitUntilFlushedToBucket(bucketCache, hfileBlockPair.getBlockName(),
862          hfileBlockPair.getBlock(), false);
863      }
864      usedByteSize = bucketCache.getAllocator().getUsedSize();
865      assertNotEquals(0, usedByteSize);
866      // persist cache to file
867      bucketCache.shutdown();
868      assertTrue(new File(persistencePath).exists());
869
870      // restore cache from file
871      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
872        constructedBlockSizes, writeThreads, writerQLen, persistencePath);
873      assertTrue(bucketCache.waitForCacheInitialization(10000));
874      assertEquals(usedByteSize, bucketCache.getAllocator().getUsedSize());
875
876      for (HFileBlockPair hfileBlockPair : hfileBlockPairs) {
877        BlockCacheKey blockCacheKey = hfileBlockPair.getBlockName();
878        bucketCache.evictBlock(blockCacheKey);
879      }
880      assertEquals(0, bucketCache.getAllocator().getUsedSize());
881      assertEquals(0, bucketCache.backingMap.size());
882    } finally {
883      bucketCache.shutdown();
884      HBASE_TESTING_UTILITY.cleanupTestDir();
885    }
886  }
887
888  @TestTemplate
889  public void testBlockAdditionWaitWhenCache() throws Exception {
890    BucketCache bucketCache = null;
891    try {
892      final Path dataTestDir = createAndGetTestDir();
893
894      String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache";
895      String persistencePath = dataTestDir + "/bucketNoRecycler.persistence";
896
897      Configuration config = HBASE_TESTING_UTILITY.getConfiguration();
898      config.setLong(QUEUE_ADDITION_WAIT_TIME, 1000);
899
900      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
901        constructedBlockSizes, 1, 1, persistencePath, DEFAULT_ERROR_TOLERATION_DURATION, config);
902      assertTrue(bucketCache.waitForCacheInitialization(10000));
903      long usedByteSize = bucketCache.getAllocator().getUsedSize();
904      assertEquals(0, usedByteSize);
905
906      HFileBlockPair[] hfileBlockPairs =
907        CacheTestUtils.generateHFileBlocks(constructedBlockSize, 10);
908      String[] names = CacheTestUtils.getHFileNames(hfileBlockPairs);
909      // Add blocks
910      for (HFileBlockPair hfileBlockPair : hfileBlockPairs) {
911        bucketCache.cacheBlock(hfileBlockPair.getBlockName(), hfileBlockPair.getBlock(), false,
912          true);
913      }
914
915      // Max wait for 10 seconds.
916      long timeout = 10000;
917      // Wait for blocks size to match the number of blocks.
918      while (bucketCache.backingMap.size() != 10) {
919        if (timeout <= 0) break;
920        Threads.sleep(100);
921        timeout -= 100;
922      }
923      for (HFileBlockPair hfileBlockPair : hfileBlockPairs) {
924        assertTrue(bucketCache.backingMap.containsKey(hfileBlockPair.getBlockName()));
925      }
926      usedByteSize = bucketCache.getAllocator().getUsedSize();
927      assertNotEquals(0, usedByteSize);
928      // persist cache to file
929      bucketCache.shutdown();
930      assertTrue(new File(persistencePath).exists());
931
932      // restore cache from file
933      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
934        constructedBlockSizes, writeThreads, writerQLen, persistencePath);
935      assertTrue(bucketCache.waitForCacheInitialization(10000));
936      assertEquals(usedByteSize, bucketCache.getAllocator().getUsedSize());
937      BlockCacheKey[] newKeys = CacheTestUtils.regenerateKeys(hfileBlockPairs, names);
938      for (BlockCacheKey key : newKeys) {
939        bucketCache.evictBlock(key);
940      }
941      assertEquals(0, bucketCache.getAllocator().getUsedSize());
942      assertEquals(0, bucketCache.backingMap.size());
943    } finally {
944      if (bucketCache != null) {
945        bucketCache.shutdown();
946      }
947      HBASE_TESTING_UTILITY.cleanupTestDir();
948    }
949  }
950
951  @TestTemplate
952  public void testOnConfigurationChange() throws Exception {
953    BucketCache bucketCache = null;
954    try {
955      final Path dataTestDir = createAndGetTestDir();
956
957      String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache";
958
959      Configuration config = HBASE_TESTING_UTILITY.getConfiguration();
960
961      bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
962        constructedBlockSizes, 1, 1, null, DEFAULT_ERROR_TOLERATION_DURATION, config);
963
964      assertTrue(bucketCache.waitForCacheInitialization(10000));
965
966      config.setFloat(ACCEPT_FACTOR_CONFIG_NAME, 0.9f);
967      config.setFloat(MIN_FACTOR_CONFIG_NAME, 0.8f);
968      config.setFloat(EXTRA_FREE_FACTOR_CONFIG_NAME, 0.15f);
969      config.setFloat(SINGLE_FACTOR_CONFIG_NAME, 0.2f);
970      config.setFloat(MULTI_FACTOR_CONFIG_NAME, 0.6f);
971      config.setFloat(MEMORY_FACTOR_CONFIG_NAME, 0.2f);
972      config.setLong(QUEUE_ADDITION_WAIT_TIME, 100);
973      config.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, 500);
974      config.setLong(BACKING_MAP_PERSISTENCE_CHUNK_SIZE, 1000);
975
976      bucketCache.onConfigurationChange(config);
977
978      assertEquals(0.9f, bucketCache.getAcceptableFactor(), 0.01);
979      assertEquals(0.8f, bucketCache.getMinFactor(), 0.01);
980      assertEquals(0.15f, bucketCache.getExtraFreeFactor(), 0.01);
981      assertEquals(0.2f, bucketCache.getSingleFactor(), 0.01);
982      assertEquals(0.6f, bucketCache.getMultiFactor(), 0.01);
983      assertEquals(0.2f, bucketCache.getMemoryFactor(), 0.01);
984      assertEquals(100L, bucketCache.getQueueAdditionWaitTime());
985      assertEquals(500L, bucketCache.getBucketcachePersistInterval());
986      assertEquals(1000L, bucketCache.getPersistenceChunkSize());
987
988    } finally {
989      if (bucketCache != null) {
990        bucketCache.shutdown();
991      }
992      HBASE_TESTING_UTILITY.cleanupTestDir();
993    }
994  }
995
996  @TestTemplate
997  public void testNotifyFileCachingCompletedSuccess() throws Exception {
998    BucketCache bucketCache = null;
999    try {
1000      Path filePath =
1001        new Path(HBASE_TESTING_UTILITY.getDataTestDir(), "testNotifyFileCachingCompletedSuccess");
1002      bucketCache = testNotifyFileCachingCompletedForTenBlocks(filePath, 10, false);
1003      if (bucketCache.getStats().getFailedInserts() > 0) {
1004        LOG.info("There were {} fail inserts, "
1005          + "will assert if total blocks in backingMap equals (10 - failInserts) "
1006          + "and file isn't listed as fully cached.", bucketCache.getStats().getFailedInserts());
1007        assertEquals(10 - bucketCache.getStats().getFailedInserts(), bucketCache.backingMap.size());
1008        assertFalse(bucketCache.fullyCachedFiles.containsKey(filePath.getName()));
1009      } else {
1010        assertTrue(bucketCache.fullyCachedFiles.containsKey(filePath.getName()));
1011      }
1012    } finally {
1013      if (bucketCache != null) {
1014        bucketCache.shutdown();
1015      }
1016      HBASE_TESTING_UTILITY.cleanupTestDir();
1017    }
1018  }
1019
1020  @TestTemplate
1021  public void testNotifyFileCachingCompletedForEncodedDataSuccess() throws Exception {
1022    BucketCache bucketCache = null;
1023    try {
1024      Path filePath = new Path(HBASE_TESTING_UTILITY.getDataTestDir(),
1025        "testNotifyFileCachingCompletedForEncodedDataSuccess");
1026      bucketCache = testNotifyFileCachingCompletedForTenBlocks(filePath, 10, true);
1027      if (bucketCache.getStats().getFailedInserts() > 0) {
1028        LOG.info("There were {} fail inserts, "
1029          + "will assert if total blocks in backingMap equals (10 - failInserts) "
1030          + "and file isn't listed as fully cached.", bucketCache.getStats().getFailedInserts());
1031        assertEquals(10 - bucketCache.getStats().getFailedInserts(), bucketCache.backingMap.size());
1032        assertFalse(bucketCache.fullyCachedFiles.containsKey(filePath.getName()));
1033      } else {
1034        assertTrue(bucketCache.fullyCachedFiles.containsKey(filePath.getName()));
1035      }
1036    } finally {
1037      if (bucketCache != null) {
1038        bucketCache.shutdown();
1039      }
1040      HBASE_TESTING_UTILITY.cleanupTestDir();
1041    }
1042  }
1043
1044  @TestTemplate
1045  public void testNotifyFileCachingCompletedNotAllCached() throws Exception {
1046    BucketCache bucketCache = null;
1047    try {
1048      Path filePath = new Path(HBASE_TESTING_UTILITY.getDataTestDir(),
1049        "testNotifyFileCachingCompletedNotAllCached");
1050      // Deliberately passing more blocks than we have created to test that
1051      // notifyFileCachingCompleted will not consider the file fully cached
1052      bucketCache = testNotifyFileCachingCompletedForTenBlocks(filePath, 12, false);
1053      assertFalse(bucketCache.fullyCachedFiles.containsKey(filePath.getName()));
1054    } finally {
1055      if (bucketCache != null) {
1056        bucketCache.shutdown();
1057      }
1058      HBASE_TESTING_UTILITY.cleanupTestDir();
1059    }
1060  }
1061
1062  private BucketCache testNotifyFileCachingCompletedForTenBlocks(Path filePath,
1063    int totalBlocksToCheck, boolean encoded) throws Exception {
1064    final Path dataTestDir = createAndGetTestDir();
1065    String ioEngineName = "file:" + dataTestDir + "/bucketNoRecycler.cache";
1066    BucketCache bucketCache = new BucketCache(ioEngineName, capacitySize, constructedBlockSize,
1067      constructedBlockSizes, 1, 1, null);
1068    assertTrue(bucketCache.waitForCacheInitialization(10000));
1069    long usedByteSize = bucketCache.getAllocator().getUsedSize();
1070    assertEquals(0, usedByteSize);
1071    HFileBlockPair[] hfileBlockPairs =
1072      CacheTestUtils.generateBlocksForPath(constructedBlockSize, 10, filePath, encoded);
1073    // Add blocks
1074    for (HFileBlockPair hfileBlockPair : hfileBlockPairs) {
1075      bucketCache.cacheBlock(hfileBlockPair.getBlockName(), hfileBlockPair.getBlock(), false, true);
1076    }
1077    bucketCache.notifyFileCachingCompleted(filePath, totalBlocksToCheck, totalBlocksToCheck,
1078      totalBlocksToCheck * constructedBlockSize);
1079    return bucketCache;
1080  }
1081
1082  @TestTemplate
1083  public void testEvictOrphansOutOfGracePeriod() throws Exception {
1084    BucketCache bucketCache = testEvictOrphans(0);
1085    assertEquals(10, bucketCache.getBackingMap().size());
1086    assertEquals(0, bucketCache.blocksByHFile.stream()
1087      .filter(key -> key.getHfileName().equals("testEvictOrphans-orphan")).count());
1088  }
1089
1090  @TestTemplate
1091  public void testEvictOrphansWithinGracePeriod() throws Exception {
1092    BucketCache bucketCache = testEvictOrphans(60 * 60 * 1000L);
1093    assertEquals(18, bucketCache.getBackingMap().size());
1094    assertTrue(bucketCache.blocksByHFile.stream()
1095      .filter(key -> key.getHfileName().equals("testEvictOrphans-orphan")).count() > 0);
1096  }
1097
1098  private BucketCache testEvictOrphans(long orphanEvictionGracePeriod) throws Exception {
1099    Path validFile = new Path(HBASE_TESTING_UTILITY.getDataTestDir(), "testEvictOrphans-valid");
1100    Path orphanFile = new Path(HBASE_TESTING_UTILITY.getDataTestDir(), "testEvictOrphans-orphan");
1101    Map<String, HRegion> onlineRegions = new HashMap<>();
1102    List<HStore> stores = new ArrayList<>();
1103    Collection<HStoreFile> storeFiles = new ArrayList<>();
1104    HRegion mockedRegion = mock(HRegion.class);
1105    HStore mockedStore = mock(HStore.class);
1106    HStoreFile mockedStoreFile = mock(HStoreFile.class);
1107    when(mockedStoreFile.getPath()).thenReturn(validFile);
1108    storeFiles.add(mockedStoreFile);
1109    when(mockedStore.getStorefiles()).thenReturn(storeFiles);
1110    stores.add(mockedStore);
1111    when(mockedRegion.getStores()).thenReturn(stores);
1112    onlineRegions.put("mocked_region", mockedRegion);
1113    HBASE_TESTING_UTILITY.getConfiguration().setDouble(MIN_FACTOR_CONFIG_NAME, 0.99);
1114    HBASE_TESTING_UTILITY.getConfiguration().setDouble(ACCEPT_FACTOR_CONFIG_NAME, 1);
1115    HBASE_TESTING_UTILITY.getConfiguration().setDouble(EXTRA_FREE_FACTOR_CONFIG_NAME, 0.01);
1116    HBASE_TESTING_UTILITY.getConfiguration().setLong(BLOCK_ORPHAN_GRACE_PERIOD,
1117      orphanEvictionGracePeriod);
1118    BucketCache bucketCache = new BucketCache(ioEngineName, (constructedBlockSize + 1024) * 21,
1119      constructedBlockSize, new int[] { constructedBlockSize + 1024 }, 1, 1, null, 60 * 1000,
1120      HBASE_TESTING_UTILITY.getConfiguration(), onlineRegions);
1121    HFileBlockPair[] validBlockPairs =
1122      CacheTestUtils.generateBlocksForPath(constructedBlockSize, 10, validFile, false);
1123    HFileBlockPair[] orphanBlockPairs =
1124      CacheTestUtils.generateBlocksForPath(constructedBlockSize, 10, orphanFile, false);
1125    for (HFileBlockPair pair : validBlockPairs) {
1126      bucketCache.cacheBlockWithWait(pair.getBlockName(), pair.getBlock(), false, true);
1127    }
1128    waitUntilAllFlushedToBucket(bucketCache);
1129    assertEquals(10, bucketCache.getBackingMap().size());
1130    bucketCache.freeSpace("test");
1131    assertEquals(10, bucketCache.getBackingMap().size());
1132    for (HFileBlockPair pair : orphanBlockPairs) {
1133      bucketCache.cacheBlockWithWait(pair.getBlockName(), pair.getBlock(), false, true);
1134    }
1135    waitUntilAllFlushedToBucket(bucketCache);
1136    assertEquals(20, bucketCache.getBackingMap().size());
1137    bucketCache.freeSpace("test");
1138    return bucketCache;
1139  }
1140
1141  @TestTemplate
1142  public void testBlockPriority() throws Exception {
1143    HFileBlockPair block = CacheTestUtils.generateHFileBlocks(BLOCK_SIZE, 1)[0];
1144    cacheAndWaitUntilFlushedToBucket(cache, block.getBlockName(), block.getBlock(), true);
1145    assertEquals(cache.backingMap.get(block.getBlockName()).getPriority(), BlockPriority.SINGLE);
1146    cache.getBlock(block.getBlockName(), true, false, true);
1147    assertEquals(cache.backingMap.get(block.getBlockName()).getPriority(), BlockPriority.MULTI);
1148  }
1149
1150  @TestTemplate
1151  public void testIOTimePerHitReturnsZeroWhenNoHits()
1152    throws NoSuchFieldException, IllegalAccessException {
1153    CacheStats cacheStats = cache.getStats();
1154    assertTrue(cacheStats instanceof BucketCacheStats);
1155    BucketCacheStats bucketCacheStats = (BucketCacheStats) cacheStats;
1156
1157    Field field = BucketCacheStats.class.getDeclaredField("ioHitCount");
1158    field.setAccessible(true);
1159    LongAdder ioHitCount = (LongAdder) field.get(bucketCacheStats);
1160
1161    assertEquals(0, ioHitCount.sum());
1162    double ioTimePerHit = bucketCacheStats.getIOTimePerHit();
1163    assertEquals(0, ioTimePerHit, 0.0);
1164  }
1165}