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.mob;
019
020import static org.junit.jupiter.api.Assertions.assertEquals;
021import static org.junit.jupiter.api.Assertions.assertTrue;
022import static org.junit.jupiter.api.Assertions.fail;
023
024import java.io.IOException;
025import java.util.Arrays;
026import java.util.List;
027import java.util.stream.Collectors;
028import java.util.stream.Stream;
029import org.apache.hadoop.conf.Configuration;
030import org.apache.hadoop.fs.FileStatus;
031import org.apache.hadoop.fs.FileSystem;
032import org.apache.hadoop.fs.Path;
033import org.apache.hadoop.hbase.HBaseParameterizedTestTemplate;
034import org.apache.hadoop.hbase.HBaseTestingUtil;
035import org.apache.hadoop.hbase.ServerName;
036import org.apache.hadoop.hbase.TableName;
037import org.apache.hadoop.hbase.client.Admin;
038import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
039import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
040import org.apache.hadoop.hbase.client.CompactionState;
041import org.apache.hadoop.hbase.client.Put;
042import org.apache.hadoop.hbase.client.Result;
043import org.apache.hadoop.hbase.client.ResultScanner;
044import org.apache.hadoop.hbase.client.Table;
045import org.apache.hadoop.hbase.client.TableDescriptor;
046import org.apache.hadoop.hbase.regionserver.storefiletracker.StoreFileTrackerFactory;
047import org.apache.hadoop.hbase.testclassification.LargeTests;
048import org.apache.hadoop.hbase.util.Bytes;
049import org.apache.hadoop.hbase.util.RegionSplitter;
050import org.junit.jupiter.api.AfterEach;
051import org.junit.jupiter.api.BeforeEach;
052import org.junit.jupiter.api.Tag;
053import org.junit.jupiter.api.TestInfo;
054import org.junit.jupiter.api.TestTemplate;
055import org.junit.jupiter.params.provider.Arguments;
056import org.slf4j.Logger;
057import org.slf4j.LoggerFactory;
058
059/**
060 * Mob file compaction base test. 1. Enables batch mode for regular MOB compaction, Sets batch size
061 * to 7 regions. (Optional) 2. Disables periodic MOB compactions, sets minimum age to archive to 10
062 * sec 3. Creates MOB table with 20 regions 4. Loads MOB data (randomized keys, 1000 rows), flushes
063 * data. 5. Repeats 4. two more times 6. Verifies that we have 20 *3 = 60 mob files (equals to
064 * number of regions x 3) 7. Runs major MOB compaction. 8. Verifies that number of MOB files in a
065 * mob directory is 20 x4 = 80 9. Waits for a period of time larger than minimum age to archive 10.
066 * Runs Mob cleaner chore 11 Verifies that number of MOB files in a mob directory is 20. 12 Runs
067 * scanner and checks all 3 * 1000 rows.
068 */
069@Tag(LargeTests.TAG)
070@HBaseParameterizedTestTemplate(name = "{index}: useFileBasedSFT={0}")
071public class TestMobCompactionWithDefaults {
072  private static final Logger LOG = LoggerFactory.getLogger(TestMobCompactionWithDefaults.class);
073
074  protected HBaseTestingUtil HTU;
075  protected static Configuration conf;
076  protected static long minAgeToArchive = 10000;
077
078  protected final static String famStr = "f1";
079  protected final static byte[] fam = Bytes.toBytes(famStr);
080  protected final static byte[] qualifier = Bytes.toBytes("q1");
081  protected final static long mobLen = 10;
082  protected final static byte[] mobVal = Bytes
083    .toBytes("01234567890123456789012345678901234567890123456789012345678901234567890123456789");
084
085  private String testMethodName;
086  protected TableDescriptor tableDescriptor;
087  private ColumnFamilyDescriptor familyDescriptor;
088  protected Admin admin;
089  protected TableName table = null;
090  protected int numRegions = 20;
091  protected int rows = 1000;
092
093  protected Boolean useFileBasedSFT;
094
095  public TestMobCompactionWithDefaults(Boolean useFileBasedSFT) {
096    this.useFileBasedSFT = useFileBasedSFT;
097  }
098
099  public static Stream<Arguments> parameters() {
100    return Stream.of(false, true).map(Arguments::of);
101  }
102
103  protected void htuStart() throws Exception {
104    HTU = new HBaseTestingUtil();
105    conf = HTU.getConfiguration();
106    conf.setInt("hfile.format.version", 3);
107    // Disable automatic MOB compaction
108    conf.setLong(MobConstants.MOB_COMPACTION_CHORE_PERIOD, 0);
109    // Disable automatic MOB file cleaner chore
110    conf.setLong(MobConstants.MOB_CLEANER_PERIOD, 0);
111    // Set minimum age to archive to 10 sec
112    conf.setLong(MobConstants.MIN_AGE_TO_ARCHIVE_KEY, minAgeToArchive);
113    // Set compacted file discharger interval to a half minAgeToArchive
114    conf.setLong("hbase.hfile.compaction.discharger.interval", minAgeToArchive / 2);
115    conf.setBoolean("hbase.regionserver.compaction.enabled", false);
116    if (useFileBasedSFT) {
117      conf.set(StoreFileTrackerFactory.TRACKER_IMPL,
118        "org.apache.hadoop.hbase.regionserver.storefiletracker.FileBasedStoreFileTracker");
119    }
120    additonalConfigSetup();
121    HTU.startMiniCluster();
122  }
123
124  protected void additonalConfigSetup() {
125  }
126
127  @BeforeEach
128  public void setUp(TestInfo testInfo) throws Exception {
129    testMethodName = testInfo.getTestMethod().get().getName()
130      + testInfo.getDisplayName().replaceAll("[:= ]", "_").replaceAll("_+", "_").trim();
131    htuStart();
132    admin = HTU.getAdmin();
133    familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(fam).setMobEnabled(true)
134      .setMobThreshold(mobLen).setMaxVersions(1).build();
135    tableDescriptor = HTU.createModifyableTableDescriptor(TestMobUtils.getTableName(testMethodName))
136      .setColumnFamily(familyDescriptor).build();
137    RegionSplitter.UniformSplit splitAlgo = new RegionSplitter.UniformSplit();
138    byte[][] splitKeys = splitAlgo.split(numRegions);
139    table = HTU.createTable(tableDescriptor, splitKeys).getName();
140  }
141
142  private void loadData(TableName tableName, int num) {
143    LOG.info("Started loading {} rows into {}", num, tableName);
144    try (final Table table = HTU.getConnection().getTable(tableName)) {
145      for (int i = 0; i < num; i++) {
146        byte[] key = new byte[32];
147        Bytes.random(key);
148        Put p = new Put(key);
149        p.addColumn(fam, qualifier, mobVal);
150        table.put(p);
151      }
152      admin.flush(tableName);
153      LOG.info("Finished loading {} rows into {}", num, tableName);
154    } catch (Exception e) {
155      LOG.error("MOB file compaction chore test FAILED", e);
156      fail("MOB file compaction chore test FAILED");
157    }
158  }
159
160  @AfterEach
161  public void tearDown() throws Exception {
162    admin.disableTable(tableDescriptor.getTableName());
163    admin.deleteTable(tableDescriptor.getTableName());
164    HTU.shutdownMiniCluster();
165  }
166
167  @TestTemplate
168  public void baseTestMobFileCompaction() throws InterruptedException, IOException {
169    LOG.info("MOB compaction " + description() + " started");
170    loadAndFlushThreeTimes(rows, table, famStr);
171    mobCompact(tableDescriptor, familyDescriptor);
172    assertEquals(numRegions * 4, getNumberOfMobFiles(table, famStr),
173      "Should have 4 MOB files per region due to 3xflush + compaction.");
174    cleanupAndVerifyCounts(table, famStr, 3 * rows);
175    LOG.info("MOB compaction " + description() + " finished OK");
176  }
177
178  @TestTemplate
179  public void testMobFileCompactionAfterSnapshotClone() throws InterruptedException, IOException {
180    final TableName clone = TableName.valueOf(TestMobUtils.getTableName(testMethodName) + "-clone");
181    LOG.info("MOB compaction of cloned snapshot, " + description() + " started");
182    loadAndFlushThreeTimes(rows, table, famStr);
183    LOG.debug("Taking snapshot and cloning table {}", table);
184    admin.snapshot(TestMobUtils.getTableName(testMethodName), table);
185    admin.cloneSnapshot(TestMobUtils.getTableName(testMethodName), clone);
186    assertEquals(3 * numRegions, getNumberOfMobFiles(clone, famStr),
187      "Should have 3 hlinks per region in MOB area from snapshot clone");
188    mobCompact(admin.getDescriptor(clone), familyDescriptor);
189    assertEquals(4 * numRegions, getNumberOfMobFiles(clone, famStr),
190      "Should have 3 hlinks + 1 MOB file per region due to clone + compact");
191    cleanupAndVerifyCounts(clone, famStr, 3 * rows);
192    LOG.info("MOB compaction of cloned snapshot, " + description() + " finished OK");
193  }
194
195  @TestTemplate
196  public void testMobFileCompactionAfterSnapshotCloneAndFlush()
197    throws InterruptedException, IOException {
198    final TableName clone = TableName.valueOf(TestMobUtils.getTableName(testMethodName) + "-clone");
199    LOG.info("MOB compaction of cloned snapshot after flush, " + description() + " started");
200    loadAndFlushThreeTimes(rows, table, famStr);
201    LOG.debug("Taking snapshot and cloning table {}", table);
202    admin.snapshot(TestMobUtils.getTableName(testMethodName), table);
203    admin.cloneSnapshot(TestMobUtils.getTableName(testMethodName), clone);
204    assertEquals(3 * numRegions, getNumberOfMobFiles(clone, famStr),
205      "Should have 3 hlinks per region in MOB area from snapshot clone");
206    loadAndFlushThreeTimes(rows, clone, famStr);
207    mobCompact(admin.getDescriptor(clone), familyDescriptor);
208    assertEquals(7 * numRegions, getNumberOfMobFiles(clone, famStr),
209      "Should have 7 MOB file per region due to clone + 3xflush + compact");
210    cleanupAndVerifyCounts(clone, famStr, 6 * rows);
211    LOG.info("MOB compaction of cloned snapshot w flush, " + description() + " finished OK");
212  }
213
214  protected void loadAndFlushThreeTimes(int rows, TableName table, String family)
215    throws IOException {
216    final long start = getNumberOfMobFiles(table, family);
217    // Load and flush data 3 times
218    loadData(table, rows);
219    loadData(table, rows);
220    loadData(table, rows);
221    assertEquals(start + numRegions * 3, getNumberOfMobFiles(table, family),
222      "Should have 3 more mob files per region from flushing.");
223  }
224
225  protected String description() {
226    return "regular mode";
227  }
228
229  protected void enableCompactions() throws IOException {
230    final List<String> serverList =
231      admin.getRegionServers().stream().map(sn -> sn.getServerName()).collect(Collectors.toList());
232    admin.compactionSwitch(true, serverList);
233  }
234
235  protected void disableCompactions() throws IOException {
236    final List<String> serverList =
237      admin.getRegionServers().stream().map(sn -> sn.getServerName()).collect(Collectors.toList());
238    admin.compactionSwitch(false, serverList);
239  }
240
241  /**
242   * compact the given table and return once it is done. should presume compactions are disabled
243   * when called. should ensure compactions are disabled before returning.
244   */
245  protected void mobCompact(TableDescriptor tableDescriptor,
246    ColumnFamilyDescriptor familyDescriptor) throws IOException, InterruptedException {
247    LOG.debug("Major compact MOB table " + tableDescriptor.getTableName());
248    enableCompactions();
249    mobCompactImpl(tableDescriptor, familyDescriptor);
250    waitUntilCompactionIsComplete(tableDescriptor.getTableName());
251    disableCompactions();
252  }
253
254  /**
255   * Call the API for compaction specific to the test set. should not wait for compactions to
256   * finish. may assume compactions are enabled when called.
257   */
258  protected void mobCompactImpl(TableDescriptor tableDescriptor,
259    ColumnFamilyDescriptor familyDescriptor) throws IOException, InterruptedException {
260    admin.majorCompact(tableDescriptor.getTableName(), familyDescriptor.getName());
261  }
262
263  protected void waitUntilCompactionIsComplete(TableName table)
264    throws IOException, InterruptedException {
265    CompactionState state = admin.getCompactionState(table);
266    while (state != CompactionState.NONE) {
267      LOG.debug("Waiting for compaction on {} to complete. current state {}", table, state);
268      Thread.sleep(100);
269      state = admin.getCompactionState(table);
270    }
271    LOG.debug("done waiting for compaction on {}", table);
272  }
273
274  protected void cleanupAndVerifyCounts(TableName table, String family, int rows)
275    throws InterruptedException, IOException {
276    // We have guarantee, that compacted file discharger will run during this pause
277    // because it has interval less than this wait time
278    LOG.info("Waiting for {}ms", minAgeToArchive + 1000);
279
280    Thread.sleep(minAgeToArchive + 1000);
281    LOG.info("Cleaning up MOB files");
282
283    // run cleaner chore on each RS
284    for (ServerName sn : admin.getRegionServers()) {
285      HTU.getMiniHBaseCluster().getRegionServer(sn).getRSMobFileCleanerChore().chore();
286    }
287
288    assertEquals(numRegions, getNumberOfMobFiles(table, family),
289      "After cleaning, we should have 1 MOB file per region based on size.");
290
291    LOG.debug("checking count of rows");
292    long scanned = scanTable(table);
293    assertEquals(rows, scanned, "Got the wrong number of rows in table " + table + " cf " + family);
294
295  }
296
297  protected long getNumberOfMobFiles(TableName tableName, String family) throws IOException {
298    FileSystem fs = FileSystem.get(conf);
299    Path dir = MobUtils.getMobFamilyPath(conf, tableName, family);
300    FileStatus[] stat = fs.listStatus(dir);
301    for (FileStatus st : stat) {
302      LOG.debug("MOB Directory content: {}", st.getPath());
303    }
304    LOG.debug("MOB Directory content total files: {}", stat.length);
305
306    return stat.length;
307  }
308
309  protected long scanTable(TableName tableName) {
310    try (final Table table = HTU.getConnection().getTable(tableName);
311      final ResultScanner scanner = table.getScanner(fam)) {
312      Result result;
313      long counter = 0;
314      while ((result = scanner.next()) != null) {
315        assertTrue(Arrays.equals(result.getValue(fam, qualifier), mobVal));
316        counter++;
317      }
318      return counter;
319    } catch (Exception e) {
320      LOG.error("MOB file compaction test FAILED", e);
321      if (HTU != null) {
322        fail(e.getMessage());
323      } else {
324        System.exit(-1);
325      }
326    }
327    return 0;
328  }
329}