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.backup;
019
020import static org.apache.hadoop.hbase.IntegrationTestingUtility.createPreSplitLoadTestTable;
021import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONF_CONTINUOUS_BACKUP_WAL_DIR;
022import static org.junit.jupiter.api.Assertions.assertEquals;
023import static org.junit.jupiter.api.Assertions.assertFalse;
024import static org.junit.jupiter.api.Assertions.assertTrue;
025import static org.junit.jupiter.api.Assertions.fail;
026
027import java.io.IOException;
028import java.nio.charset.Charset;
029import java.util.ArrayList;
030import java.util.List;
031import java.util.Map;
032import java.util.Set;
033import org.apache.hadoop.fs.FileStatus;
034import org.apache.hadoop.fs.FileSystem;
035import org.apache.hadoop.fs.LocatedFileStatus;
036import org.apache.hadoop.fs.Path;
037import org.apache.hadoop.fs.RemoteIterator;
038import org.apache.hadoop.hbase.Cell;
039import org.apache.hadoop.hbase.HBaseTestingUtil;
040import org.apache.hadoop.hbase.IntegrationTestBase;
041import org.apache.hadoop.hbase.IntegrationTestingUtility;
042import org.apache.hadoop.hbase.ServerName;
043import org.apache.hadoop.hbase.TableName;
044import org.apache.hadoop.hbase.backup.impl.BackupAdminImpl;
045import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
046import org.apache.hadoop.hbase.backup.util.BackupUtils;
047import org.apache.hadoop.hbase.chaos.actions.RestartRandomRsExceptMetaAction;
048import org.apache.hadoop.hbase.chaos.monkies.PolicyBasedChaosMonkey;
049import org.apache.hadoop.hbase.chaos.policies.PeriodicRandomActionPolicy;
050import org.apache.hadoop.hbase.chaos.policies.Policy;
051import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
052import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
053import org.apache.hadoop.hbase.client.Connection;
054import org.apache.hadoop.hbase.client.Result;
055import org.apache.hadoop.hbase.client.ResultScanner;
056import org.apache.hadoop.hbase.client.Scan;
057import org.apache.hadoop.hbase.client.Table;
058import org.apache.hadoop.hbase.client.TableDescriptor;
059import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
060import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
061import org.junit.jupiter.api.AfterEach;
062import org.junit.jupiter.api.Assertions;
063import org.slf4j.Logger;
064import org.slf4j.LoggerFactory;
065
066import org.apache.hbase.thirdparty.com.google.common.base.MoreObjects;
067import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
068import org.apache.hbase.thirdparty.com.google.common.util.concurrent.Uninterruptibles;
069import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
070
071/**
072 * An abstract base class that is used to run backup, restore, and delete integration tests. This
073 * class performs both full backups and incremental backups. Both continuous backup and
074 * non-continuous backup test cases are supported. The number of incremental backups performed
075 * depends on the number of iterations defined by the user. The class performs the backup/restore in
076 * a separate thread, where one thread is created per table. The number of tables is user-defined,
077 * along with other various configurations.
078 */
079public abstract class IntegrationTestBackupRestoreBase extends IntegrationTestBase {
080  protected static final Logger LOG =
081    LoggerFactory.getLogger(IntegrationTestBackupRestoreBase.class);
082  protected static final String NUMBER_OF_TABLES_KEY = "num_tables";
083  protected static final String COLUMN_NAME = "f";
084  protected static final String REGION_COUNT_KEY = "regions_per_rs";
085  protected static final String REGIONSERVER_COUNT_KEY = "region_servers";
086  protected static final String ROWS_PER_ITERATION_KEY = "rows_in_iteration";
087  protected static final String NUM_ITERATIONS_KEY = "num_iterations";
088  protected static final int DEFAULT_REGION_COUNT = 10;
089  protected static final int DEFAULT_REGIONSERVER_COUNT = 5;
090  protected static final int DEFAULT_NUMBER_OF_TABLES = 1;
091  protected static final int DEFAULT_NUM_ITERATIONS = 10;
092  protected static final int DEFAULT_ROWS_IN_ITERATION = 10000;
093  protected static final String SLEEP_TIME_KEY = "sleeptime";
094  // Short default interval because tests don't run very long.
095  protected static final long SLEEP_TIME_DEFAULT = 50000L;
096  protected static String DEFAULT_BACKUP_ROOT_DIR = "backupIT";
097
098  // These test parameters can be configured using a Configuration object or via the command line.
099  protected int rowsInIteration;
100  protected int regionsCountPerServer;
101  protected int regionServerCount;
102  protected int numIterations;
103  protected int numTables;
104  protected long sleepTime;
105
106  protected static TableName[] tableNames;
107  protected FileSystem fs;
108  protected String backupRootDir;
109
110  /*
111   * This class is used to run the backup and restore thread(s). Throwing an exception in this
112   * thread will not cause the test to fail, so the purpose of this class is to both kick off the
113   * backup and restore, as well as record any exceptions that occur so they can be thrown in the
114   * main thread.
115   */
116  protected class BackupAndRestoreThread implements Runnable {
117    private final TableName table;
118    private final boolean isContinuousBackupEnabled;
119    private Throwable throwable;
120
121    public BackupAndRestoreThread(TableName table, boolean isContinuousBackupEnabled) {
122      this.table = table;
123      this.isContinuousBackupEnabled = isContinuousBackupEnabled;
124      this.throwable = null;
125    }
126
127    public Throwable getThrowable() {
128      return this.throwable;
129    }
130
131    @Override
132    public void run() {
133      try {
134        LOG.info("Running backup and restore test for {} in thread {}", this.table,
135          Thread.currentThread());
136        runTestSingle(this.table, isContinuousBackupEnabled);
137      } catch (Throwable t) {
138        LOG.error(
139          "An error occurred in thread {} when performing a backup and restore with table {}: ",
140          Thread.currentThread().getName(), this.table.getNameAsString(), t);
141        this.throwable = t;
142      }
143    }
144  }
145
146  @AfterEach
147  public void tearDown() throws IOException {
148    LOG.info("Cleaning up after test.");
149    if (util.isDistributedCluster()) {
150      deleteTablesIfAny();
151      LOG.info("Cleaning up after test. Deleted tables");
152      cleanUpBackupDir();
153    }
154    LOG.info("Restoring cluster.");
155    util.restoreCluster();
156    LOG.info("Cluster restored.");
157  }
158
159  @Override
160  public void setUpMonkey() throws Exception {
161    Policy p =
162      new PeriodicRandomActionPolicy(sleepTime, new RestartRandomRsExceptMetaAction(sleepTime));
163    this.monkey = new PolicyBasedChaosMonkey(util, p);
164    startMonkey();
165  }
166
167  private void deleteTablesIfAny() throws IOException {
168    for (TableName table : tableNames) {
169      util.deleteTableIfAny(table);
170    }
171  }
172
173  protected void createTables(String tableBaseName) throws Exception {
174    tableNames = new TableName[numTables];
175    for (int i = 0; i < numTables; i++) {
176      tableNames[i] = TableName.valueOf(tableBaseName + ".table." + i);
177    }
178    for (TableName table : tableNames) {
179      createTable(table);
180    }
181  }
182
183  /**
184   * Creates a directory specified by backupWALDir and sets this directory to
185   * CONF_CONTINUOUS_BACKUP_WAL_DIR in the configuration.
186   */
187  protected void createAndSetBackupWalDir() throws IOException {
188    Path root = util.getDataTestDirOnTestFS();
189    Path backupWalDir = new Path(root, "backupWALDir");
190    FileSystem fs = FileSystem.get(conf);
191    fs.mkdirs(backupWalDir);
192    conf.set(CONF_CONTINUOUS_BACKUP_WAL_DIR, backupWalDir.toString());
193    LOG.info(
194      "The continuous backup WAL directory has been created and set in the configuration to: {}",
195      backupWalDir);
196  }
197
198  private void cleanUpBackupDir() throws IOException {
199    FileSystem fs = FileSystem.get(util.getConfiguration());
200    fs.delete(new Path(backupRootDir), true);
201  }
202
203  /**
204   * This is the main driver method used by tests that extend this abstract base class. This method
205   * starts one backup and restore thread per table.
206   * @param isContinuousBackupEnabled Boolean flag used to specify if the backups should have
207   *                                  continuous backup enabled.
208   */
209  protected void runTestMulti(boolean isContinuousBackupEnabled) {
210    Thread[] workers = new Thread[numTables];
211    BackupAndRestoreThread[] backupAndRestoreThreads = new BackupAndRestoreThread[numTables];
212    for (int i = 0; i < numTables; i++) {
213      final TableName table = tableNames[i];
214      BackupAndRestoreThread backupAndRestoreThread =
215        new BackupAndRestoreThread(table, isContinuousBackupEnabled);
216      backupAndRestoreThreads[i] = backupAndRestoreThread;
217      workers[i] = new Thread(backupAndRestoreThread);
218      workers[i].start();
219    }
220    // Wait for all workers to finish and check for errors
221    Throwable error = null;
222    Throwable threadThrowable;
223    for (int i = 0; i < numTables; i++) {
224      Uninterruptibles.joinUninterruptibly(workers[i]);
225      threadThrowable = backupAndRestoreThreads[i].getThrowable();
226      if (threadThrowable == null) {
227        continue;
228      }
229      if (error == null) {
230        error = threadThrowable;
231      } else {
232        error.addSuppressed(threadThrowable);
233      }
234    }
235    // Throw any found errors after all threads have completed
236    if (error != null) {
237      throw new AssertionError("An error occurred in a backup and restore thread", error);
238    }
239    LOG.info("IT backup & restore finished");
240  }
241
242  /**
243   * This method is what performs the actual backup, restore, merge, and delete operations. This
244   * method is run in a separate thread. It first performs a full backup. After, it iteratively
245   * performs a series of incremental backups and restores. Later, it deletes the backups.
246   * @param tableName                 The table the backups are performed on
247   * @param isContinuousBackupEnabled Boolean flag used to indicate if the backups should have
248   *                                  continuous backup enabled.
249   */
250  private void runTestSingle(TableName tableName, boolean isContinuousBackupEnabled)
251    throws IOException, InterruptedException {
252    String enabledOrDisabled = isContinuousBackupEnabled ? "enabled" : "disabled";
253    List<String> backupIds = new ArrayList<>();
254
255    try (Connection conn = util.getConnection(); BackupAdmin client = new BackupAdminImpl(conn);
256      Table tableConn = conn.getTable(tableName)) {
257      loadData(tableName, rowsInIteration);
258
259      // First create a full backup for the table
260      LOG.info("Creating full backup image for {} with continuous backup {}", tableName,
261        enabledOrDisabled);
262      List<TableName> tables = Lists.newArrayList(tableName);
263      BackupRequest.Builder builder = new BackupRequest.Builder();
264      BackupRequest request = builder.withBackupType(BackupType.FULL).withTableList(tables)
265        .withTargetRootDir(backupRootDir).withContinuousBackupEnabled(isContinuousBackupEnabled)
266        .build();
267
268      String fullBackupId = backup(request, client, backupIds);
269      LOG.info("Created full backup with ID: {}", fullBackupId);
270
271      verifySnapshotExists(tableName, fullBackupId);
272
273      // Run full backup verifications specific to continuous backup
274      if (isContinuousBackupEnabled) {
275        BackupTestUtil.verifyReplicationPeerSubscription(util, tableName);
276        Path backupWALs = verifyWALsDirectoryExists();
277        Path walPartitionDir = verifyWALPartitionDirExists(backupWALs);
278        verifyBackupWALFiles(walPartitionDir);
279      }
280
281      // Now continue with incremental backups
282      String incrementalBackupId;
283      for (int count = 1; count <= numIterations; count++) {
284        LOG.info("{} - Starting incremental backup iteration {} of {} for {}",
285          Thread.currentThread().getName(), count, numIterations, tableName);
286        loadData(tableName, rowsInIteration);
287        if (isContinuousBackupEnabled) {
288          long latestPutTimestamp = getLatestPutTimestamp(tableConn);
289          waitForCheckpointTimestampsToUpdate(conn, latestPutTimestamp, tableName);
290        }
291
292        // Do incremental backup
293        LOG.info("Creating incremental backup number {} with continuous backup {} for {}", count,
294          enabledOrDisabled, tableName);
295        builder = new BackupRequest.Builder();
296        request = builder.withBackupType(BackupType.INCREMENTAL).withTableList(tables)
297          .withTargetRootDir(backupRootDir).withContinuousBackupEnabled(isContinuousBackupEnabled)
298          .build();
299        incrementalBackupId = backup(request, client, backupIds);
300        LOG.info("Created incremental backup with ID: {}", incrementalBackupId);
301
302        // Restore table using backup taken "two backups ago"
303        // On the first iteration, this backup will be the full backup
304        String previousBackupId = backupIds.get(backupIds.size() - 2);
305        if (previousBackupId.equals(fullBackupId)) {
306          LOG.info("Restoring {} using original full backup with ID: {}", tableName,
307            previousBackupId);
308        } else {
309          LOG.info("Restoring {} using second most recent incremental backup with ID: {}",
310            tableName, previousBackupId);
311        }
312        restoreTableAndVerifyRowCount(conn, client, tableName, previousBackupId,
313          (long) rowsInIteration * count);
314
315        // Restore table using the most recently created incremental backup
316        LOG.info("Restoring {} using most recent incremental backup with ID: {}", tableName,
317          incrementalBackupId);
318        restoreTableAndVerifyRowCount(conn, client, tableName, incrementalBackupId,
319          (long) rowsInIteration * (count + 1));
320        LOG.info("{} - Finished incremental backup iteration {} of {} for {}",
321          Thread.currentThread().getName(), count, numIterations, tableName);
322      }
323
324      // Now merge all incremental and restore
325      String[] incrementalBackupIds = getAllIncrementalBackupIds(backupIds);
326      merge(incrementalBackupIds, client);
327      verifyBackupExistenceAfterMerge(backupIds);
328      removeNonexistentBackups(backupIds);
329      // Restore the last incremental backup
330      incrementalBackupId = incrementalBackupIds[incrementalBackupIds.length - 1];
331      // restore incremental backup for table, with overwrite
332      TableName[] tablesToRestoreFrom = new TableName[] { tableName };
333      restore(createRestoreRequest(backupRootDir, incrementalBackupId, false, tablesToRestoreFrom,
334        null, true), client);
335      Table hTable = conn.getTable(tableName);
336      Assertions.assertEquals(rowsInIteration * (numIterations + 1),
337        HBaseTestingUtil.countRows(hTable));
338      hTable.close();
339
340      // Create another incremental backup to show it can be deleted on its own
341      backup(request, client, backupIds);
342      deleteMostRecentIncrementalBackup(backupIds, client);
343      // The full backup and the second most recent incremental backup should still exist
344      assertEquals(2, backupIds.size());
345      verifyAllBackupsExist(backupIds);
346      // Delete the full backup, which should also automatically delete any incremental backups that
347      // depend on it
348      LOG.info("Deleting full backup: {}. This will also delete any remaining incremental backups",
349        fullBackupId);
350      delete(new String[] { fullBackupId }, client);
351      verifyNoBackupsExist(backupIds);
352    }
353  }
354
355  private void createTable(TableName tableName) throws Exception {
356    long startTime, endTime;
357
358    TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableName);
359
360    TableDescriptor desc = builder.build();
361    ColumnFamilyDescriptorBuilder cbuilder =
362      ColumnFamilyDescriptorBuilder.newBuilder(COLUMN_NAME.getBytes(Charset.defaultCharset()));
363    ColumnFamilyDescriptor[] columns = new ColumnFamilyDescriptor[] { cbuilder.build() };
364    LOG.info("Creating table {} with {} splits.", tableName,
365      regionsCountPerServer * regionServerCount);
366    startTime = EnvironmentEdgeManager.currentTime();
367    createPreSplitLoadTestTable(util.getConfiguration(), desc, columns, regionsCountPerServer);
368    util.waitTableAvailable(tableName);
369    endTime = EnvironmentEdgeManager.currentTime();
370    LOG.info("Pre-split table created successfully in {}ms.", (endTime - startTime));
371  }
372
373  private void loadData(TableName table, int numRows) throws IOException {
374    Connection conn = util.getConnection();
375    // #0- insert some data to a table
376    Table t1 = conn.getTable(table);
377    util.loadRandomRows(t1, new byte[] { 'f' }, 100, numRows);
378    // flush table
379    conn.getAdmin().flush(TableName.valueOf(table.getName()));
380  }
381
382  private String backup(BackupRequest request, BackupAdmin client, List<String> backupIds)
383    throws IOException {
384    String backupId = client.backupTables(request);
385    assertTrue(checkSucceeded(backupId));
386    verifyBackupExists(backupId);
387    backupIds.add(backupId);
388    return backupId;
389  }
390
391  private void restore(RestoreRequest request, BackupAdmin client) throws IOException {
392    client.restore(request);
393  }
394
395  private void merge(String[] backupIds, BackupAdmin client) throws IOException {
396    client.mergeBackups(backupIds);
397  }
398
399  private void delete(String[] backupIds, BackupAdmin client) throws IOException {
400    client.deleteBackups(backupIds);
401  }
402
403  /**
404   * Verifies a snapshot's "data.manifest" file exists after a full backup has been performed for a
405   * table. The "data.manifest" file's path will look like the following:
406   * .../backupRootDir/backup_1760572298945/default/TABLE_NAME/.hbase-snapshot/
407   * snapshot_1760572306407_default_TABLE_NAME/data.manifest
408   */
409  private void verifySnapshotExists(TableName tableName, String backupId) throws IOException {
410    RemoteIterator<LocatedFileStatus> fileStatusIterator =
411      fs.listFiles(new Path(backupRootDir, backupId), true);
412    Path dataManifestPath = null;
413    while (fileStatusIterator.hasNext()) {
414      LocatedFileStatus fileStatus = fileStatusIterator.next();
415      if (fileStatus.getPath().getName().endsWith("data.manifest")) {
416        dataManifestPath = fileStatus.getPath();
417        LOG.info("Found snapshot manifest for table '{}' at: {}", tableName, dataManifestPath);
418      }
419    }
420
421    if (dataManifestPath == null) {
422      fail("Could not find snapshot data manifest for table '" + tableName + "'");
423    }
424  }
425
426  /** Verifies the .../backupWALDir/WALs directory exists and returns its Path */
427  private Path verifyWALsDirectoryExists() throws IOException {
428    String backupWALDir = conf.get(CONF_CONTINUOUS_BACKUP_WAL_DIR);
429    Path backupWALs = new Path(backupWALDir, "WALs");
430    assertTrue(fs.exists(backupWALs),
431      "There should be a WALs directory inside of the backup WAL directory at: " + backupWALDir);
432    return backupWALs;
433  }
434
435  /**
436   * Waits for a WAL partition directory to exist inside the backup WAL directory. The directory
437   * should be something like: .../backupWALDir/WALs/2025-10-17. The directory's existence is either
438   * eventually asserted, or an assertion error is thrown if it does not exist past the wait
439   * deadline. This verification is to be used for full backups with continuous backup enabled.
440   * @param backupWALs The directory that should contain the partition directory. i.e.
441   *                   .../backupWALDir/WALs
442   * @return The Path to the WAL partition directory
443   */
444  private Path verifyWALPartitionDirExists(Path backupWALs)
445    throws IOException, InterruptedException {
446    long currentTimeMs = System.currentTimeMillis();
447    String currentDateUTC = BackupUtils.formatToDateString(currentTimeMs);
448    Path walPartitionDir = new Path(backupWALs, currentDateUTC);
449    int waitTimeSec = 30;
450    while (true) {
451      try {
452        assertTrue(fs.exists(walPartitionDir),
453          "A backup WALs subdirectory with today's date should exist: " + walPartitionDir);
454        // The directory exists - stop waiting
455        break;
456      } catch (AssertionError e) {
457        // Reach here when the directory currently does not exist
458        if ((System.currentTimeMillis() - currentTimeMs) >= waitTimeSec * 1000) {
459          throw new AssertionError(e);
460        }
461        LOG.info("Waiting up to {} seconds for WAL partition directory to exist: {}", waitTimeSec,
462          walPartitionDir);
463        Thread.sleep(1000);
464      }
465    }
466    return walPartitionDir;
467  }
468
469  /**
470   * Verifies the WAL partition directory contains a backup WAL file. The WAL file's path will look
471   * something like the following:
472   * .../backupWALDir/WALs/2025-10-17/wal_file.1760738249595.1880be89-0b69-4bad-8d0e-acbf25c63b7e
473   * @param walPartitionDir The date directory for a backup WAL i.e.
474   *                        .../backupWALDir/WALs/2025-10-17
475   */
476  private void verifyBackupWALFiles(Path walPartitionDir) throws IOException {
477    FileStatus[] fileStatuses = fs.listStatus(walPartitionDir);
478    for (FileStatus fileStatus : fileStatuses) {
479      String walFileName = fileStatus.getPath().getName();
480      String[] splitName = walFileName.split("\\.");
481      assertEquals("wal_file", splitName[0],
482        "The WAL partition directory should only have files that start with 'wal_file'");
483      assertEquals(walPartitionDir.getName(),
484        BackupUtils.formatToDateString(Long.parseLong(splitName[1])),
485        "The timestamp in the WAL file's name should match the date for the WAL partition directory");
486    }
487  }
488
489  /**
490   * Checks if all timestamps in a map with ServerName and timestamp key-values pairs are past the
491   * provided timestamp threshold.
492   * @param timestamps Map with ServerName and timestamp key-value pairs
493   * @param threshold  Timestamp to check all timestamp values in the map against
494   * @return True if all timestamps values in the map have met or are path the timestamp threshold;
495   *         False otherwise
496   */
497  private boolean areAllTimestampsPastThreshold(Map<ServerName, Long> timestamps, long threshold,
498    TableName tableName) {
499    boolean haveAllTimestampsReachedThreshold = true;
500    LOG.info("Checking if each region server in the replication check point "
501      + "has caught up to the latest Put on {}", tableName.getNameAsString());
502    for (Map.Entry<ServerName, Long> entry : timestamps.entrySet()) {
503      LOG.debug("host={}, checkpoint timestamp={}, latest put timestamp={}, caught up={}",
504        entry.getKey(), entry.getValue(), threshold, entry.getValue() >= threshold);
505      if (entry.getValue() < threshold) {
506        // Not returning right away so all hosts and timestamps are logged
507        haveAllTimestampsReachedThreshold = false;
508      }
509    }
510    return haveAllTimestampsReachedThreshold;
511  }
512
513  /**
514   * Waits for the replication checkpoint timestamp of each region server to meet or pass the
515   * timestamp of the latest Put operation on the backed-up table.
516   * @param conn               Minicluster connection
517   * @param latestPutTimestamp Timestamp of the latest Put operation on the backed-up table
518   */
519  private void waitForCheckpointTimestampsToUpdate(Connection conn, long latestPutTimestamp,
520    TableName tableName) throws IOException, InterruptedException {
521    BackupSystemTable backupSystemTable = new BackupSystemTable(conn);
522    Map<ServerName, Long> checkpointTimestamps = backupSystemTable.getBackupCheckpointTimestamps();
523    int i = 0;
524    int sleepTimeSec = 10;
525    int waitThresholdMs = 15 * 60 * 1000;
526    long waitStartTime = System.currentTimeMillis();
527    while (!areAllTimestampsPastThreshold(checkpointTimestamps, latestPutTimestamp, tableName)) {
528      if ((System.currentTimeMillis() - waitStartTime) >= waitThresholdMs) {
529        throw new RuntimeException("Timed out waiting for the replication checkpoint timestamp of "
530          + "each region server to catch up with timestamp of the latest Put on "
531          + tableName.getNameAsString());
532      }
533      LOG.info(
534        "Waiting {} seconds for the replication checkpoint timestamp for each region server "
535          + "to catch up with the timestamp of the latest Put on {}",
536        sleepTimeSec, tableName.getNameAsString());
537      Thread.sleep(sleepTimeSec * 1000);
538      checkpointTimestamps = backupSystemTable.getBackupCheckpointTimestamps();
539      i++;
540    }
541    LOG.info("Done waiting. Total wait time: {} seconds", i * sleepTimeSec);
542  }
543
544  /**
545   * Scans the backed-up table and returns the timestamp (ms) of the latest Put operation on the
546   * table.
547   * @param table The backed-up table to scan
548   * @return Timestamp of the most recent Put on the backed-up table
549   */
550  private long getLatestPutTimestamp(Table table) throws IOException {
551    Scan scan = new Scan();
552    ResultScanner resultScanner = table.getScanner(scan);
553    long timestamp = 0;
554    for (Result result : resultScanner) {
555      for (Cell cell : result.rawCells()) {
556        if (cell.getTimestamp() > timestamp) {
557          timestamp = cell.getTimestamp();
558        }
559      }
560    }
561    return timestamp;
562  }
563
564  /**
565   * Restores a table using the provided backup ID and ensure the table has the correct row count
566   * after
567   */
568  private void restoreTableAndVerifyRowCount(Connection conn, BackupAdmin client, TableName table,
569    String backupId, long expectedRows) throws IOException {
570    TableName[] tablesRestoreIncMultiple = new TableName[] { table };
571    restore(
572      createRestoreRequest(backupRootDir, backupId, false, tablesRestoreIncMultiple, null, true),
573      client);
574    Table hTable = conn.getTable(table);
575    Assertions.assertEquals(expectedRows, HBaseTestingUtil.countRows(hTable));
576    hTable.close();
577  }
578
579  /**
580   * Uses the list of all backup IDs to return a sublist of incremental backup IDs. This method
581   * assumes the first backup in the list is a full backup, followed by incremental backups.
582   */
583  private String[] getAllIncrementalBackupIds(List<String> backupIds) {
584    int size = backupIds.size();
585    backupIds = backupIds.subList(1, size);
586    String[] arr = new String[size - 1];
587    backupIds.toArray(arr);
588    return arr;
589  }
590
591  /** Returns status of backup */
592  protected boolean checkSucceeded(String backupId) throws IOException {
593    BackupInfo status = getBackupInfo(backupId);
594    if (status == null) {
595      return false;
596    }
597    return status.getState() == BackupInfo.BackupState.COMPLETE;
598  }
599
600  private BackupInfo getBackupInfo(String backupId) throws IOException {
601    try (BackupSystemTable table = new BackupSystemTable(util.getConnection())) {
602      return table.readBackupInfo(backupId);
603    }
604  }
605
606  /**
607   * Get restore request.
608   * @param backupId    backup ID
609   * @param check       check the backup
610   * @param fromTables  table names to restore from
611   * @param toTables    new table names to restore to
612   * @param isOverwrite overwrite the table(s)
613   * @return an instance of RestoreRequest
614   */
615  public RestoreRequest createRestoreRequest(String backupRootDir, String backupId, boolean check,
616    TableName[] fromTables, TableName[] toTables, boolean isOverwrite) {
617    RestoreRequest.Builder builder = new RestoreRequest.Builder();
618    return builder.withBackupRootDir(backupRootDir).withBackupId(backupId).withCheck(check)
619      .withFromTables(fromTables).withToTables(toTables).withOverwrite(isOverwrite).build();
620  }
621
622  /**
623   * Iterates through the list of backups and verifies the full backup and latest incremental backup
624   * still exist, while also verifying all other backups no longer exist. This method is meant to be
625   * run after all incremental backups have been merged.
626   */
627  private void verifyBackupExistenceAfterMerge(List<String> backupIds) throws IOException {
628    String fullBackupId = backupIds.get(0);
629    String mostRecentIncrementalBackup = backupIds.get(backupIds.size() - 1);
630    for (String backupId : backupIds) {
631      if (backupId.equals(fullBackupId) || backupId.equals(mostRecentIncrementalBackup)) {
632        verifyBackupExists(backupId);
633      } else {
634        verifyBackupDoesNotExist(backupId);
635      }
636    }
637  }
638
639  private void removeNonexistentBackups(List<String> backupIds) throws IOException {
640    List<String> backupsToRemove = new ArrayList<>();
641    for (String backupId : backupIds) {
642      if (!fs.exists(new Path(backupRootDir, backupId))) {
643        backupsToRemove.add(backupId);
644      }
645    }
646    for (String backupId : backupsToRemove) {
647      LOG.info("Removing {} from list of backup IDs since it no longer exists", backupId);
648      backupIds.remove(backupId);
649    }
650  }
651
652  /**
653   * Performs the delete command for the most recently taken incremental backup, and also removes
654   * this backup from the list of backup IDs.
655   */
656  private void deleteMostRecentIncrementalBackup(List<String> backupIds, BackupAdmin client)
657    throws IOException {
658    String incrementalBackupId = backupIds.get(backupIds.size() - 1);
659    LOG.info("Deleting the most recently created incremental backup: {}", incrementalBackupId);
660    verifyBackupExists(incrementalBackupId);
661    delete(new String[] { incrementalBackupId }, client);
662    backupIds.remove(backupIds.size() - 1);
663    verifyBackupDoesNotExist(incrementalBackupId);
664  }
665
666  /**
667   * Verifies all backups in the list of backup IDs actually exist on the filesystem.
668   */
669  private void verifyAllBackupsExist(List<String> backupIds) throws IOException {
670    for (String backupId : backupIds) {
671      verifyBackupExists(backupId);
672    }
673  }
674
675  /**
676   * Verifies zero backups in the list of backup IDs exist on the filesystem.
677   */
678  private void verifyNoBackupsExist(List<String> backupIds) throws IOException {
679    for (String backupId : backupIds) {
680      verifyBackupDoesNotExist(backupId);
681    }
682  }
683
684  private void verifyBackupExists(String backupId) throws IOException {
685    assertTrue(fs.exists(new Path(backupRootDir, backupId)),
686      "Backup " + backupId + " should exist inside of " + backupRootDir);
687  }
688
689  private void verifyBackupDoesNotExist(String backupId) throws IOException {
690    assertFalse(fs.exists(new Path(backupRootDir, backupId)),
691      "Backup " + backupId + " should not exist inside of " + backupRootDir);
692  }
693
694  @Override
695  public void setUpCluster() throws Exception {
696    util = getTestingUtil(getConf());
697    conf = getConf();
698    BackupTestUtil.enableBackup(conf);
699    LOG.debug("Initializing/checking cluster has {} servers", regionServerCount);
700    util.initializeCluster(regionServerCount);
701    LOG.debug("Done initializing/checking cluster");
702  }
703
704  @Override
705  public TableName getTablename() {
706    // That is only valid when Monkey is CALM (no monkey)
707    return null;
708  }
709
710  @Override
711  protected Set<String> getColumnFamilies() {
712    // That is only valid when Monkey is CALM (no monkey)
713    return null;
714  }
715
716  /**
717   * This method is useful for ensuring important test parameters are initialized when running
718   * integration tests in IntelliJ or in the command line with Maven. Tests executed in the command
719   * line via the bin/hbase command have their parameters set in
720   * {@link IntegrationTestBase#processOptions}
721   */
722  protected void initializeTestParameters() {
723    util = new IntegrationTestingUtility();
724    conf = util.getConfiguration();
725
726    regionsCountPerServer = conf.getInt(REGION_COUNT_KEY,
727      regionsCountPerServer > 0 ? regionsCountPerServer : DEFAULT_REGION_COUNT);
728    LOG.debug("regionsCountPerServer is set to {}", regionsCountPerServer);
729
730    regionServerCount = conf.getInt(REGIONSERVER_COUNT_KEY,
731      regionServerCount > 0 ? regionServerCount : DEFAULT_REGIONSERVER_COUNT);
732    LOG.debug("regionServerCount is set to {}", regionServerCount);
733
734    rowsInIteration = conf.getInt(ROWS_PER_ITERATION_KEY,
735      rowsInIteration > 0 ? rowsInIteration : DEFAULT_ROWS_IN_ITERATION);
736    LOG.debug("rowsInIteration is set to {}", rowsInIteration);
737
738    numIterations =
739      conf.getInt(NUM_ITERATIONS_KEY, numIterations > 0 ? numIterations : DEFAULT_NUM_ITERATIONS);
740    LOG.debug("numIterations is set to {}", numIterations);
741
742    numTables =
743      conf.getInt(NUMBER_OF_TABLES_KEY, numTables > 0 ? numTables : DEFAULT_NUMBER_OF_TABLES);
744    LOG.debug("numTables is set to {}", numTables);
745
746    sleepTime = conf.getLong(SLEEP_TIME_KEY, sleepTime > 0 ? sleepTime : SLEEP_TIME_DEFAULT);
747    LOG.debug("sleepTime is set to {}", sleepTime);
748  }
749
750  @Override
751  protected void addOptions() {
752    addOptWithArg(REGIONSERVER_COUNT_KEY,
753      "Total number of region servers. Default: '" + DEFAULT_REGIONSERVER_COUNT + "'");
754    addOptWithArg(REGION_COUNT_KEY, "Total number of regions. Default: " + DEFAULT_REGION_COUNT);
755    addOptWithArg(ROWS_PER_ITERATION_KEY,
756      "Total number of data rows to be loaded during one iteration." + " Default: "
757        + DEFAULT_ROWS_IN_ITERATION);
758    addOptWithArg(NUM_ITERATIONS_KEY,
759      "Total number iterations." + " Default: " + DEFAULT_NUM_ITERATIONS);
760    addOptWithArg(NUMBER_OF_TABLES_KEY,
761      "Total number of tables in the test." + " Default: " + DEFAULT_NUMBER_OF_TABLES);
762    addOptWithArg(SLEEP_TIME_KEY, "Sleep time of chaos monkey in ms "
763      + "to restart random region server. Default: " + SLEEP_TIME_DEFAULT);
764  }
765
766  @Override
767  protected void processOptions(CommandLine cmd) {
768    super.processOptions(cmd);
769    regionsCountPerServer = Integer
770      .parseInt(cmd.getOptionValue(REGION_COUNT_KEY, Integer.toString(DEFAULT_REGION_COUNT)));
771    regionServerCount = Integer.parseInt(
772      cmd.getOptionValue(REGIONSERVER_COUNT_KEY, Integer.toString(DEFAULT_REGIONSERVER_COUNT)));
773    rowsInIteration = Integer.parseInt(
774      cmd.getOptionValue(ROWS_PER_ITERATION_KEY, Integer.toString(DEFAULT_ROWS_IN_ITERATION)));
775    numIterations = Integer
776      .parseInt(cmd.getOptionValue(NUM_ITERATIONS_KEY, Integer.toString(DEFAULT_NUM_ITERATIONS)));
777    numTables = Integer.parseInt(
778      cmd.getOptionValue(NUMBER_OF_TABLES_KEY, Integer.toString(DEFAULT_NUMBER_OF_TABLES)));
779    sleepTime =
780      Long.parseLong(cmd.getOptionValue(SLEEP_TIME_KEY, Long.toString(SLEEP_TIME_DEFAULT)));
781
782    LOG.info(MoreObjects.toStringHelper("Parsed Options")
783      .add(REGION_COUNT_KEY, regionsCountPerServer).add(REGIONSERVER_COUNT_KEY, regionServerCount)
784      .add(ROWS_PER_ITERATION_KEY, rowsInIteration).add(NUM_ITERATIONS_KEY, numIterations)
785      .add(NUMBER_OF_TABLES_KEY, numTables).add(SLEEP_TIME_KEY, sleepTime).toString());
786  }
787
788}