DieterDePaepe commented on code in PR #8694:
URL: https://github.com/apache/hbase/pull/8694#discussion_r4286920215
##########
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java:
##########
@@ -248,6 +263,75 @@ private void performSnapshots(Admin admin) throws
IOException {
}
}
+ /**
+ * Computes the per-host log boundaries stored by this full backup, using
Review Comment:
I had a hard time parsing this text when starting the review, though it's
completely correct. Some thoughts - pick what you agree with:
- Perhaps introduce a short starting paragraph that explains the purpose,
list the details in later paragraphs.
- `are included in the next incremental backup` -> `will be included`
- The text `The live servers are read ... treating live servers as dead.`
might be more suited for inline documentation (as its an implementation detail)?
##########
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/util/BackupUtils.java:
##########
@@ -206,6 +208,84 @@ public static String parseHostNameFromLogFile(Path p) {
}
}
+ /**
+ * Returns the log roll result of every region server that took part in the
log roll that happened
+ * between reading {@code previousLogRolls} and {@code latestLogRolls}. A
region server that did
+ * not take part (for example because it is offline) keeps its old roll
result, which is never
+ * removed, so it is recognized by its roll result not changing.
+ * @param previousLogRolls roll results by host, read before the log roll
+ * @param latestLogRolls roll results by host, read after the log roll
+ * @return roll results by host, for the hosts that took part in the log roll
+ */
+ public static Map<String, Long> getRolledHosts(Map<String, Long>
previousLogRolls,
+ Map<String, Long> latestLogRolls) {
+ return latestLogRolls.entrySet().stream()
+ .filter(entry ->
!entry.getValue().equals(previousLogRolls.get(entry.getKey())))
+ .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+ }
+
+ /**
+ * Computes the per-host log boundaries stored by a backup that has no
previous boundaries to
+ * carry forward, such as a full backup. See
+ * {@link #computeLogBoundaries(Map, Map, Set, Collection, Collection)}.
+ */
+ public static Map<String, Long> computeLogBoundaries(Map<String, Long>
rolledHosts,
+ Collection<String> coveredLogs, Collection<String> pendingLogs) throws
IOException {
+ return computeLogBoundaries(rolledHosts, Collections.emptyMap(),
Collections.emptySet(),
+ coveredLogs, pendingLogs);
+ }
+
+ /**
+ * Computes the per-host log boundaries stored by a backup. Everything up to
a host's boundary is
+ * covered by backups, so later incremental backups only include that host's
logs that are newer.
+ * <ul>
+ * <li>A host that took part in the backup's log roll gets its roll
result.</li>
+ * <li>Any other host gets the creation time of its newest covered log,
capped to just below its
+ * oldest pending log, so that pending logs are included in a later
backup.</li>
+ * <li>A host without covered or pending logs keeps its previous boundary
while it still has logs,
+ * and otherwise gets no boundary.</li>
+ * </ul>
+ * Capping is needed because a host's logs can still be pending while some
of its newer logs are
+ * already covered, for example when a dead server's logs are split and
archived out of order. A
Review Comment:
Suggestion:
```
* already covered, for example when a dead server's logs are split and
archived out of order. As
* a result, the next backup includes the covered logs that are newer than
the cap a second time:
* losing the pending logs is worse than repeating edits that are already
backed up. A host is also...
```
##########
hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupOfflineRS.java:
##########
@@ -0,0 +1,439 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.backup;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.SingleProcessHBaseCluster;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.impl.BackupAdminImpl;
+import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
+import org.apache.hadoop.hbase.backup.impl.FullTableBackupClient;
+import org.apache.hadoop.hbase.backup.impl.TableBackupClient;
+import org.apache.hadoop.hbase.backup.util.BackupUtils;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.master.HMaster;
+import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
+
+/**
+ * Tests that WAL files from offline/inactive RegionServers are handled
correctly during backup.
+ * Specifically verifies that WALs from an offline RS are:
+ * <ol>
+ * <li>Backed up once in the first backup after the RS goes offline</li>
+ * <li>NOT re-backed up in subsequent backups</li>
+ * </ol>
+ */
+@Tag(LargeTests.TAG)
+public class TestBackupOfflineRS extends TestBackupBase {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TestBackupOfflineRS.class);
+
+ @BeforeAll
+ public static void setUp() throws Exception {
+ TEST_UTIL = new HBaseTestingUtil();
+ conf1 = TEST_UTIL.getConfiguration();
+ conf1.setInt("hbase.regionserver.info.port", -1);
+ conf1.setInt(HMaster.HBASE_MASTER_CLEANER_INTERVAL, Integer.MAX_VALUE);
+ autoRestoreOnFailure = true;
+ useSecondCluster = false;
+ setUpHelper();
+ TEST_UTIL.getMiniHBaseCluster().startRegionServer();
+ TEST_UTIL.waitTableAvailable(table1);
+ }
+
+ private static Runnable afterSnapshotHook;
+
+ /** Full backup client that runs {@link #afterSnapshotHook} once the table
snapshots exist. */
+ public static class FullTableBackupClientWithHook extends
FullTableBackupClient {
+ @Override
+ protected void snapshotCopy(BackupInfo backupInfo) throws IOException {
+ afterSnapshotHook.run();
+ super.snapshotCopy(backupInfo);
+ }
+ }
+
+ private static void stopRegionServerAndWait(ServerName serverName) throws
Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ HMaster master = cluster.getMaster();
+ cluster.stopRegionServer(serverName);
+ cluster.waitForRegionServerToStop(serverName, 60_000);
+ TEST_UTIL.waitFor(60_000,
+ () ->
master.getProcedures().stream().filter(ServerCrashProcedure.class::isInstance)
+ .map(ServerCrashProcedure.class::cast)
+ .anyMatch(scp -> scp.getServerName().equals(serverName) &&
scp.isFinished()));
+ TEST_UTIL.waitUntilNoRegionsInTransition(60_000);
+ }
+
+ private static void restore(String backupId, TableName sourceTable,
TableName restoredTable)
+ throws Exception {
+ try (Connection conn = ConnectionFactory.createConnection(conf1);
+ BackupAdminImpl backupAdmin = new BackupAdminImpl(conn)) {
+ backupAdmin.restore(BackupUtils.createRestoreRequest(BACKUP_ROOT_DIR,
backupId, false,
+ new TableName[] { sourceTable }, new TableName[] { restoredTable },
true));
+ }
+ }
+
+ private static void restoreAndAssertAllRowsPresent(String backupId, String
restoredTableName,
+ String message) throws Exception {
+ TableName restoredTable = TableName.valueOf(restoredTableName);
+ restore(backupId, table1, restoredTable);
+ assertEquals(TEST_UTIL.countRows(table1),
TEST_UTIL.countRows(restoredTable), message);
+ }
+
+ private static Long boundary(BackupSystemTable sysTable, String host) throws
IOException {
+ return sysTable.readLogTimestampMap(BACKUP_ROOT_DIR).get(table1).get(host);
+ }
+
+ private static void moveRegionAndWait(TableName table, HRegionServer
destination)
+ throws Exception {
+ try (Admin admin = TEST_UTIL.getConnection().getAdmin()) {
+ RegionInfo region = admin.getRegions(table).get(0);
+ admin.move(region.getEncodedNameAsBytes(), destination.getServerName());
+ }
+ TEST_UTIL.waitFor(60_000, () -> !destination.getRegions(table).isEmpty());
+ TEST_UTIL.waitUntilAllRegionsAssigned(table);
+ }
+
+ /**
+ * Tests that when a full backup is taken while an RS is offline (with WALs
in oldlogs), the
+ * offline host's timestamps are recorded so subsequent incremental backups
don't reinclude those
+ * WALs.
+ */
+ @Test
+ public void testBackupWithOfflineRS() throws Exception {
+ LOG.info("Starting testBackupWithOfflineRS");
+
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+
+ HRegionServer rsBeforeStop =
cluster.startRegionServerAndWait(10000).getRegionServer();
+ moveRegionAndWait(table1, rsBeforeStop);
+
+ LOG.info("Inserting data to generate WAL entries");
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 2, 100);
+ }
+
+ String offlineHost =
+ rsBeforeStop.getServerName().getHostname() + ":" +
rsBeforeStop.getServerName().getPort();
+ LOG.info("Stopping RS: {}", offlineHost);
+
+ stopRegionServerAndWait(rsBeforeStop.getServerName());
+
+ LOG.info("Taking full backup (with offline RS WALs in oldlogs)");
+ String fullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(fullBackupId), "Full backup should succeed");
+
+ try (BackupSystemTable sysTable = new
BackupSystemTable(TEST_UTIL.getConnection())) {
+ Map<TableName, Map<String, Long>> timestamps =
sysTable.readLogTimestampMap(BACKUP_ROOT_DIR);
+ Map<String, Long> rsTimestamps = timestamps.get(table1);
+ LOG.info("RS timestamps after full backup: {}", rsTimestamps);
+
+ Long tsAfterFullBackup = rsTimestamps.get(offlineHost);
+ assertNotNull(tsAfterFullBackup,
+ "Offline host should have timestamp recorded in trslm after full
backup");
+
+ LOG.info("Taking incremental backup (should NOT include offline RS
WALs)");
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId), "Incremental backup should
succeed");
+
+ assertEquals(tsAfterFullBackup, boundary(sysTable, offlineHost),
+ "Incremental backup moved the boundary of the offline host");
+ }
+ }
+
+ /**
+ * Tests that WALs written to an RS after a full backup are correctly
included in the subsequent
+ * incremental backup, even if that RS has gone offline before the
incremental runs.
+ */
+ @Test
+ public void testRSGoesOfflineAfterFullBackupBeforeIncremental() throws
Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+
+ HRegionServer rsBeforeStop =
cluster.startRegionServerAndWait(10000).getRegionServer();
+
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 3, 50);
+ }
+
+ String fullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(fullBackupId), "Full backup should succeed");
+
+ moveRegionAndWait(table1, rsBeforeStop);
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 4, 50);
+ }
+
+ rsBeforeStop.getWalRoller().requestRollAll();
+ rsBeforeStop.getWalRoller().waitUntilWalRollFinished();
+ String offlineHost =
+ rsBeforeStop.getServerName().getHostname() + ":" +
rsBeforeStop.getServerName().getPort();
+
+ stopRegionServerAndWait(rsBeforeStop.getServerName());
+
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId), "Incremental backup should
succeed");
+
+ restoreAndAssertAllRowsPresent(incrBackupId,
"table1_rs_offline_after_full_backup",
+ "Restored table should contain the rows written to the RS that went
offline");
+
+ try (BackupSystemTable sysTable = new
BackupSystemTable(TEST_UTIL.getConnection())) {
+ Long boundaryAfterIncr1 = boundary(sysTable, offlineHost);
+ assertNotNull(boundaryAfterIncr1,
+ "Offline RS should have a timestamp boundary after the incremental
backed up its WALs");
+ Long staleLogRoll =
Review Comment:
add:
```
assertFalse(walsOfHost(sysTable, incrBackupId, offlineHost).isEmpty(),
"First incremental after the RS went offline should include its
WALs");
```
##########
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalBackupManager.java:
##########
@@ -217,18 +236,14 @@ private List<String> getLogFilesForNewBackup(Map<String,
Long> olderTimestamps,
resultLogFiles.add(currentLogFile);
}
- // It is possible that a host in .oldlogs is an obsolete region server
- // so newestTimestamps.get(host) here can be null.
- // Even if these logs belong to a obsolete region server, we still need
- // to include they to avoid loss of edits for backup.
Long newTimestamp = newestTimestamps.get(host);
- if (newTimestamp == null || currentLogTS > newTimestamp) {
+ if (newTimestamp != null && currentLogTS > newTimestamp) {
newestLogs.add(currentLogFile);
}
}
// remove newest log per host because they are still in use
Review Comment:
```
// Held-back logs are newer than the roll result of their host, or
belong to a host that did not
// take part in the log roll. A later backup includes them.
```
##########
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalBackupManager.java:
##########
Review Comment:
Needs updating:
```
// A host without a roll result did not take part in this backup's
log roll: it is offline
// (for example a dead server whose logs are still being split), or
it started after the
// roll. Its logs are held back, like logs newer than the roll
result of a host that took
// part. BackupUtils#computeLogBoundaries keeps the boundary of the
host below held-back
// logs, so a later backup includes them.
Long ts = newestTimestamps.get(host);
if (ts == null) {
LOG.debug("Holding back log {} of host {}, which did not take part
in the log roll",
log.getPath(), host);
}
```
##########
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/util/BackupUtils.java:
##########
@@ -206,6 +208,84 @@ public static String parseHostNameFromLogFile(Path p) {
}
}
+ /**
+ * Returns the log roll result of every region server that took part in the
log roll that happened
+ * between reading {@code previousLogRolls} and {@code latestLogRolls}. A
region server that did
+ * not take part (for example because it is offline) keeps its old roll
result, which is never
+ * removed, so it is recognized by its roll result not changing.
+ * @param previousLogRolls roll results by host, read before the log roll
+ * @param latestLogRolls roll results by host, read after the log roll
+ * @return roll results by host, for the hosts that took part in the log roll
+ */
+ public static Map<String, Long> getRolledHosts(Map<String, Long>
previousLogRolls,
+ Map<String, Long> latestLogRolls) {
+ return latestLogRolls.entrySet().stream()
+ .filter(entry ->
!entry.getValue().equals(previousLogRolls.get(entry.getKey())))
+ .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+ }
+
+ /**
+ * Computes the per-host log boundaries stored by a backup that has no
previous boundaries to
+ * carry forward, such as a full backup. See
+ * {@link #computeLogBoundaries(Map, Map, Set, Collection, Collection)}.
+ */
+ public static Map<String, Long> computeLogBoundaries(Map<String, Long>
rolledHosts,
+ Collection<String> coveredLogs, Collection<String> pendingLogs) throws
IOException {
+ return computeLogBoundaries(rolledHosts, Collections.emptyMap(),
Collections.emptySet(),
+ coveredLogs, pendingLogs);
+ }
+
+ /**
+ * Computes the per-host log boundaries stored by a backup. Everything up to
a host's boundary is
+ * covered by backups, so later incremental backups only include that host's
logs that are newer.
+ * <ul>
+ * <li>A host that took part in the backup's log roll gets its roll
result.</li>
+ * <li>Any other host gets the creation time of its newest covered log,
capped to just below its
+ * oldest pending log, so that pending logs are included in a later
backup.</li>
+ * <li>A host without covered or pending logs keeps its previous boundary
while it still has logs,
+ * and otherwise gets no boundary.</li>
+ * </ul>
+ * Capping is needed because a host's logs can still be pending while some
of its newer logs are
+ * already covered, for example when a dead server's logs are split and
archived out of order. A
+ * host is also kept while any of its logs are pending, even if none are
covered, because without
+ * a boundary the next backup would treat it as unknown and skip its older
logs. For the same
+ * reason a host keeps its previous boundary while it still has logs:
without it, the next backup
+ * would include its logs that are already covered, which can bring back
deleted data.
+ * @param rolledHosts roll results of the hosts that took part in the
log roll
+ * @param previousBoundaries boundaries stored by the previous backup
+ * @param hostsWithLogs hosts that still have logs, whether or not this
backup includes them
+ * @param coveredLogs logs whose edits are covered by this backup
+ * @param pendingLogs logs whose edits might not be covered by this
backup
Review Comment:
The "might not be" was a bit confusing for me. Suggestion:
```
* @param coveredLogs logs whose edits are included in the backup
* @param pendingLogs logs whose edits are not to be included in
the backup
```
##########
hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupOfflineRS.java:
##########
@@ -0,0 +1,439 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.backup;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.SingleProcessHBaseCluster;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.impl.BackupAdminImpl;
+import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
+import org.apache.hadoop.hbase.backup.impl.FullTableBackupClient;
+import org.apache.hadoop.hbase.backup.impl.TableBackupClient;
+import org.apache.hadoop.hbase.backup.util.BackupUtils;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.master.HMaster;
+import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
+
+/**
+ * Tests that WAL files from offline/inactive RegionServers are handled
correctly during backup.
+ * Specifically verifies that WALs from an offline RS are:
+ * <ol>
+ * <li>Backed up once in the first backup after the RS goes offline</li>
+ * <li>NOT re-backed up in subsequent backups</li>
+ * </ol>
+ */
+@Tag(LargeTests.TAG)
+public class TestBackupOfflineRS extends TestBackupBase {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TestBackupOfflineRS.class);
+
+ @BeforeAll
+ public static void setUp() throws Exception {
+ TEST_UTIL = new HBaseTestingUtil();
+ conf1 = TEST_UTIL.getConfiguration();
+ conf1.setInt("hbase.regionserver.info.port", -1);
+ conf1.setInt(HMaster.HBASE_MASTER_CLEANER_INTERVAL, Integer.MAX_VALUE);
+ autoRestoreOnFailure = true;
+ useSecondCluster = false;
+ setUpHelper();
+ TEST_UTIL.getMiniHBaseCluster().startRegionServer();
+ TEST_UTIL.waitTableAvailable(table1);
+ }
+
+ private static Runnable afterSnapshotHook;
+
+ /** Full backup client that runs {@link #afterSnapshotHook} once the table
snapshots exist. */
+ public static class FullTableBackupClientWithHook extends
FullTableBackupClient {
+ @Override
+ protected void snapshotCopy(BackupInfo backupInfo) throws IOException {
+ afterSnapshotHook.run();
+ super.snapshotCopy(backupInfo);
+ }
+ }
+
+ private static void stopRegionServerAndWait(ServerName serverName) throws
Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ HMaster master = cluster.getMaster();
+ cluster.stopRegionServer(serverName);
+ cluster.waitForRegionServerToStop(serverName, 60_000);
+ TEST_UTIL.waitFor(60_000,
+ () ->
master.getProcedures().stream().filter(ServerCrashProcedure.class::isInstance)
+ .map(ServerCrashProcedure.class::cast)
+ .anyMatch(scp -> scp.getServerName().equals(serverName) &&
scp.isFinished()));
+ TEST_UTIL.waitUntilNoRegionsInTransition(60_000);
+ }
+
+ private static void restore(String backupId, TableName sourceTable,
TableName restoredTable)
+ throws Exception {
+ try (Connection conn = ConnectionFactory.createConnection(conf1);
+ BackupAdminImpl backupAdmin = new BackupAdminImpl(conn)) {
+ backupAdmin.restore(BackupUtils.createRestoreRequest(BACKUP_ROOT_DIR,
backupId, false,
+ new TableName[] { sourceTable }, new TableName[] { restoredTable },
true));
+ }
+ }
+
+ private static void restoreAndAssertAllRowsPresent(String backupId, String
restoredTableName,
+ String message) throws Exception {
+ TableName restoredTable = TableName.valueOf(restoredTableName);
+ restore(backupId, table1, restoredTable);
+ assertEquals(TEST_UTIL.countRows(table1),
TEST_UTIL.countRows(restoredTable), message);
+ }
+
+ private static Long boundary(BackupSystemTable sysTable, String host) throws
IOException {
+ return sysTable.readLogTimestampMap(BACKUP_ROOT_DIR).get(table1).get(host);
+ }
+
+ private static void moveRegionAndWait(TableName table, HRegionServer
destination)
+ throws Exception {
+ try (Admin admin = TEST_UTIL.getConnection().getAdmin()) {
+ RegionInfo region = admin.getRegions(table).get(0);
+ admin.move(region.getEncodedNameAsBytes(), destination.getServerName());
+ }
+ TEST_UTIL.waitFor(60_000, () -> !destination.getRegions(table).isEmpty());
+ TEST_UTIL.waitUntilAllRegionsAssigned(table);
+ }
+
+ /**
+ * Tests that when a full backup is taken while an RS is offline (with WALs
in oldlogs), the
+ * offline host's timestamps are recorded so subsequent incremental backups
don't reinclude those
+ * WALs.
+ */
+ @Test
+ public void testBackupWithOfflineRS() throws Exception {
+ LOG.info("Starting testBackupWithOfflineRS");
+
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+
+ HRegionServer rsBeforeStop =
cluster.startRegionServerAndWait(10000).getRegionServer();
+ moveRegionAndWait(table1, rsBeforeStop);
+
+ LOG.info("Inserting data to generate WAL entries");
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 2, 100);
+ }
+
+ String offlineHost =
+ rsBeforeStop.getServerName().getHostname() + ":" +
rsBeforeStop.getServerName().getPort();
+ LOG.info("Stopping RS: {}", offlineHost);
+
+ stopRegionServerAndWait(rsBeforeStop.getServerName());
+
+ LOG.info("Taking full backup (with offline RS WALs in oldlogs)");
+ String fullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(fullBackupId), "Full backup should succeed");
+
+ try (BackupSystemTable sysTable = new
BackupSystemTable(TEST_UTIL.getConnection())) {
+ Map<TableName, Map<String, Long>> timestamps =
sysTable.readLogTimestampMap(BACKUP_ROOT_DIR);
+ Map<String, Long> rsTimestamps = timestamps.get(table1);
+ LOG.info("RS timestamps after full backup: {}", rsTimestamps);
+
+ Long tsAfterFullBackup = rsTimestamps.get(offlineHost);
+ assertNotNull(tsAfterFullBackup,
+ "Offline host should have timestamp recorded in trslm after full
backup");
+
+ LOG.info("Taking incremental backup (should NOT include offline RS
WALs)");
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId), "Incremental backup should
succeed");
+
+ assertEquals(tsAfterFullBackup, boundary(sysTable, offlineHost),
+ "Incremental backup moved the boundary of the offline host");
+ }
Review Comment:
```
List<String> offlineHostWals = walsOfHost(sysTable, incrBackupId,
offlineHost);
assertTrue(offlineHostWals.isEmpty(),
"Incremental backup included WALs of the offline host that the full
backup covers: "
+ offlineHostWals);
```
with `walsOfHost`:
```
private static List<String> walsOfHost(BackupSystemTable sysTable, String
backupId, String host)
throws IOException {
List<String> wals =
sysTable.readBackupInfo(backupId).getIncrBackupFileList();
if (wals == null) {
return List.of();
}
return wals.stream()
.filter(wal -> host.equals(BackupUtils.parseHostNameFromLogFile(new
Path(wal)))).toList();
}
```
##########
hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupOfflineRS.java:
##########
@@ -0,0 +1,439 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.backup;
+
+import static org.junit.jupiter.api.Assertions.assertAll;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.SingleProcessHBaseCluster;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.impl.BackupAdminImpl;
+import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
+import org.apache.hadoop.hbase.backup.impl.FullTableBackupClient;
+import org.apache.hadoop.hbase.backup.impl.TableBackupClient;
+import org.apache.hadoop.hbase.backup.util.BackupUtils;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Get;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.client.ResultScanner;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.client.Table;
+import org.apache.hadoop.hbase.master.HMaster;
+import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
+import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
+
+/**
+ * Tests that WAL files from offline/inactive RegionServers are handled
correctly during backup.
+ * Specifically verifies that WALs from an offline RS are:
+ * <ol>
+ * <li>Backed up once in the first backup after the RS goes offline</li>
+ * <li>NOT re-backed up in subsequent backups</li>
+ * </ol>
+ */
+@Tag(LargeTests.TAG)
+public class TestBackupOfflineRS extends TestBackupBase {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TestBackupOfflineRS.class);
+
+ @BeforeAll
+ public static void setUp() throws Exception {
+ TEST_UTIL = new HBaseTestingUtil();
+ conf1 = TEST_UTIL.getConfiguration();
+ conf1.setInt("hbase.regionserver.info.port", -1);
+ conf1.setInt(HMaster.HBASE_MASTER_CLEANER_INTERVAL, Integer.MAX_VALUE);
+ autoRestoreOnFailure = true;
+ useSecondCluster = false;
+ setUpHelper();
+ TEST_UTIL.getMiniHBaseCluster().startRegionServer();
+ TEST_UTIL.waitTableAvailable(table1);
+ }
+
+ private static Runnable afterSnapshotHook;
+
+ /** Full backup client that runs {@link #afterSnapshotHook} once the table
snapshots exist. */
+ public static class FullTableBackupClientWithHook extends
FullTableBackupClient {
+ @Override
+ protected void snapshotCopy(BackupInfo backupInfo) throws IOException {
+ afterSnapshotHook.run();
+ super.snapshotCopy(backupInfo);
+ }
+ }
+
+ private static void stopRegionServerAndWait(ServerName serverName) throws
Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ HMaster master = cluster.getMaster();
+ cluster.stopRegionServer(serverName);
+ cluster.waitForRegionServerToStop(serverName, 60_000);
+ TEST_UTIL.waitFor(60_000,
+ () ->
master.getProcedures().stream().filter(ServerCrashProcedure.class::isInstance)
+ .map(ServerCrashProcedure.class::cast)
+ .anyMatch(scp -> scp.getServerName().equals(serverName) &&
scp.isFinished()));
+ TEST_UTIL.waitUntilNoRegionsInTransition(60_000);
+ }
+
+ private static void restore(String backupId, TableName sourceTable,
TableName restoredTable)
+ throws Exception {
+ try (Connection conn = ConnectionFactory.createConnection(conf1);
+ BackupAdminImpl backupAdmin = new BackupAdminImpl(conn)) {
+ backupAdmin.restore(BackupUtils.createRestoreRequest(BACKUP_ROOT_DIR,
backupId, false,
+ new TableName[] { sourceTable }, new TableName[] { restoredTable },
true));
+ }
+ }
+
+ private static void restoreAndAssertAllRowsPresent(String backupId, String
restoredTableName,
+ String message) throws Exception {
+ TableName restoredTable = TableName.valueOf(restoredTableName);
+ restore(backupId, table1, restoredTable);
+ assertEquals(TEST_UTIL.countRows(table1),
TEST_UTIL.countRows(restoredTable), message);
+ }
+
+ private static Long boundary(BackupSystemTable sysTable, String host) throws
IOException {
+ return sysTable.readLogTimestampMap(BACKUP_ROOT_DIR).get(table1).get(host);
+ }
+
+ private static void moveRegionAndWait(TableName table, HRegionServer
destination)
+ throws Exception {
+ try (Admin admin = TEST_UTIL.getConnection().getAdmin()) {
+ RegionInfo region = admin.getRegions(table).get(0);
+ admin.move(region.getEncodedNameAsBytes(), destination.getServerName());
+ }
+ TEST_UTIL.waitFor(60_000, () -> !destination.getRegions(table).isEmpty());
+ TEST_UTIL.waitUntilAllRegionsAssigned(table);
+ }
+
+ /**
+ * Tests that when a full backup is taken while an RS is offline (with WALs
in oldlogs), the
+ * offline host's timestamps are recorded so subsequent incremental backups
don't reinclude those
+ * WALs.
+ */
+ @Test
+ public void testBackupWithOfflineRS() throws Exception {
+ LOG.info("Starting testBackupWithOfflineRS");
+
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+
+ HRegionServer rsBeforeStop =
cluster.startRegionServerAndWait(10000).getRegionServer();
+ moveRegionAndWait(table1, rsBeforeStop);
+
+ LOG.info("Inserting data to generate WAL entries");
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 2, 100);
+ }
+
+ String offlineHost =
+ rsBeforeStop.getServerName().getHostname() + ":" +
rsBeforeStop.getServerName().getPort();
+ LOG.info("Stopping RS: {}", offlineHost);
+
+ stopRegionServerAndWait(rsBeforeStop.getServerName());
+
+ LOG.info("Taking full backup (with offline RS WALs in oldlogs)");
+ String fullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(fullBackupId), "Full backup should succeed");
+
+ try (BackupSystemTable sysTable = new
BackupSystemTable(TEST_UTIL.getConnection())) {
+ Map<TableName, Map<String, Long>> timestamps =
sysTable.readLogTimestampMap(BACKUP_ROOT_DIR);
+ Map<String, Long> rsTimestamps = timestamps.get(table1);
+ LOG.info("RS timestamps after full backup: {}", rsTimestamps);
+
+ Long tsAfterFullBackup = rsTimestamps.get(offlineHost);
+ assertNotNull(tsAfterFullBackup,
+ "Offline host should have timestamp recorded in trslm after full
backup");
+
+ LOG.info("Taking incremental backup (should NOT include offline RS
WALs)");
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId), "Incremental backup should
succeed");
+
+ assertEquals(tsAfterFullBackup, boundary(sysTable, offlineHost),
+ "Incremental backup moved the boundary of the offline host");
+ }
+ }
+
+ /**
+ * Tests that WALs written to an RS after a full backup are correctly
included in the subsequent
+ * incremental backup, even if that RS has gone offline before the
incremental runs.
+ */
+ @Test
+ public void testRSGoesOfflineAfterFullBackupBeforeIncremental() throws
Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+
+ HRegionServer rsBeforeStop =
cluster.startRegionServerAndWait(10000).getRegionServer();
+
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 3, 50);
+ }
+
+ String fullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(fullBackupId), "Full backup should succeed");
+
+ moveRegionAndWait(table1, rsBeforeStop);
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 4, 50);
+ }
+
+ rsBeforeStop.getWalRoller().requestRollAll();
+ rsBeforeStop.getWalRoller().waitUntilWalRollFinished();
+ String offlineHost =
+ rsBeforeStop.getServerName().getHostname() + ":" +
rsBeforeStop.getServerName().getPort();
+
+ stopRegionServerAndWait(rsBeforeStop.getServerName());
+
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId), "Incremental backup should
succeed");
+
+ restoreAndAssertAllRowsPresent(incrBackupId,
"table1_rs_offline_after_full_backup",
+ "Restored table should contain the rows written to the RS that went
offline");
+
+ try (BackupSystemTable sysTable = new
BackupSystemTable(TEST_UTIL.getConnection())) {
+ Long boundaryAfterIncr1 = boundary(sysTable, offlineHost);
+ assertNotNull(boundaryAfterIncr1,
+ "Offline RS should have a timestamp boundary after the incremental
backed up its WALs");
+ Long staleLogRoll =
+
sysTable.readRegionServerLastLogRollResult(BACKUP_ROOT_DIR).get(offlineHost);
+
+ String incrBackupId2 = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId2), "Second incremental backup
should succeed");
+ Long boundaryAfterIncr2 = boundary(sysTable, offlineHost);
+
+ String incrBackupId3 = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId3), "Third incremental backup
should succeed");
+ Long boundaryAfterIncr3 = boundary(sysTable, offlineHost);
+
+ String incrBackupId4 = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId4), "Fourth incremental backup
should succeed");
+ Long boundaryAfterIncr4 = boundary(sysTable, offlineHost);
+
+ String message = "Incremental backup moved the boundary of the offline
RS, which would back "
Review Comment:
```
String message = "Incremental backup moved the boundary of the offline
RS, which would back "
+ "up its WALs again (last log roll = " + staleLogRoll + ")";
List<String> walsBackedUpAgain = new ArrayList<>();
for (String laterBackupId : List.of(incrBackupId2, incrBackupId3,
incrBackupId4)) {
walsBackedUpAgain.addAll(walsOfHost(sysTable, laterBackupId,
offlineHost));
}
assertAll(() -> assertEquals(boundaryAfterIncr1, boundaryAfterIncr2,
message),
() -> assertEquals(boundaryAfterIncr1, boundaryAfterIncr3, message),
() -> assertEquals(boundaryAfterIncr1, boundaryAfterIncr4, message),
() -> assertTrue(walsBackedUpAgain.isEmpty(),
"Later incremental backups included WALs of the offline RS again:
" + walsBackedUpAgain));
}
```
##########
hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java:
##########
@@ -248,6 +263,75 @@ private void performSnapshots(Admin admin) throws
IOException {
}
}
+ /**
+ * Computes the per-host log boundaries stored by this full backup, using
+ * {@link BackupUtils#computeLogBoundaries}. A host that took part in this
backup's log roll gets
+ * its roll result: WALs created after the roll are included in the next
incremental backup, even
+ * though some of their edits may already be in the snapshot, which is safe
because deletes are
+ * replayed along with the puts. For any other host, its archived WALs and
the WALs of dead
+ * servers are covered by the snapshot, so the next incremental backup does
not replay them. The
+ * WALs of a live server that did not take part in the roll (for example one
that started during
+ * it) are pending, because they can still receive edits that are not in the
snapshot. The live
+ * servers are read only after the WAL directories are listed: a region
server creates its WAL
+ * directory only after the master registers it, so every live server whose
directory was listed
+ * is found. If the live servers do not include every host that took part in
the roll, the
+ * master's server list is incomplete (for example right after a master
failover), and the backup
+ * fails instead of treating live servers as dead. It must run right after
the log roll, before
+ * the snapshot is taken, so that a host starting later gets no boundary and
has all of its WALs
+ * included in the next incremental backup.
+ */
+ @RestrictedApi(
+ explanation = "Package-private for test visibility only. Do not use
outside tests.",
+ link = "",
+ allowedOnPath =
"(.*/src/test/.*|.*/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java)")
+ static Map<String, Long> computeLogBoundaries(FileSystem fs, Path walRootDir,
+ Map<String, Long> rolledHosts, Admin admin) throws IOException {
+ Path logDir = new Path(walRootDir, HConstants.HREGION_LOGDIR_NAME);
+ Path oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
+
+ Map<ServerName, List<String>> logsByServer = new HashMap<>();
+ for (FileStatus serverLogDir : fs.listStatus(logDir)) {
+ ServerName serverName =
+
AbstractFSWALProvider.getServerNameFromWALDirectoryName(serverLogDir.getPath());
+ if (serverName == null) {
+ continue;
+ }
+ List<String> logs = logsByServer.computeIfAbsent(serverName, k -> new
ArrayList<>());
+ for (FileStatus log : fs.listStatus(serverLogDir.getPath())) {
+ if (!AbstractFSWALProvider.isMetaFile(log.getPath())) {
+ logs.add(log.getPath().toString());
+ }
+ }
+ }
+
+ Set<ServerName> live = new HashSet<>(admin.getRegionServers());
+ Set<String> liveAddresses =
+ live.stream().map(sn ->
sn.getAddress().toString()).collect(Collectors.toSet());
+ if (!liveAddresses.containsAll(rolledHosts.keySet())) {
+ throw new IOException("Live region servers " + live + " do not include
every host that took"
+ + " part in this backup's log roll " + rolledHosts.keySet() + ". The
master's server list"
+ + " may be incomplete, for example during a master failover, so the
backup is failed to be"
Review Comment:
`for example during a master failover, or a region server stopped after the
log roll, so the backup is failed to be retried.`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
