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 java.util.Map;
032import org.apache.hadoop.conf.Configuration;
033import org.apache.hadoop.fs.Path;
034import org.apache.hadoop.hbase.HBaseConfiguration;
035import org.apache.hadoop.hbase.HBaseTestingUtil;
036import org.apache.hadoop.hbase.Waiter;
037import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
038import org.apache.hadoop.hbase.io.hfile.BlockPriority;
039import org.apache.hadoop.hbase.io.hfile.BlockType;
040import org.apache.hadoop.hbase.io.hfile.CacheTestUtils;
041import org.apache.hadoop.hbase.io.hfile.Cacheable;
042import org.apache.hadoop.hbase.testclassification.RegionServerTests;
043import org.apache.hadoop.hbase.testclassification.SmallTests;
044import org.junit.jupiter.api.Tag;
045import org.junit.jupiter.api.Test;
046import org.junit.jupiter.params.ParameterizedTest;
047import org.junit.jupiter.params.provider.ValueSource;
048
049/**
050 * Basic test for check file's integrity before start BucketCache in fileIOEngine
051 */
052@Tag(SmallTests.TAG)
053@Tag(RegionServerTests.TAG)
054public class TestRecoveryPersistentBucketCache {
055
056  final long capacitySize = 32 * 1024 * 1024;
057  final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS;
058  final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS;
059
060  @Test
061  public void testBucketCacheRecovery() throws Exception {
062    HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
063    Path testDir = TEST_UTIL.getDataTestDir();
064    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
065    Configuration conf = HBaseConfiguration.create();
066    // Disables the persister thread by setting its interval to MAX_VALUE
067    conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE);
068    int[] bucketSizes = new int[] { 8 * 1024 + 1024 };
069    BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
070      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
071      DEFAULT_ERROR_TOLERATION_DURATION, conf);
072    assertTrue(bucketCache.waitForCacheInitialization(1000));
073    assertTrue(
074      bucketCache.isCacheInitialized("testBucketCacheRecovery") && bucketCache.isCacheEnabled());
075
076    CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 4);
077    String[] names = CacheTestUtils.getHFileNames(blocks);
078
079    CacheTestUtils.HFileBlockPair[] smallerBlocks = CacheTestUtils.generateHFileBlocks(4096, 1);
080    String[] smallerNames = CacheTestUtils.getHFileNames(smallerBlocks);
081    // Add four blocks
082    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock());
083    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock());
084    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName(), blocks[2].getBlock());
085    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[3].getBlockName(), blocks[3].getBlock());
086    // saves the current state of the cache
087    bucketCache.persistToFile();
088    // evicts the 4th block
089    bucketCache.evictBlock(blocks[3].getBlockName());
090    // now adds a 5th block to bucket cache. This block is half the size of the previous
091    // blocks, and it will be added in the same offset of the previous evicted block.
092    // This overwrites part of the 4th block. Because we persisted only up to the
093    // 4th block addition, recovery would try to read the whole 4th block, but the cached time
094    // validation will fail, and we'll recover only the first three blocks
095    cacheAndWaitUntilFlushedToBucket(bucketCache, smallerBlocks[0].getBlockName(),
096      smallerBlocks[0].getBlock());
097
098    // Creates new bucket cache instance without persisting to file after evicting 4th block
099    // and caching 5th block. Here the cache file has the first three blocks, followed by the
100    // 5th block and the second half of 4th block (we evicted 4th block, freeing up its
101    // offset in the cache, then added 5th block which is half the size of other blocks, so it's
102    // going to override the first half of the 4th block in the cache). That's fine because
103    // the in-memory backing map has the right blocks and related offsets. However, the
104    // persistent map file only has information about the first four blocks. We validate the
105    // cache time recorded in the back map against the block data in the cache. This is recorded
106    // in the cache as the first 8 bytes of a block, so the 4th block had its first 8 blocks
107    // now overridden by the 5th block, causing this check to fail and removal of
108    // the 4th block from the backing map.
109    BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
110      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
111      DEFAULT_ERROR_TOLERATION_DURATION, conf);
112    assertTrue(newBucketCache.waitForCacheInitialization(1000));
113    BlockCacheKey[] newKeys = CacheTestUtils.regenerateKeys(blocks, names);
114    BlockCacheKey[] newKeysSmaller = CacheTestUtils.regenerateKeys(smallerBlocks, smallerNames);
115    // The new bucket cache would have only the first three blocks. Although we have persisted the
116    // the cache state when it had the first four blocks, the 4th block was evicted and then we
117    // added a 5th block, which overrides part of the 4th block in the cache. This would cause a
118    // checksum failure for this block offset, when we try to read from the cache, and we would
119    // consider that block as invalid and its offset available in the cache.
120    assertNull(newBucketCache.getBlock(newKeys[3], false, false, false));
121    assertNull(newBucketCache.getBlock(newKeysSmaller[0], false, false, false));
122    assertEquals(blocks[0].getBlock(), newBucketCache.getBlock(newKeys[0], false, false, false));
123    assertEquals(blocks[1].getBlock(), newBucketCache.getBlock(newKeys[1], false, false, false));
124    assertEquals(blocks[2].getBlock(), newBucketCache.getBlock(newKeys[2], false, false, false));
125    TEST_UTIL.cleanupTestDir();
126  }
127
128  @Test
129  public void testBucketCacheEvictByHFileAfterRecovery() throws Exception {
130    HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
131    Path testDir = TEST_UTIL.getDataTestDir();
132    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
133    Configuration conf = HBaseConfiguration.create();
134    // Disables the persister thread by setting its interval to MAX_VALUE
135    conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE);
136    int[] bucketSizes = new int[] { 8 * 1024 + 1024 };
137    BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
138      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
139      DEFAULT_ERROR_TOLERATION_DURATION, conf);
140    assertTrue(bucketCache.waitForCacheInitialization(10000));
141
142    CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 4);
143
144    // Add four blocks
145    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock());
146    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock());
147    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName(), blocks[2].getBlock());
148    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[3].getBlockName(), blocks[3].getBlock());
149
150    String firstFileName = blocks[0].getBlockName().getHfileName();
151
152    // saves the current state of the cache
153    bucketCache.persistToFile();
154
155    BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
156      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
157      DEFAULT_ERROR_TOLERATION_DURATION, conf);
158    assertTrue(newBucketCache.waitForCacheInitialization(10000));
159    assertEquals(4, newBucketCache.backingMap.size());
160
161    newBucketCache.evictBlocksByHfileName(firstFileName);
162    assertEquals(3, newBucketCache.backingMap.size());
163    TEST_UTIL.cleanupTestDir();
164  }
165
166  @Test
167  public void testBlockPriorityAfterRecovery() throws Exception {
168    HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
169    Path testDir = TEST_UTIL.getDataTestDir();
170    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
171    Configuration conf = HBaseConfiguration.create();
172    // Disables the persister thread by setting its interval to MAX_VALUE
173    conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE);
174    int[] bucketSizes = new int[] { 8 * 1024 + 1024 };
175    BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
176      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
177      DEFAULT_ERROR_TOLERATION_DURATION, conf);
178    assertTrue(bucketCache.waitForCacheInitialization(10000));
179
180    CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 3);
181    String[] names = CacheTestUtils.getHFileNames(blocks);
182
183    // first block is only cached, so it stays SINGLE
184    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock());
185    // second block is read after being cached, so it is promoted to MULTI
186    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock());
187    bucketCache.getBlock(blocks[1].getBlockName(), false, false, false);
188    // third block is an in-memory block, so it is MEMORY
189    bucketCache.cacheBlock(blocks[2].getBlockName(), blocks[2].getBlock(), true);
190    waitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName());
191
192    assertEquals(BlockPriority.SINGLE,
193      bucketCache.backingMap.get(blocks[0].getBlockName()).getPriority());
194    assertEquals(BlockPriority.MULTI,
195      bucketCache.backingMap.get(blocks[1].getBlockName()).getPriority());
196    assertEquals(BlockPriority.MEMORY,
197      bucketCache.backingMap.get(blocks[2].getBlockName()).getPriority());
198
199    // saves the current state of the cache
200    bucketCache.persistToFile();
201
202    BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
203      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
204      DEFAULT_ERROR_TOLERATION_DURATION, conf);
205    assertTrue(newBucketCache.waitForCacheInitialization(10000));
206    BlockCacheKey[] newKeys = CacheTestUtils.regenerateKeys(blocks, names);
207
208    // the recovered blocks should keep the priority they had before persisting
209    assertEquals(BlockPriority.SINGLE, newBucketCache.backingMap.get(newKeys[0]).getPriority());
210    assertEquals(BlockPriority.MULTI, newBucketCache.backingMap.get(newKeys[1]).getPriority());
211    assertEquals(BlockPriority.MEMORY, newBucketCache.backingMap.get(newKeys[2]).getPriority());
212    TEST_UTIL.cleanupTestDir();
213  }
214
215  @ParameterizedTest
216  @ValueSource(booleans = { false, true })
217  public void testArchivedFlagAfterRecovery(boolean archived) throws Exception {
218    HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
219    Path testDir = TEST_UTIL.getDataTestDir();
220    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
221    Configuration conf = HBaseConfiguration.create();
222    conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE);
223    int[] bucketSizes = new int[] { 8 * 1024 + 1024 };
224    CacheTestUtils.HFileBlockPair block = CacheTestUtils.generateHFileBlocks(8192, 1)[0];
225    BlockCacheKey key =
226      new BlockCacheKey("test-file", "cf", "test-region", 0, true, BlockType.DATA, archived);
227    BucketCache bucketCache = null;
228    BucketCache recoveredCache = null;
229    try {
230      bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, 8192,
231        bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
232        DEFAULT_ERROR_TOLERATION_DURATION, conf);
233      assertTrue(bucketCache.waitForCacheInitialization(10000));
234      cacheAndWaitUntilFlushedToBucket(bucketCache, key, block.getBlock());
235      long cachedSize = bucketCache.backingMap.get(key).getLength();
236      Map<String, Long> expectedRegionCachedSize =
237        archived ? Map.of() : Map.of(key.getRegionName(), cachedSize);
238      assertEquals(expectedRegionCachedSize, bucketCache.getRegionCachedInfo().get());
239
240      bucketCache.shutdown();
241      bucketCache = null;
242      recoveredCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, 8192,
243        bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
244        DEFAULT_ERROR_TOLERATION_DURATION, conf);
245      assertTrue(recoveredCache.waitForCacheInitialization(10000));
246      assertEquals(1, recoveredCache.backingMap.size());
247      // Inspect the recovered key: BlockCacheKey.equals does not compare the archived flag.
248      assertEquals(archived, recoveredCache.backingMap.keySet().iterator().next().isArchived());
249      assertEquals(expectedRegionCachedSize, recoveredCache.getRegionCachedInfo().get());
250    } finally {
251      if (recoveredCache != null) {
252        recoveredCache.shutdown();
253      }
254      if (bucketCache != null) {
255        bucketCache.shutdown();
256      }
257      block.getBlock().release();
258      TEST_UTIL.cleanupTestDir();
259    }
260  }
261
262  @Test
263  public void testValidateCacheInitialization() throws Exception {
264    HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
265    Path testDir = TEST_UTIL.getDataTestDir();
266    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
267    Configuration conf = HBaseConfiguration.create();
268    // Disables the persister thread by setting its interval to MAX_VALUE
269    conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE);
270    int[] bucketSizes = new int[] { 8 * 1024 + 1024 };
271    BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
272      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
273      DEFAULT_ERROR_TOLERATION_DURATION, conf);
274    assertTrue(bucketCache.waitForCacheInitialization(10000));
275
276    CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 4);
277
278    // Add four blocks
279    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock());
280    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock());
281    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName(), blocks[2].getBlock());
282    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[3].getBlockName(), blocks[3].getBlock());
283    // saves the current state of the cache
284    bucketCache.persistToFile();
285
286    BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", capacitySize,
287      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
288      DEFAULT_ERROR_TOLERATION_DURATION, conf);
289    assertTrue(newBucketCache.waitForCacheInitialization(10000));
290
291    // Set the state of bucket cache to INITIALIZING
292    newBucketCache.setCacheState(BucketCache.CacheState.INITIALIZING);
293
294    // Validate that zero values are returned for the cache being initialized.
295    assertEquals(0, newBucketCache.acceptableSize());
296    assertEquals(0, newBucketCache.getPartitionSize(1));
297    assertEquals(0, newBucketCache.getFreeSize());
298    assertEquals(0, newBucketCache.getCurrentSize());
299    assertEquals(false, newBucketCache.blockFitsIntoTheCache(blocks[0].getBlock()).get());
300
301    newBucketCache.setCacheState(BucketCache.CacheState.ENABLED);
302
303    // Validate that non-zero values are returned for enabled cache
304    assertTrue(newBucketCache.acceptableSize() > 0);
305    assertTrue(newBucketCache.getPartitionSize(1) > 0);
306    assertTrue(newBucketCache.getFreeSize() > 0);
307    assertTrue(newBucketCache.getCurrentSize() > 0);
308    assertTrue(newBucketCache.blockFitsIntoTheCache(blocks[0].getBlock()).get());
309
310    TEST_UTIL.cleanupTestDir();
311  }
312
313  @Test
314  public void testBucketCacheRecoveryWithAllocationInconsistencies() throws Exception {
315    HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
316    Path testDir = TEST_UTIL.getDataTestDir();
317    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
318    Configuration conf = HBaseConfiguration.create();
319    // Disables the persister thread by setting its interval to MAX_VALUE
320    conf.setLong(BUCKETCACHE_PERSIST_INTERVAL_KEY, Long.MAX_VALUE);
321    conf.setDouble(MIN_FACTOR_CONFIG_NAME, 0.99);
322    conf.setDouble(ACCEPT_FACTOR_CONFIG_NAME, 1);
323    conf.setDouble(EXTRA_FREE_FACTOR_CONFIG_NAME, 0.01);
324    int[] bucketSizes = new int[] { 8 * 1024 + 1024 };
325    BucketCache bucketCache = new BucketCache("file:" + testDir + "/bucket.cache", 36 * 1024, 8192,
326      bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
327      DEFAULT_ERROR_TOLERATION_DURATION, conf);
328    assertTrue(bucketCache.waitForCacheInitialization(1000));
329    assertTrue(
330      bucketCache.isCacheInitialized("testBucketCacheRecovery") && bucketCache.isCacheEnabled());
331
332    CacheTestUtils.HFileBlockPair[] blocks = CacheTestUtils.generateHFileBlocks(8192, 5);
333    String[] names = CacheTestUtils.getHFileNames(blocks);
334
335    // Add four blocks
336    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[0].getBlockName(), blocks[0].getBlock());
337    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[1].getBlockName(), blocks[1].getBlock());
338    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[2].getBlockName(), blocks[2].getBlock());
339    cacheAndWaitUntilFlushedToBucket(bucketCache, blocks[3].getBlockName(), blocks[3].getBlock());
340
341    // creates a entry for a 5th block with the same cache offset of the 1st block. Just add it
342    // straight to the backingMap, bypassing caching, in order to fabricate an inconsistency
343    BucketEntry bucketEntry =
344      new BucketEntry(bucketCache.backingMap.get(blocks[0].getBlockName()).offset(),
345        blocks[4].getBlock().getSerializedLength(), blocks[4].getBlock().getOnDiskSizeWithHeader(),
346        0, false, bucketCache::createRecycler, blocks[4].getBlock().getByteBuffAllocator());
347    bucketEntry.setDeserializerReference(blocks[4].getBlock().getDeserializer());
348    bucketCache.getBackingMap().put(blocks[4].getBlockName(), bucketEntry);
349
350    // saves the current state of the cache: 5 blocks in the map, but we only have cached 4. The
351    // 5th block has same cache offset as the first
352    bucketCache.persistToFile();
353
354    BucketCache newBucketCache = new BucketCache("file:" + testDir + "/bucket.cache", 36 * 1024,
355      8192, bucketSizes, writeThreads, writerQLen, testDir + "/bucket.persistence",
356      DEFAULT_ERROR_TOLERATION_DURATION, conf);
357    // backingMapValidated is set inside retrieveFromFile(), but isCacheEnabled() (which
358    // getBlock() checks) is set to true only in the finally block that follows. Waiting on
359    // backingMapValidated alone leaves a race window; await both conditions together.
360    await().atMost(Duration.ofSeconds(30)).pollInterval(Duration.ofMillis(10)).until(
361      () -> newBucketCache.getBackingMapValidated().get() && newBucketCache.isCacheEnabled());
362
363    BlockCacheKey[] newKeys = CacheTestUtils.regenerateKeys(blocks, names);
364
365    assertNull(newBucketCache.getBlock(newKeys[4], false, false, false));
366    // The backing map entry with key blocks[0].getBlockName() for the may point to a valid entry
367    // or null based on different ordering of the keys in the backing map.
368    // Hence, skipping the check for that key.
369    assertEquals(blocks[1].getBlock(), newBucketCache.getBlock(newKeys[1], false, false, false));
370    assertEquals(blocks[2].getBlock(), newBucketCache.getBlock(newKeys[2], false, false, false));
371    assertEquals(blocks[3].getBlock(), newBucketCache.getBlock(newKeys[3], false, false, false));
372    assertEquals(4, newBucketCache.backingMap.size());
373    TEST_UTIL.cleanupTestDir();
374  }
375
376  private void waitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey)
377    throws InterruptedException {
378    Waiter.waitFor(HBaseConfiguration.create(), 12000,
379      () -> (cache.backingMap.containsKey(cacheKey) && !cache.ramCache.containsKey(cacheKey)));
380  }
381
382  // BucketCache.cacheBlock is async, it first adds block to ramCache and writeQueue, then writer
383  // threads will flush it to the bucket and put reference entry in backingMap.
384  private void cacheAndWaitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey,
385    Cacheable block) throws InterruptedException {
386    cache.cacheBlock(cacheKey, block);
387    waitUntilFlushedToBucket(cache, cacheKey);
388  }
389
390}