Source code

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.junit.Assert.assertEquals;
021import static org.junit.Assert.assertNotEquals;
022import static org.junit.Assert.assertTrue;
023
024import java.io.BufferedWriter;
025import java.io.FileOutputStream;
026import java.io.OutputStreamWriter;
027import java.nio.file.FileSystems;
028import java.nio.file.Files;
029import java.nio.file.attribute.FileTime;
030import java.time.Instant;
031import java.util.Arrays;
032import org.apache.hadoop.fs.Path;
033import org.apache.hadoop.hbase.HBaseClassTestRule;
034import org.apache.hadoop.hbase.HBaseTestingUtility;
035import org.apache.hadoop.hbase.io.hfile.BlockCacheKey;
036import org.apache.hadoop.hbase.io.hfile.CacheTestUtils;
037import org.apache.hadoop.hbase.io.hfile.Cacheable;
038import org.apache.hadoop.hbase.testclassification.SmallTests;
039import org.junit.ClassRule;
040import org.junit.Test;
041import org.junit.experimental.categories.Category;
042import org.junit.runner.RunWith;
043import org.junit.runners.Parameterized;
044
045/**
046 * Basic test for check file's integrity before start BucketCache in fileIOEngine
047 */
048@RunWith(Parameterized.class)
049@Category(SmallTests.class)
050public class TestVerifyBucketCacheFile {
051  @ClassRule
052  public static final HBaseClassTestRule CLASS_RULE =
053    HBaseClassTestRule.forClass(TestVerifyBucketCacheFile.class);
054
055  @Parameterized.Parameters(name = "{index}: blockSize={0}, bucketSizes={1}")
056  public static Iterable<Object[]> data() {
057    return Arrays.asList(new Object[][] { { 8192, null },
058      { 16 * 1024,
059        new int[] { 2 * 1024 + 1024, 4 * 1024 + 1024, 8 * 1024 + 1024, 16 * 1024 + 1024,
060          28 * 1024 + 1024, 32 * 1024 + 1024, 64 * 1024 + 1024, 96 * 1024 + 1024,
061          128 * 1024 + 1024 } } });
062  }
063
064  @Parameterized.Parameter(0)
065  public int constructedBlockSize;
066
067  @Parameterized.Parameter(1)
068  public int[] constructedBlockSizes;
069
070  final long capacitySize = 32 * 1024 * 1024;
071  final int writeThreads = BucketCache.DEFAULT_WRITER_THREADS;
072  final int writerQLen = BucketCache.DEFAULT_WRITER_QUEUE_ITEMS;
073
074  /**
075   * Test cache file or persistence file does not exist whether BucketCache starts normally (1)
076   * Start BucketCache and add some blocks, then shutdown BucketCache and persist cache to file.
077   * Restart BucketCache and it can restore cache from file. (2) Delete bucket cache file after
078   * shutdown BucketCache. Restart BucketCache and it can't restore cache from file, the cache file
079   * and persistence file would be deleted before BucketCache start normally. (3) Delete persistence
080   * file after shutdown BucketCache. Restart BucketCache and it can't restore cache from file, the
081   * cache file and persistence file would be deleted before BucketCache start normally.
082   * @throws Exception the exception
083   */
084  @Test
085  public void testRetrieveFromFile() throws Exception {
086    HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
087    Path testDir = TEST_UTIL.getDataTestDir();
088    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
089
090    BucketCache bucketCache =
091      new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize,
092        constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence");
093    long usedSize = bucketCache.getAllocator().getUsedSize();
094    assertEquals(0, usedSize);
095    CacheTestUtils.HFileBlockPair[] blocks =
096      CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1);
097    // Add blocks
098    for (CacheTestUtils.HFileBlockPair block : blocks) {
099      cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock());
100    }
101    usedSize = bucketCache.getAllocator().getUsedSize();
102    assertNotEquals(0, usedSize);
103    // 1.persist cache to file
104    bucketCache.shutdown();
105    // restore cache from file
106    bucketCache =
107      new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize,
108        constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence");
109    assertEquals(usedSize, bucketCache.getAllocator().getUsedSize());
110    // persist cache to file
111    bucketCache.shutdown();
112
113    // 2.delete bucket cache file
114    final java.nio.file.Path cacheFile =
115      FileSystems.getDefault().getPath(testDir.toString(), "bucket.cache");
116    assertTrue(Files.deleteIfExists(cacheFile));
117    // can't restore cache from file
118    bucketCache =
119      new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize,
120        constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence");
121    assertEquals(0, bucketCache.getAllocator().getUsedSize());
122    assertEquals(0, bucketCache.backingMap.size());
123    // Add blocks
124    for (CacheTestUtils.HFileBlockPair block : blocks) {
125      cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock());
126    }
127    usedSize = bucketCache.getAllocator().getUsedSize();
128    assertNotEquals(0, usedSize);
129    // persist cache to file
130    bucketCache.shutdown();
131
132    // 3.delete backingMap persistence file
133    final java.nio.file.Path mapFile =
134      FileSystems.getDefault().getPath(testDir.toString(), "bucket.persistence");
135    assertTrue(Files.deleteIfExists(mapFile));
136    // can't restore cache from file
137    bucketCache =
138      new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize,
139        constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence");
140    assertEquals(0, bucketCache.getAllocator().getUsedSize());
141    assertEquals(0, bucketCache.backingMap.size());
142
143    TEST_UTIL.cleanupTestDir();
144  }
145
146  /**
147   * Test whether BucketCache is started normally after modifying the cache file. Start BucketCache
148   * and add some blocks, then shutdown BucketCache and persist cache to file. Restart BucketCache
149   * after modify cache file's data, and it can't restore cache from file, the cache file and
150   * persistence file would be deleted before BucketCache start normally.
151   * @throws Exception the exception
152   */
153  @Test
154  public void testModifiedBucketCacheFileData() throws Exception {
155    HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
156    Path testDir = TEST_UTIL.getDataTestDir();
157    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
158
159    BucketCache bucketCache =
160      new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize,
161        constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence");
162    long usedSize = bucketCache.getAllocator().getUsedSize();
163    assertEquals(0, usedSize);
164
165    CacheTestUtils.HFileBlockPair[] blocks =
166      CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1);
167    // Add blocks
168    for (CacheTestUtils.HFileBlockPair block : blocks) {
169      cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock());
170    }
171    usedSize = bucketCache.getAllocator().getUsedSize();
172    assertNotEquals(0, usedSize);
173    // persist cache to file
174    bucketCache.shutdown();
175
176    // modified bucket cache file
177    String file = testDir + "/bucket.cache";
178    try (BufferedWriter out =
179      new BufferedWriter(new OutputStreamWriter(new FileOutputStream(file, false)))) {
180      out.write("test bucket cache");
181    }
182    // can't restore cache from file
183    bucketCache =
184      new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize,
185        constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence");
186    assertEquals(0, bucketCache.getAllocator().getUsedSize());
187    assertEquals(0, bucketCache.backingMap.size());
188
189    TEST_UTIL.cleanupTestDir();
190  }
191
192  /**
193   * Test whether BucketCache is started normally after modifying the cache file's last modified
194   * time. First Start BucketCache and add some blocks, then shutdown BucketCache and persist cache
195   * to file. Then Restart BucketCache after modify cache file's last modified time, and it can't
196   * restore cache from file, the cache file and persistence file would be deleted before
197   * BucketCache start normally.
198   * @throws Exception the exception
199   */
200  @Test
201  public void testModifiedBucketCacheFileTime() throws Exception {
202    HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
203    Path testDir = TEST_UTIL.getDataTestDir();
204    TEST_UTIL.getTestFileSystem().mkdirs(testDir);
205
206    BucketCache bucketCache =
207      new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize,
208        constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence");
209    long usedSize = bucketCache.getAllocator().getUsedSize();
210    assertEquals(0, usedSize);
211
212    CacheTestUtils.HFileBlockPair[] blocks =
213      CacheTestUtils.generateHFileBlocks(constructedBlockSize, 1);
214    // Add blocks
215    for (CacheTestUtils.HFileBlockPair block : blocks) {
216      cacheAndWaitUntilFlushedToBucket(bucketCache, block.getBlockName(), block.getBlock());
217    }
218    usedSize = bucketCache.getAllocator().getUsedSize();
219    assertNotEquals(0, usedSize);
220    // persist cache to file
221    bucketCache.shutdown();
222
223    // modified bucket cache file LastModifiedTime
224    final java.nio.file.Path file =
225      FileSystems.getDefault().getPath(testDir.toString(), "bucket.cache");
226    Files.setLastModifiedTime(file, FileTime.from(Instant.now().plusMillis(1_000)));
227    // can't restore cache from file
228    bucketCache =
229      new BucketCache("file:" + testDir + "/bucket.cache", capacitySize, constructedBlockSize,
230        constructedBlockSizes, writeThreads, writerQLen, testDir + "/bucket.persistence");
231    assertEquals(0, bucketCache.getAllocator().getUsedSize());
232    assertEquals(0, bucketCache.backingMap.size());
233
234    TEST_UTIL.cleanupTestDir();
235  }
236
237  private void waitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey)
238    throws InterruptedException {
239    while (!cache.backingMap.containsKey(cacheKey) || cache.ramCache.containsKey(cacheKey)) {
240      Thread.sleep(100);
241    }
242  }
243
244  // BucketCache.cacheBlock is async, it first adds block to ramCache and writeQueue, then writer
245  // threads will flush it to the bucket and put reference entry in backingMap.
246  private void cacheAndWaitUntilFlushedToBucket(BucketCache cache, BlockCacheKey cacheKey,
247    Cacheable block) throws InterruptedException {
248    cache.cacheBlock(cacheKey, block);
249    waitUntilFlushedToBucket(cache, cacheKey);
250  }
251}