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.backup;
019
020import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONF_CONTINUOUS_BACKUP_WAL_DIR;
021import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONTINUOUS_BACKUP_REPLICATION_PEER;
022import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_ENABLE_CONTINUOUS_BACKUP;
023import static org.junit.jupiter.api.Assertions.assertEquals;
024import static org.junit.jupiter.api.Assertions.assertFalse;
025import static org.junit.jupiter.api.Assertions.assertTrue;
026
027import java.io.IOException;
028import java.util.Arrays;
029import java.util.HashSet;
030import java.util.List;
031import java.util.Map;
032import java.util.stream.Collectors;
033import org.apache.hadoop.fs.FileSystem;
034import org.apache.hadoop.fs.Path;
035import org.apache.hadoop.hbase.TableName;
036import org.apache.hadoop.hbase.backup.impl.BackupManifest;
037import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
038import org.apache.hadoop.hbase.client.Admin;
039import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
040import org.apache.hadoop.hbase.testclassification.LargeTests;
041import org.apache.hadoop.util.ToolRunner;
042import org.junit.jupiter.api.AfterEach;
043import org.junit.jupiter.api.BeforeEach;
044import org.junit.jupiter.api.Tag;
045import org.junit.jupiter.api.Test;
046import org.slf4j.Logger;
047import org.slf4j.LoggerFactory;
048
049import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
050
051@Tag(LargeTests.TAG)
052public class TestContinuousBackup extends TestBackupBase {
053
054  private static final Logger LOG = LoggerFactory.getLogger(TestContinuousBackup.class);
055
056  String backupWalDirName = "TestContinuousBackupWalDir";
057
058  @BeforeEach
059  public void beforeTest() throws IOException {
060    Path root = TEST_UTIL.getDataTestDirOnTestFS();
061    Path backupWalDir = new Path(root, backupWalDirName);
062    FileSystem fs = FileSystem.get(conf1);
063    fs.mkdirs(backupWalDir);
064    conf1.set(CONF_CONTINUOUS_BACKUP_WAL_DIR, backupWalDir.toString());
065  }
066
067  @AfterEach
068  public void afterTest() throws IOException {
069    Path root = TEST_UTIL.getDataTestDirOnTestFS();
070    Path backupWalDir = new Path(root, backupWalDirName);
071    FileSystem fs = FileSystem.get(conf1);
072
073    if (fs.exists(backupWalDir)) {
074      fs.delete(backupWalDir, true);
075    }
076
077    conf1.unset(CONF_CONTINUOUS_BACKUP_WAL_DIR);
078    deleteContinuousBackupReplicationPeerIfExists(TEST_UTIL.getAdmin());
079  }
080
081  @Test
082  public void testContinuousBackupWithFullBackup() throws Exception {
083    LOG.info("Testing successful continuous backup with full backup");
084    String methodName = Thread.currentThread().getStackTrace()[1].getMethodName();
085    TableName tableName = TableName.valueOf("table_" + methodName);
086    TEST_UTIL.createTable(tableName, "cf");
087
088    try (BackupSystemTable table = new BackupSystemTable(TEST_UTIL.getConnection())) {
089      int before = table.getBackupHistory().size();
090
091      // Run backup
092      String[] args = buildBackupArgs("full", new TableName[] { tableName }, true);
093      int ret = ToolRunner.run(conf1, new BackupDriver(), args);
094      assertEquals(0, ret, "Backup should succeed");
095
096      // Verify backup history increased and all the backups are succeeded
097      List<BackupInfo> backups = table.getBackupHistory();
098      assertEquals(before + 1, backups.size(), "Backup history should increase");
099      for (BackupInfo data : List.of(backups.get(0))) {
100        String backupId = data.getBackupId();
101        assertTrue(checkSucceeded(backupId));
102      }
103
104      // Verify backup manifest contains the correct tables
105      BackupManifest manifest = getLatestBackupManifest(backups);
106      assertEquals(Sets.newHashSet(tableName), new HashSet<>(manifest.getTableList()),
107        "Backup should contain the expected tables");
108    }
109
110    // Verify replication peer subscription
111    verifyReplicationPeerSubscription(tableName);
112
113    // Verify table is registered in Backup System Table
114    verifyTableInBackupSystemTable(tableName);
115  }
116
117  @Test
118  public void testContinuousBackupForMultipleTables() throws Exception {
119    LOG.info("Test continuous backup for multiple tables");
120    String methodName = Thread.currentThread().getStackTrace()[1].getMethodName();
121    TableName tableName1 = TableName.valueOf("table_" + methodName);
122    TEST_UTIL.createTable(tableName1, "cf");
123    TableName tableName2 = TableName.valueOf("table_" + methodName + "2");
124    TEST_UTIL.createTable(tableName2, "cf");
125
126    try (BackupSystemTable table = new BackupSystemTable(TEST_UTIL.getConnection())) {
127      int before = table.getBackupHistory().size();
128
129      // Create full backup for table1
130      String[] args = buildBackupArgs("full", new TableName[] { tableName1 }, true);
131      int ret = ToolRunner.run(conf1, new BackupDriver(), args);
132      assertEquals(0, ret, "Backup should succeed");
133
134      // Create full backup for table2
135      args = buildBackupArgs("full", new TableName[] { tableName2 }, true);
136      ret = ToolRunner.run(conf1, new BackupDriver(), args);
137      assertEquals(0, ret, "Backup should succeed");
138
139      // Verify backup history increased and all the backups are succeeded
140      List<BackupInfo> backups = table.getBackupHistory();
141      assertEquals(before + 2, backups.size(), "Backup history should increase");
142      for (BackupInfo data : List.of(backups.get(0), backups.get(1))) {
143        String backupId = data.getBackupId();
144        assertTrue(checkSucceeded(backupId));
145      }
146
147      // Verify backup manifest contains the correct tables
148      BackupManifest manifest = getLatestBackupManifest(backups);
149      assertEquals(Sets.newHashSet(tableName2), new HashSet<>(manifest.getTableList()),
150        "Backup should contain the expected tables");
151    }
152
153    // Verify replication peer subscription for each table
154    verifyReplicationPeerSubscription(tableName1);
155    verifyReplicationPeerSubscription(tableName2);
156
157    // Verify tables are registered in Backup System Table
158    verifyTableInBackupSystemTable(tableName1);
159    verifyTableInBackupSystemTable(tableName2);
160  }
161
162  @Test
163  public void testInvalidBackupScenarioWithContinuousEnabled() throws Exception {
164    LOG.info("Testing invalid backup scenario with continuous backup enabled");
165    String methodName = Thread.currentThread().getStackTrace()[1].getMethodName();
166    TableName tableName1 = TableName.valueOf("table_" + methodName);
167    TEST_UTIL.createTable(tableName1, "cf");
168    TableName tableName2 = TableName.valueOf("table_" + methodName + "2");
169    TEST_UTIL.createTable(tableName2, "cf");
170
171    try (BackupSystemTable table = new BackupSystemTable(TEST_UTIL.getConnection())) {
172      int before = table.getBackupHistory().size();
173
174      // Create full backup for table1 with continuous backup enabled
175      String[] args = buildBackupArgs("full", new TableName[] { tableName1 }, true);
176      int ret = ToolRunner.run(conf1, new BackupDriver(), args);
177      assertEquals(0, ret, "Backup should succeed");
178
179      // Create full backup for table2 without continuous backup enabled
180      args = buildBackupArgs("full", new TableName[] { tableName2 }, false);
181      ret = ToolRunner.run(conf1, new BackupDriver(), args);
182      assertEquals(0, ret, "Backup should succeed");
183
184      // Attempt full backup for both tables without continuous backup enabled (should fail)
185      args = buildBackupArgs("full", new TableName[] { tableName1, tableName2 }, false);
186      ret = ToolRunner.run(conf1, new BackupDriver(), args);
187      assertTrue(ret != 0, "Backup should fail due to mismatch in continuous backup settings");
188
189      // Verify backup history size is unchanged after the failed backup
190      int after = table.getBackupHistory().size();
191      assertEquals(before + 2, after, "Backup history should remain unchanged on failure");
192    }
193  }
194
195  @Test
196  public void testContinuousBackupWithWALDirNotSpecified() throws Exception {
197    LOG.info("Testing that continuous backup fails when WAL directory is not specified");
198    String methodName = Thread.currentThread().getStackTrace()[1].getMethodName();
199    TableName tableName = TableName.valueOf("table_" + methodName);
200    TEST_UTIL.createTable(tableName, "cf");
201
202    conf1.unset(CONF_CONTINUOUS_BACKUP_WAL_DIR);
203    LOG.info("CONF_CONTINUOUS_BACKUP_WAL_DIR: {}", conf1.get(CONF_CONTINUOUS_BACKUP_WAL_DIR));
204
205    try (BackupSystemTable table = new BackupSystemTable(TEST_UTIL.getConnection())) {
206      int before = table.getBackupHistory().size();
207
208      // Run full backup without specifying WAL directory (invalid scenario)
209      String[] args = buildBackupArgs("full", new TableName[] { tableName }, true);
210      int ret = ToolRunner.run(conf1, new BackupDriver(), args);
211
212      assertTrue(ret != 0, "Backup should fail when WAL directory is not specified");
213
214      List<BackupInfo> backups = table.getBackupHistory();
215      int after = backups.size();
216      assertEquals(before + 1, after, "Backup history should increase");
217
218      // last backup should be a failure
219      assertFalse(checkSucceeded(backups.get(0).getBackupId()));
220    }
221  }
222
223  @Test
224  public void testContinuousBackupWithIncrementalBackup() throws Exception {
225    LOG.info("Testing that continuous backup cannot be enabled with incremental backup");
226    String methodName = Thread.currentThread().getStackTrace()[1].getMethodName();
227    TableName tableName = TableName.valueOf("table_" + methodName);
228    TEST_UTIL.createTable(tableName, "cf");
229
230    try (BackupSystemTable table = new BackupSystemTable(TEST_UTIL.getConnection())) {
231      int before = table.getBackupHistory().size();
232
233      // Run incremental backup with continuous backup flag (invalid scenario)
234      String[] args = buildBackupArgs("incremental", new TableName[] { tableName }, true);
235      int ret = ToolRunner.run(conf1, new BackupDriver(), args);
236
237      assertTrue(ret != 0, "Backup should fail when using continuous backup with incremental mode");
238
239      // Backup history should remain unchanged
240      int after = table.getBackupHistory().size();
241      assertEquals(before, after, "Backup history should remain unchanged on failure");
242    }
243  }
244
245  private void verifyReplicationPeerSubscription(TableName table) throws IOException {
246    try (Admin admin = TEST_UTIL.getAdmin()) {
247      ReplicationPeerDescription peerDesc = admin.listReplicationPeers().stream()
248        .filter(peer -> peer.getPeerId().equals(CONTINUOUS_BACKUP_REPLICATION_PEER)).findFirst()
249        .orElseThrow(() -> new AssertionError("Replication peer not found"));
250
251      assertTrue(peerDesc.getPeerConfig().getTableCFsMap().containsKey(table),
252        "Table should be subscribed to the replication peer");
253    }
254  }
255
256  String[] buildBackupArgs(String backupType, TableName[] tables, boolean continuousEnabled) {
257    String tableNames =
258      Arrays.stream(tables).map(TableName::getNameAsString).collect(Collectors.joining(","));
259
260    if (continuousEnabled) {
261      return new String[] { "create", backupType, BACKUP_ROOT_DIR, "-t", tableNames,
262        "-" + OPTION_ENABLE_CONTINUOUS_BACKUP };
263    } else {
264      return new String[] { "create", backupType, BACKUP_ROOT_DIR, "-t", tableNames };
265    }
266  }
267
268  private void verifyTableInBackupSystemTable(TableName table) throws IOException {
269    try (BackupSystemTable backupTable = new BackupSystemTable(TEST_UTIL.getConnection())) {
270      Map<TableName, Long> tableBackupMap = backupTable.getContinuousBackupTableSet();
271
272      assertTrue(tableBackupMap.containsKey(table),
273        "Table is missing in the continuous backup table set");
274
275      assertTrue(tableBackupMap.get(table) > 0, "Timestamp for table should be greater than 0");
276    }
277  }
278
279}