You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by hg...@apache.org on 2026/10/15 23:50:53 UTC
(hbase) branch branch-3.0 updated: HBASE-29776 Log filtering in IncrementalBackupManager can lead to data loss (apache#8694) (#8777)
This is an automated email from the ASF dual-hosted git repository.
hgromer pushed a commit to branch branch-3.0
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/branch-3.0 by this push:
new 4cd40325980 HBASE-29776 Log filtering in IncrementalBackupManager can lead to data loss (apache#8694) (#8777)
4cd40325980 is described below
commit 4cd40325980b330aad3675bebee318a19da0d846
Author: Hernan Romer <na...@gmail.com>
AuthorDate: Thu Oct 15 19:50:46 2026 -0400
HBASE-29776 Log filtering in IncrementalBackupManager can lead to data loss (apache#8694) (#8777)
Co-authored-by: Hernan Gelaf-Romer <hg...@hubspot.com>
---
.../hbase/backup/impl/BackupSystemTable.java | 5 +-
.../hbase/backup/impl/FullTableBackupClient.java | 94 ++++-
.../backup/impl/IncrementalBackupManager.java | 64 +--
.../hadoop/hbase/backup/util/BackupUtils.java | 82 ++++
.../hadoop/hbase/backup/TestBackupOfflineRS.java | 464 +++++++++++++++++++++
.../hadoop/hbase/backup/TestBackupUtils.java | 81 ++++
.../hbase/backup/TestIncrementalBackupManager.java | 333 +++++++++++++++
.../TestFullTableBackupClientLogBoundaries.java | 148 +++++++
8 files changed, 1243 insertions(+), 28 deletions(-)
diff --git a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java
index 7a11fe58a01..21c836adb8a 100644
--- a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java
+++ b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/BackupSystemTable.java
@@ -639,8 +639,9 @@ public final class BackupSystemTable implements Closeable {
/**
* Write the current timestamps for each regionserver to backup system table after a successful
- * full or incremental backup. The saved timestamp is of the last log file that was backed up
- * already.
+ * full or incremental backup. For a region server that took part in the backup's log roll, the
+ * saved timestamp is the result of that roll. For a region server that did not (offline or
+ * decommissioned), it is the creation time of the newest of its log files that was backed up.
* @param tables tables
* @param newTimestamps timestamps
* @param backupRoot root directory path to backup
diff --git a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java
index 37aaa30d2b2..ee8f0494849 100644
--- a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java
+++ b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java
@@ -23,10 +23,20 @@ import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_BACK
import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.DEFAULT_BACKUP_MAX_ATTEMPTS;
import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.JOB_NAME_CONF_KEY;
+import com.google.errorprone.annotations.RestrictedApi;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.backup.BackupCopyJob;
import org.apache.hadoop.hbase.backup.BackupInfo;
@@ -38,7 +48,9 @@ import org.apache.hadoop.hbase.backup.BackupType;
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.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -146,9 +158,13 @@ public class FullTableBackupClient extends TableBackupClient {
// snapshots for the same reason as the log rolls.
List<BulkLoad> bulkLoadsToDelete = backupManager.readBulkloadRows(tableList);
+ Map<String, Long> previousLogRollsByHost = backupManager.readRegionServerLastLogRollResult();
BackupUtils.logRoll(conn, backupInfo.getBackupRootDir(), conf);
- newTimestamps = backupManager.readRegionServerLastLogRollResult();
+ Map<String, Long> latestLogRollsByHost = backupManager.readRegionServerLastLogRollResult();
+ Path walRootDir = CommonFSUtils.getWALRootDir(conf);
+ newTimestamps = computeLogBoundaries(walRootDir.getFileSystem(conf), walRootDir,
+ BackupUtils.getRolledHosts(previousLogRollsByHost, latestLogRollsByHost), admin);
// SNAPSHOT_TABLES:
backupInfo.setPhase(BackupPhase.SNAPSHOT);
@@ -216,4 +232,80 @@ public class FullTableBackupClient extends TableBackupClient {
}
throw new IOException("Failed to snapshot table " + tableName);
}
+
+ /**
+ * Computes the per-host log boundaries stored by this full backup, so that the next incremental
+ * backup includes exactly the WALs whose edits may be missing from the snapshot. 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.
+ * <p>
+ * Boundaries are computed by {@link BackupUtils#computeLogBoundaries}:
+ * <ul>
+ * <li>A host that took part in this backup's log roll gets its roll result. WALs created after
+ * the roll will be 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.</li>
+ * <li>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.</li>
+ * <li>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.</li>
+ * </ul>
+ * If the master's server list is incomplete (for example right after a master failover), the
+ * backup fails instead of treating live servers as dead.
+ */
+ @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());
+ }
+ }
+ }
+
+ // 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, or a host stopped after the roll.
+ 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, or a region server stopped"
+ + " after the log roll, so the backup is failed to be retried.");
+ }
+
+ List<String> coveredLogs = new ArrayList<>();
+ List<String> pendingLogs = new ArrayList<>();
+ for (Map.Entry<ServerName, List<String>> entry : logsByServer.entrySet()) {
+ if (live.contains(entry.getKey())) {
+ pendingLogs.addAll(entry.getValue());
+ } else {
+ coveredLogs.addAll(entry.getValue());
+ }
+ }
+
+ if (fs.exists(oldLogDir)) {
+ BackupUtils.getFiles(fs, oldLogDir, coveredLogs,
+ path -> !AbstractFSWALProvider.isMetaFile(path));
+ }
+
+ return BackupUtils.computeLogBoundaries(rolledHosts, coveredLogs, pendingLogs);
+ }
}
diff --git a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalBackupManager.java b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalBackupManager.java
index 2d7aeb646db..00cf6875fc0 100644
--- a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalBackupManager.java
+++ b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalBackupManager.java
@@ -20,8 +20,10 @@ package org.apache.hadoop.hbase.backup.impl;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -53,12 +55,14 @@ public class IncrementalBackupManager extends BackupManager {
/**
* Obtain the list of logs that need to be copied out for this incremental backup. The list is set
* in BackupInfo.
- * @return The new HashMap of RS log time stamps after the log roll for this incremental backup.
+ * @return The new map of RS log time stamps for this incremental backup, as computed by
+ * {@link BackupUtils#computeLogBoundaries}: the included logs are covered, the logs held
+ * back for a later backup (for example, logs still being split) are pending, and a host
+ * that still has logs keeps its previous boundary if this backup gives it no new one.
* @throws IOException exception
*/
public Map<String, Long> getIncrBackupLogFileMap() throws IOException {
List<String> logList;
- Map<String, Long> newTimestamps;
Map<String, Long> previousTimestampMins =
BackupUtils.getRSLogTimestampMins(readLogTimestampMap());
@@ -68,18 +72,28 @@ public class IncrementalBackupManager extends BackupManager {
+ "In order to create an incremental backup, at least one full backup is needed.");
}
+ Map<String, Long> previousLogRollByHost = readRegionServerLastLogRollResult();
LOG.info("Execute roll log procedure for incremental backup ...");
BackupUtils.logRoll(conn, backupInfo.getBackupRootDir(), conf);
- newTimestamps = readRegionServerLastLogRollResult();
+ Map<String, Long> rolledHosts =
+ BackupUtils.getRolledHosts(previousLogRollByHost, readRegionServerLastLogRollResult());
- logList = getLogFilesForNewBackup(previousTimestampMins, newTimestamps, conf);
- logList = excludeProcV2WALs(logList);
+ LogFileSelection selection = getLogFilesForNewBackup(previousTimestampMins, rolledHosts, conf);
+ logList = excludeProcV2WALs(selection.included());
backupInfo.setIncrBackupFileList(logList);
+ Map<String, Long> newTimestamps = BackupUtils.computeLogBoundaries(rolledHosts,
+ previousTimestampMins, selection.hostsWithLogs(), logList, selection.heldBack());
+ LOG.debug("Log boundaries for incremental backup {}: {}", backupInfo.getBackupId(),
+ newTimestamps);
return newTimestamps;
}
+ private record LogFileSelection(List<String> included, List<String> heldBack,
+ Set<String> hostsWithLogs) {
+ }
+
private List<String> excludeProcV2WALs(List<String> logList) {
List<String> list = new ArrayList<>();
for (int i = 0; i < logList.size(); i++) {
@@ -96,15 +110,17 @@ public class IncrementalBackupManager extends BackupManager {
}
/**
- * For each region server: get all log files newer than the last timestamps but not newer than the
- * newest timestamps.
+ * Gather all log files that either: 1) are newer than the older timestamps, but not newer than
+ * the newest timestamps, or 2) are archived logs whose host name does not occur in the newest
+ * timestamps.
* @param olderTimestamps the timestamp for each region server of the last backup.
* @param newestTimestamps the timestamp for each region server that the backup should lead to.
* @param conf the Hadoop and Hbase configuration
- * @return a list of log files to be backed up
+ * @return the log files to be backed up, the log files held back for a later backup, and the
+ * hosts that have any log files, including ones not backed up
* @throws IOException exception
*/
- private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
+ private LogFileSelection getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
Map<String, Long> newestTimestamps, Configuration conf) throws IOException {
LOG.debug("In getLogFilesForNewBackup()\n" + "olderTimestamps: " + olderTimestamps
+ "\n newestTimestamps: " + newestTimestamps);
@@ -118,6 +134,7 @@ public class IncrementalBackupManager extends BackupManager {
List<String> resultLogFiles = new ArrayList<>();
List<String> newestLogs = new ArrayList<>();
+ Set<String> hostsWithLogs = new HashSet<>();
/*
* The old region servers and timestamps info we kept in backup system table may be out of sync
@@ -142,6 +159,7 @@ public class IncrementalBackupManager extends BackupManager {
if (host == null) {
continue;
}
+ hostsWithLogs.add(host);
FileStatus[] logs;
oldTimeStamp = olderTimestamps.get(host);
// It is possible that there is no old timestamp in backup system table for this host if
@@ -164,17 +182,15 @@ public class IncrementalBackupManager extends BackupManager {
resultLogFiles.add(currentLogFile);
currentLogTS = BackupUtils.getCreationTime(log.getPath());
- // If newestTimestamps.get(host) is null, means that
- // either RS (host) has been restarted recently with different port number
- // or RS is down (was decommisioned). In any case, we treat this
- // log file as eligible for inclusion into incremental backup log list
+ // 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.warn("ORPHAN log found: " + log + " host=" + host);
- LOG.debug("Known hosts (from newestTimestamps):");
- for (String s : newestTimestamps.keySet()) {
- LOG.debug(s);
- }
+ LOG.debug("Holding back log {} of host {}, which did not take part in the log roll",
+ log.getPath(), host);
}
if (ts == null || currentLogTS > ts) {
newestLogs.add(currentLogFile);
@@ -197,6 +213,7 @@ public class IncrementalBackupManager extends BackupManager {
if (host == null) {
continue;
}
+ hostsWithLogs.add(host);
currentLogTS = BackupUtils.getCreationTime(p);
oldTimeStamp = olderTimestamps.get(host);
/*
@@ -216,18 +233,15 @@ public class IncrementalBackupManager extends BackupManager {
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
+ // 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.
resultLogFiles.removeAll(newestLogs);
- return resultLogFiles;
+ return new LogFileSelection(resultLogFiles, newestLogs, hostsWithLogs);
}
static class NewestLogFilter implements PathFilter {
diff --git a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/util/BackupUtils.java b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/util/BackupUtils.java
index 138426d9079..ece35c74c01 100644
--- a/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/util/BackupUtils.java
+++ b/hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/util/BackupUtils.java
@@ -21,12 +21,14 @@ import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URLDecoder;
import java.util.ArrayList;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.Set;
import java.util.TreeSet;
import java.util.function.Predicate;
import java.util.stream.Collectors;
@@ -186,6 +188,86 @@ public final class BackupUtils {
}
}
+ /**
+ * 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. 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 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 included in the backup
+ * @param pendingLogs logs whose edits are not to be included in the backup
+ * @return boundaries by host
+ */
+ public static Map<String, Long> computeLogBoundaries(Map<String, Long> rolledHosts,
+ Map<String, Long> previousBoundaries, Set<String> hostsWithLogs, Collection<String> coveredLogs,
+ Collection<String> pendingLogs) throws IOException {
+ Map<String, Long> otherHosts = new HashMap<>();
+ for (String log : coveredLogs) {
+ Path path = new Path(log);
+ String host = parseHostNameFromLogFile(path);
+ if (host != null && !rolledHosts.containsKey(host)) {
+ otherHosts.merge(host, getCreationTime(path), Math::max);
+ }
+ }
+ for (String log : pendingLogs) {
+ Path path = new Path(log);
+ String host = parseHostNameFromLogFile(path);
+ if (host != null && !rolledHosts.containsKey(host)) {
+ otherHosts.merge(host, getCreationTime(path) - 1, Math::min);
+ }
+ }
+ Map<String, Long> boundaries = new HashMap<>(rolledHosts);
+ boundaries.putAll(otherHosts);
+ for (Entry<String, Long> previous : previousBoundaries.entrySet()) {
+ if (hostsWithLogs.contains(previous.getKey())) {
+ boundaries.putIfAbsent(previous.getKey(), previous.getValue());
+ }
+ }
+ return boundaries;
+ }
+
/**
* Returns WAL file name
* @param walFileName WAL file name
diff --git a/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupOfflineRS.java b/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupOfflineRS.java
new file mode 100644
index 00000000000..c45078e1045
--- /dev/null
+++ b/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupOfflineRS.java
@@ -0,0 +1,464 @@
+/*
+ * 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.assertFalse;
+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.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.fs.Path;
+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 Exception {
+ 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 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();
+ }
+
+ 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");
+ List<String> offlineHostWals = walsOfHost(sysTable, incrBackupId, offlineHost);
+ assertTrue(offlineHostWals.isEmpty(),
+ "Incremental backup included WALs of the offline host that the full backup covers: "
+ + offlineHostWals);
+ }
+ }
+
+ /**
+ * 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");
+ assertFalse(walsOfHost(sysTable, incrBackupId, offlineHost).isEmpty(),
+ "First incremental after the RS went offline should include 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 "
+ + "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));
+ }
+ }
+
+ /**
+ * Tests that a brand-new RS that comes online and goes offline before any backup correctly has
+ * its WALs covered by the full backup.
+ */
+ @Test
+ public void testTransientRSBeforeFullBackup() throws Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+
+ HRegionServer transientRS = cluster.startRegionServerAndWait(10000).getRegionServer();
+ moveRegionAndWait(table1, transientRS);
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 5, 50);
+ }
+ transientRS.getWalRoller().requestRollAll();
+ transientRS.getWalRoller().waitUntilWalRollFinished();
+ String transientHost =
+ transientRS.getServerName().getHostname() + ":" + transientRS.getServerName().getPort();
+
+ stopRegionServerAndWait(transientRS.getServerName());
+
+ String fullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(fullBackupId), "Full backup should succeed");
+
+ try (BackupSystemTable sysTable = new BackupSystemTable(TEST_UTIL.getConnection())) {
+ Long boundaryAfterFullBackup = boundary(sysTable, transientHost);
+ assertNotNull(boundaryAfterFullBackup,
+ "Transient RS that went offline before full backup should have its WAL boundary recorded");
+
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId),
+ "Incremental backup after transient RS should succeed");
+
+ assertEquals(boundaryAfterFullBackup, boundary(sysTable, transientHost),
+ "Incremental backup moved the boundary of the transient RS");
+ }
+ }
+
+ /**
+ * Tests that WALs from an RS that comes online and goes offline between a full backup and an
+ * incremental backup are correctly included in the incremental backup and not re-included in
+ * subsequent incremental backups.
+ */
+ @Test
+ public void testTransientRSAfterFullBackupBeforeIncremental() throws Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 6, 50);
+ }
+
+ String fullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(fullBackupId), "Full backup should succeed");
+
+ HRegionServer transientRS = cluster.startRegionServerAndWait(10000).getRegionServer();
+ moveRegionAndWait(table1, transientRS);
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 7, 50);
+ }
+ transientRS.getWalRoller().requestRollAll();
+ transientRS.getWalRoller().waitUntilWalRollFinished();
+ String transientHost =
+ transientRS.getServerName().getHostname() + ":" + transientRS.getServerName().getPort();
+
+ stopRegionServerAndWait(transientRS.getServerName());
+
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId), "Incremental backup should succeed");
+
+ restoreAndAssertAllRowsPresent(incrBackupId, "table1_transient_rs_after_full_backup",
+ "Restored table should contain the rows written to the transient RS");
+
+ try (BackupSystemTable sysTable = new BackupSystemTable(TEST_UTIL.getConnection())) {
+ Long boundaryAfterIncr1 = boundary(sysTable, transientHost);
+ assertNotNull(boundaryAfterIncr1,
+ "Transient RS should have a timestamp boundary after the incremental backed up its WALs");
+
+ String incrBackupId2 = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId2), "Second incremental backup should succeed");
+
+ assertEquals(boundaryAfterIncr1, boundary(sysTable, transientHost),
+ "Second incremental backup moved the boundary of the transient RS");
+ }
+ }
+
+ /**
+ * Tests that edits written during a full backup, after its log roll and table snapshot, to an RS
+ * that started after that log roll, are included in the next incremental backup.
+ */
+ @Test
+ public void testRSStartedDuringFullBackup() throws Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+
+ afterSnapshotHook = () -> {
+ try {
+ HRegionServer newRS = cluster.startRegionServerAndWait(10000).getRegionServer();
+ moveRegionAndWait(table1, newRS);
+ try (Connection conn = ConnectionFactory.createConnection(conf1)) {
+ insertIntoTable(conn, table1, famName, 8, 50).close();
+ }
+ stopRegionServerAndWait(newRS.getServerName());
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ };
+ conf1.set(TableBackupClient.BACKUP_CLIENT_IMPL_CLASS,
+ FullTableBackupClientWithHook.class.getName());
+ String fullBackupId;
+ try {
+ fullBackupId = fullTableBackup(tables);
+ } finally {
+ conf1.unset(TableBackupClient.BACKUP_CLIENT_IMPL_CLASS);
+ }
+ assertTrue(checkSucceeded(fullBackupId), "Full backup should succeed");
+
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId), "Incremental backup should succeed");
+
+ restoreAndAssertAllRowsPresent(incrBackupId, "table1_rs_started_during_full_backup",
+ "Restored table should contain all rows, including those written to the new RS");
+ }
+
+ /**
+ * Tests that a row deleted before a full backup is not brought back by the WALs of a region
+ * server that went offline between two backups. The row and its delete marker are compacted away
+ * before the full backup, so only the offline RS's old WAL, which still holds the put, could
+ * bring it back.
+ */
+ @Test
+ public void testDeletedRowIsNotResurrectedByOfflineRSWALs() throws Exception {
+ SingleProcessHBaseCluster cluster = TEST_UTIL.getMiniHBaseCluster();
+ List<TableName> tables = Lists.newArrayList(table1);
+ byte[] deletedRow = Bytes.toBytes("row-deleted");
+
+ HRegionServer offlineRS = cluster.startRegionServerAndWait(10000).getRegionServer();
+ HRegionServer liveRS = cluster.startRegionServerAndWait(10000).getRegionServer();
+
+ try (Table table = TEST_UTIL.getConnection().getTable(table1)) {
+ moveRegionAndWait(table1, offlineRS);
+
+ String firstFullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(firstFullBackupId), "First full backup should succeed");
+
+ for (int i = 0; i < 10; i++) {
+ table.put(new Put(Bytes.toBytes("row-kept-" + i)).addColumn(famName, qualName,
+ Bytes.toBytes("value")));
+ }
+ table.put(new Put(deletedRow).addColumn(famName, qualName, Bytes.toBytes("value")));
+ for (HRegion region : offlineRS.getRegions(table1)) {
+ region.flush(true);
+ }
+
+ moveRegionAndWait(table1, liveRS);
+
+ table.delete(new Delete(deletedRow));
+ for (HRegion region : liveRS.getRegions(table1)) {
+ region.flush(true);
+ region.compact(true);
+ }
+
+ assertTrue(table.get(new Get(deletedRow)).isEmpty(),
+ "Deleted row should be gone from the source table");
+ Scan rawScan = new Scan().withStartRow(deletedRow).withStopRow(deletedRow, true).setRaw(true);
+ try (ResultScanner scanner = table.getScanner(rawScan)) {
+ assertNull(scanner.next(),
+ "Major compaction should have purged the deleted row and its delete marker");
+ }
+
+ stopRegionServerAndWait(offlineRS.getServerName());
+
+ String secondFullBackupId = fullTableBackup(tables);
+ assertTrue(checkSucceeded(secondFullBackupId), "Second full backup should succeed");
+ String incrBackupId = incrementalTableBackup(tables);
+ assertTrue(checkSucceeded(incrBackupId), "Incremental backup should succeed");
+
+ TableName restoredTable = TableName.valueOf("table1_deleted_row_not_resurrected");
+ restore(incrBackupId, table1, restoredTable);
+ try (Table restored = TEST_UTIL.getConnection().getTable(restoredTable)) {
+ assertTrue(restored.get(new Get(deletedRow)).isEmpty(),
+ "Deleted row was brought back by the WALs of the offline RS");
+ }
+ assertEquals(TEST_UTIL.countRows(table1), TEST_UTIL.countRows(restoredTable),
+ "Restored table should match the source table");
+ }
+ }
+}
diff --git a/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupUtils.java b/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupUtils.java
index cf69476270c..073bd6a90d1 100644
--- a/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupUtils.java
+++ b/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestBackupUtils.java
@@ -23,6 +23,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import java.io.IOException;
import java.security.PrivilegedAction;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -114,4 +117,82 @@ public class TestBackupUtils {
}
}
+
+ @Test
+ public void testGetRolledHostsKeepsOnlyHostsWhoseRollResultChanged() {
+ Map<String, Long> previousLogRolls =
+ Map.of("rolled:16020", 100L, "offline:16020", 200L, "removed:16020", 300L);
+ Map<String, Long> latestLogRolls =
+ Map.of("rolled:16020", 150L, "offline:16020", 200L, "new:16020", 400L);
+
+ assertEquals(Map.of("rolled:16020", 150L, "new:16020", 400L),
+ BackupUtils.getRolledHosts(previousLogRolls, latestLogRolls));
+ }
+
+ @Test
+ public void testComputeLogBoundariesUsesRollResultForRolledHosts() throws IOException {
+ Map<String, Long> rolledHosts = Map.of("rolled:16020", 100L);
+ List<String> coveredLogs = List.of("/hbase/oldWALs/rolled%2C16020%2C1.500");
+ List<String> pendingLogs = List.of("/hbase/WALs/rolled,16020,1/rolled%2C16020%2C1.50");
+
+ assertEquals(Map.of("rolled:16020", 100L),
+ BackupUtils.computeLogBoundaries(rolledHosts, coveredLogs, pendingLogs));
+ }
+
+ @Test
+ public void testComputeLogBoundariesUsesNewestCoveredLogForOtherHosts() throws IOException {
+ List<String> coveredLogs =
+ List.of("/hbase/oldWALs/offline%2C16020%2C1.200", "/hbase/oldWALs/offline%2C16020%2C1.300",
+ "/hbase/WALs/offline,16020,1/offline%2C16020%2C1.250");
+
+ assertEquals(Map.of("offline:16020", 300L),
+ BackupUtils.computeLogBoundaries(Map.of(), coveredLogs, List.of()));
+ }
+
+ @Test
+ public void testComputeLogBoundariesCapsBelowOldestPendingLog() throws IOException {
+ List<String> coveredLogs = List.of("/hbase/oldWALs/splitting%2C16020%2C1.400");
+ List<String> pendingLogs =
+ List.of("/hbase/WALs/splitting,16020,1-splitting/splitting%2C16020%2C1.350",
+ "/hbase/WALs/splitting,16020,1-splitting/splitting%2C16020%2C1.370");
+
+ assertEquals(Map.of("splitting:16020", 349L),
+ BackupUtils.computeLogBoundaries(Map.of(), coveredLogs, pendingLogs));
+ }
+
+ @Test
+ public void testComputeLogBoundariesKeepsHostWithOnlyPendingLogs() throws IOException {
+ List<String> pendingLogs = List.of("/hbase/WALs/joined,16020,1/joined%2C16020%2C1.500");
+
+ assertEquals(Map.of("joined:16020", 499L),
+ BackupUtils.computeLogBoundaries(Map.of(), List.of(), pendingLogs));
+ }
+
+ @Test
+ public void testComputeLogBoundariesSkipsUnparseableLogs() throws IOException {
+ List<String> coveredLogs = List.of("/hbase/oldWALs/not-a-wal");
+
+ assertEquals(Map.of("rolled:16020", 100L),
+ BackupUtils.computeLogBoundaries(Map.of("rolled:16020", 100L), coveredLogs, List.of()));
+ }
+
+ @Test
+ public void testComputeLogBoundariesKeepsPreviousBoundaryWhileHostHasLogs() throws IOException {
+ Map<String, Long> previousBoundaries = Map.of("stuck:16020", 1200L, "gone:16020", 800L);
+
+ assertEquals(Map.of("stuck:16020", 1200L), BackupUtils.computeLogBoundaries(Map.of(),
+ previousBoundaries, Set.of("stuck:16020"), List.of(), List.of()));
+ }
+
+ @Test
+ public void testComputeLogBoundariesPrefersNewBoundaryOverPreviousOne() throws IOException {
+ Map<String, Long> previousBoundaries =
+ Map.of("rolled:16020", 100L, "offline:16020", 200L, "pending:16020", 300L);
+ List<String> coveredLogs = List.of("/hbase/oldWALs/offline%2C16020%2C1.250");
+ List<String> pendingLogs = List.of("/hbase/WALs/pending,16020,1/pending%2C16020%2C1.400");
+
+ assertEquals(Map.of("rolled:16020", 150L, "offline:16020", 250L, "pending:16020", 399L),
+ BackupUtils.computeLogBoundaries(Map.of("rolled:16020", 150L), previousBoundaries,
+ Set.of("rolled:16020", "offline:16020", "pending:16020"), coveredLogs, pendingLogs));
+ }
}
diff --git a/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestIncrementalBackupManager.java b/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestIncrementalBackupManager.java
new file mode 100644
index 00000000000..023f535cfed
--- /dev/null
+++ b/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestIncrementalBackupManager.java
@@ -0,0 +1,333 @@
+/*
+ * 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.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.backup.impl.BackupAdminImpl;
+import org.apache.hadoop.hbase.backup.impl.IncrementalBackupManager;
+import org.apache.hadoop.hbase.backup.util.BackupUtils;
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
+import org.apache.hadoop.hbase.testclassification.LargeTests;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+import org.apache.hadoop.hbase.util.JVMClusterUtil;
+import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+@Tag(LargeTests.TAG)
+public class TestIncrementalBackupManager extends TestBackupBase {
+
+ @BeforeAll
+ public static void setUp() throws Exception {
+ TEST_UTIL = new HBaseTestingUtil();
+ conf1 = TEST_UTIL.getConfiguration();
+ autoRestoreOnFailure = true;
+ useSecondCluster = false;
+ setUpHelper();
+ }
+
+ /**
+ * WALs can be archived out of order, so a region server that took part in the log roll can have
+ * an archived WAL newer than its roll result while an older WAL is still in the WALs directory.
+ * The newer archived WAL must be deferred to a later backup instead of being backed up twice, and
+ * the older WAL must still end up in a backup.
+ */
+ @Test
+ public void testOutOfOrderArchivedWALDoesNotSkipOlderWAL() throws Exception {
+ List<TableName> tables = List.of(table1);
+ HRegionServer rs = TEST_UTIL.getMiniHBaseCluster().getRegionServer(0);
+ ServerName serverName = rs.getServerName();
+ Path walRootDir = CommonFSUtils.getWALRootDir(conf1);
+ FileSystem fs = walRootDir.getFileSystem(conf1);
+
+ try (Connection conn = ConnectionFactory.createConnection(conf1);
+ BackupAdminImpl backupAdmin = new BackupAdminImpl(conn)) {
+ String fullBackupId =
+ backupAdmin.backupTables(createBackupRequest(BackupType.FULL, tables, BACKUP_ROOT_DIR));
+ assertTrue(checkSucceeded(fullBackupId));
+
+ long olderWALTs = EnvironmentEdgeManager.currentTime() + 1;
+ long newerWALTs = olderWALTs + 1;
+ Path walDir =
+ new Path(walRootDir, AbstractFSWALProvider.getWALDirectoryName(serverName.toString()));
+ Path olderWAL =
+ new Path(walDir, serverName.toString() + BackupUtils.LOGNAME_SEPARATOR + olderWALTs);
+ Path archiveDir = new Path(walRootDir,
+ AbstractFSWALProvider.getWALArchiveDirectoryName(conf1, serverName.toString()));
+ Path newerArchivedWAL =
+ new Path(archiveDir, serverName.toString() + BackupUtils.LOGNAME_SEPARATOR + newerWALTs);
+ fs.create(olderWAL).close();
+ fs.mkdirs(archiveDir);
+ fs.create(newerArchivedWAL).close();
+
+ try {
+ List<String> firstBackupFiles;
+ try (IncrementalBackupManager manager = new IncrementalBackupManager(conn, conf1)) {
+ BackupInfo backupInfo = manager.createBackupInfo("backup_incr_1", BackupType.INCREMENTAL,
+ tables, BACKUP_ROOT_DIR, -1, -1, false);
+ Map<String, Long> boundaries = manager.getIncrBackupLogFileMap();
+ manager.writeRegionServerLogTimestamp(backupInfo.getTables(), boundaries);
+ firstBackupFiles = backupInfo.getIncrBackupFileList();
+ }
+ assertFalse(firstBackupFiles.contains(newerArchivedWAL.toString()),
+ "Archived WAL newer than the roll result should be deferred to a later backup: "
+ + firstBackupFiles);
+
+ TEST_UTIL.waitFor(30_000, () -> EnvironmentEdgeManager.currentTime() > newerWALTs);
+ rs.getWalRoller().requestRollAll();
+ rs.getWalRoller().waitUntilWalRollFinished();
+
+ List<String> secondBackupFiles;
+ try (IncrementalBackupManager manager = new IncrementalBackupManager(conn, conf1)) {
+ BackupInfo backupInfo = manager.createBackupInfo("backup_incr_2", BackupType.INCREMENTAL,
+ tables, BACKUP_ROOT_DIR, -1, -1, false);
+ manager.getIncrBackupLogFileMap();
+ secondBackupFiles = backupInfo.getIncrBackupFileList();
+ }
+
+ assertTrue(
+ firstBackupFiles.contains(olderWAL.toString())
+ || secondBackupFiles.contains(olderWAL.toString()),
+ "WAL " + olderWAL + " was not included in any backup. First backup: " + firstBackupFiles
+ + ", second backup: " + secondBackupFiles);
+ } finally {
+ fs.delete(olderWAL, false);
+ fs.delete(newerArchivedWAL, false);
+ }
+ }
+ }
+
+ /**
+ * A dead region server's WALs stay in its -splitting directory until WAL splitting finishes. An
+ * incremental backup that runs during the split must not lose them once they are archived.
+ */
+ @Test
+ public void testWALOfDeadServerStillSplittingIsBackedUpAfterArchiving() throws Exception {
+ List<TableName> tables = List.of(table1);
+ ServerName deadServer = ServerName.valueOf("deadhost", 16020, 1001L);
+ Path walRootDir = CommonFSUtils.getWALRootDir(conf1);
+ FileSystem fs = walRootDir.getFileSystem(conf1);
+ Path splittingDir = splittingDir(walRootDir, deadServer);
+ Path archiveDir = new Path(walRootDir,
+ AbstractFSWALProvider.getWALArchiveDirectoryName(conf1, deadServer.toString()));
+
+ try (Connection conn = ConnectionFactory.createConnection(conf1);
+ BackupAdminImpl backupAdmin = new BackupAdminImpl(conn)) {
+ String fullBackupId =
+ backupAdmin.backupTables(createBackupRequest(BackupType.FULL, tables, BACKUP_ROOT_DIR));
+ assertTrue(checkSucceeded(fullBackupId));
+
+ long walTs = EnvironmentEdgeManager.currentTime() + 1;
+ Path splittingWAL = new Path(splittingDir, walName(deadServer, walTs));
+ Path archivedWAL = archivedWAL(archiveDir, deadServer, walTs);
+ fs.create(splittingWAL).close();
+
+ try {
+ // Make the live servers' roll results, and so the next boundaries, newer than the WAL.
+ TEST_UTIL.waitFor(30_000, () -> EnvironmentEdgeManager.currentTime() > walTs);
+ rollAllLiveRegionServers();
+
+ List<String> firstBackupFiles = runIncrementalBackup(conn, tables, "backup_split_1");
+
+ fs.mkdirs(archiveDir);
+ assertTrue(fs.rename(splittingWAL, archivedWAL));
+ fs.delete(splittingDir, true);
+
+ List<String> secondBackupFiles = runIncrementalBackup(conn, tables, "backup_split_2");
+
+ assertTrue(
+ firstBackupFiles.contains(splittingWAL.toString())
+ || secondBackupFiles.contains(archivedWAL.toString()),
+ "WAL of the dead server was not included in any backup. First backup: " + firstBackupFiles
+ + ", second backup: " + secondBackupFiles);
+ } finally {
+ fs.delete(splittingDir, true);
+ fs.delete(archivedWAL, false);
+ }
+ }
+ }
+
+ /**
+ * WAL splitting archives a dead region server's WALs in parallel, so a newer WAL can be archived
+ * while older ones are still in the -splitting directory. Backing up the newer WAL must not move
+ * the boundary past the older ones.
+ */
+ @Test
+ public void testOlderWALsOfDeadServerAreNotSkippedWhenNewerWALIsArchivedFirst() throws Exception {
+ List<TableName> tables = List.of(table1);
+ ServerName deadServer = ServerName.valueOf("deadhost", 16020, 2002L);
+ Path walRootDir = CommonFSUtils.getWALRootDir(conf1);
+ FileSystem fs = walRootDir.getFileSystem(conf1);
+ Path splittingDir = splittingDir(walRootDir, deadServer);
+ Path archiveDir = new Path(walRootDir,
+ AbstractFSWALProvider.getWALArchiveDirectoryName(conf1, deadServer.toString()));
+
+ try (Connection conn = ConnectionFactory.createConnection(conf1);
+ BackupAdminImpl backupAdmin = new BackupAdminImpl(conn)) {
+ String fullBackupId =
+ backupAdmin.backupTables(createBackupRequest(BackupType.FULL, tables, BACKUP_ROOT_DIR));
+ assertTrue(checkSucceeded(fullBackupId));
+
+ long oldestTs = EnvironmentEdgeManager.currentTime() + 1;
+ long middleTs = oldestTs + 1;
+ long newestTs = oldestTs + 2;
+ Path oldestSplittingWAL = new Path(splittingDir, walName(deadServer, oldestTs));
+ Path middleSplittingWAL = new Path(splittingDir, walName(deadServer, middleTs));
+ Path oldestArchivedWAL = archivedWAL(archiveDir, deadServer, oldestTs);
+ Path middleArchivedWAL = archivedWAL(archiveDir, deadServer, middleTs);
+ Path newestArchivedWAL = archivedWAL(archiveDir, deadServer, newestTs);
+ fs.create(oldestSplittingWAL).close();
+ fs.create(middleSplittingWAL).close();
+ fs.mkdirs(archiveDir);
+ fs.create(newestArchivedWAL).close();
+
+ try {
+ TEST_UTIL.waitFor(30_000, () -> EnvironmentEdgeManager.currentTime() > newestTs);
+
+ List<String> firstBackupFiles = runIncrementalBackup(conn, tables, "backup_order_1");
+
+ assertTrue(fs.rename(oldestSplittingWAL, oldestArchivedWAL));
+ assertTrue(fs.rename(middleSplittingWAL, middleArchivedWAL));
+ fs.delete(splittingDir, true);
+
+ List<String> secondBackupFiles = runIncrementalBackup(conn, tables, "backup_order_2");
+
+ assertAll(
+ () -> assertTrue(
+ firstBackupFiles.contains(oldestSplittingWAL.toString())
+ || secondBackupFiles.contains(oldestArchivedWAL.toString()),
+ "Oldest WAL of the dead server was not included in any backup. First backup: "
+ + firstBackupFiles + ", second backup: " + secondBackupFiles),
+ () -> assertTrue(
+ firstBackupFiles.contains(middleSplittingWAL.toString())
+ || secondBackupFiles.contains(middleArchivedWAL.toString()),
+ "Middle WAL of the dead server was not included in any backup. First backup: "
+ + firstBackupFiles + ", second backup: " + secondBackupFiles));
+ } finally {
+ fs.delete(splittingDir, true);
+ fs.delete(oldestArchivedWAL, false);
+ fs.delete(middleArchivedWAL, false);
+ fs.delete(newestArchivedWAL, false);
+ }
+ }
+ }
+
+ /**
+ * A dead region server can keep an old WAL, already covered by its boundary, in its -splitting
+ * directory across several incremental backups. Its boundary must survive those backups, or a
+ * later backup would include that WAL again and could bring back deleted data.
+ */
+ @Test
+ public void testDeadServerKeepsBoundaryWhileOldWALIsStillSplitting() throws Exception {
+ List<TableName> tables = List.of(table1);
+ ServerName deadServer = ServerName.valueOf("deadhost", 16020, 3003L);
+ Path walRootDir = CommonFSUtils.getWALRootDir(conf1);
+ FileSystem fs = walRootDir.getFileSystem(conf1);
+ Path splittingDir = splittingDir(walRootDir, deadServer);
+ Path archiveDir = new Path(walRootDir,
+ AbstractFSWALProvider.getWALArchiveDirectoryName(conf1, deadServer.toString()));
+
+ try (Connection conn = ConnectionFactory.createConnection(conf1);
+ BackupAdminImpl backupAdmin = new BackupAdminImpl(conn)) {
+ String fullBackupId =
+ backupAdmin.backupTables(createBackupRequest(BackupType.FULL, tables, BACKUP_ROOT_DIR));
+ assertTrue(checkSucceeded(fullBackupId));
+
+ long stuckTs = EnvironmentEdgeManager.currentTime() + 1;
+ long archivedTs = stuckTs + 1;
+ Path archivedWAL = archivedWAL(archiveDir, deadServer, archivedTs);
+ Path stuckSplittingWAL = new Path(splittingDir, walName(deadServer, stuckTs));
+ Path stuckArchivedWAL = archivedWAL(archiveDir, deadServer, stuckTs);
+ fs.mkdirs(archiveDir);
+ fs.create(archivedWAL).close();
+
+ try {
+ TEST_UTIL.waitFor(30_000, () -> EnvironmentEdgeManager.currentTime() > archivedTs);
+ List<String> firstBackupFiles = runIncrementalBackup(conn, tables, "backup_stuck_1");
+ assertTrue(firstBackupFiles.contains(archivedWAL.toString()),
+ "Archived WAL of the dead server should be in the first backup: " + firstBackupFiles);
+
+ fs.create(stuckSplittingWAL).close();
+ List<String> laterBackupFiles = new ArrayList<>();
+ laterBackupFiles.addAll(runIncrementalBackup(conn, tables, "backup_stuck_2"));
+ laterBackupFiles.addAll(runIncrementalBackup(conn, tables, "backup_stuck_3"));
+
+ assertTrue(fs.rename(stuckSplittingWAL, stuckArchivedWAL));
+ fs.delete(splittingDir, true);
+ laterBackupFiles.addAll(runIncrementalBackup(conn, tables, "backup_stuck_4"));
+
+ assertFalse(
+ laterBackupFiles.contains(stuckSplittingWAL.toString())
+ || laterBackupFiles.contains(stuckArchivedWAL.toString()),
+ "WAL already covered by the dead server's boundary was backed up again: "
+ + laterBackupFiles);
+ } finally {
+ fs.delete(splittingDir, true);
+ fs.delete(archivedWAL, false);
+ fs.delete(stuckArchivedWAL, false);
+ }
+ }
+ }
+
+ private static Path splittingDir(Path walRootDir, ServerName serverName) {
+ return new Path(walRootDir, AbstractFSWALProvider.getWALDirectoryName(serverName.toString())
+ + AbstractFSWALProvider.SPLITTING_EXT);
+ }
+
+ private static Path archivedWAL(Path archiveDir, ServerName serverName, long ts) {
+ return new Path(archiveDir, walName(serverName, ts));
+ }
+
+ private static String walName(ServerName serverName, long ts) {
+ return serverName.toString() + BackupUtils.LOGNAME_SEPARATOR + ts;
+ }
+
+ private static void rollAllLiveRegionServers() throws Exception {
+ for (JVMClusterUtil.RegionServerThread rst : TEST_UTIL.getMiniHBaseCluster()
+ .getLiveRegionServerThreads()) {
+ rst.getRegionServer().getWalRoller().requestRollAll();
+ rst.getRegionServer().getWalRoller().waitUntilWalRollFinished();
+ }
+ }
+
+ private static List<String> runIncrementalBackup(Connection conn, List<TableName> tables,
+ String backupId) throws Exception {
+ try (IncrementalBackupManager manager = new IncrementalBackupManager(conn, conf1)) {
+ BackupInfo backupInfo = manager.createBackupInfo(backupId, BackupType.INCREMENTAL, tables,
+ BACKUP_ROOT_DIR, -1, -1, false);
+ Map<String, Long> boundaries = manager.getIncrBackupLogFileMap();
+ manager.writeRegionServerLogTimestamp(backupInfo.getTables(), boundaries);
+ return backupInfo.getIncrBackupFileList();
+ }
+ }
+}
diff --git a/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/impl/TestFullTableBackupClientLogBoundaries.java b/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/impl/TestFullTableBackupClientLogBoundaries.java
new file mode 100644
index 00000000000..91aaa76c642
--- /dev/null
+++ b/hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/impl/TestFullTableBackupClientLogBoundaries.java
@@ -0,0 +1,148 @@
+/*
+ * 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.impl;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.ServerName;
+import org.apache.hadoop.hbase.client.Admin;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.Test;
+
+@Tag(SmallTests.TAG)
+public class TestFullTableBackupClientLogBoundaries {
+
+ private static final HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
+
+ private FileSystem fs;
+ private Path walRootDir;
+
+ @BeforeEach
+ public void setUp() throws IOException {
+ fs = TEST_UTIL.getTestFileSystem();
+ walRootDir = TEST_UTIL.getDataTestDirOnTestFS("walRoot");
+ }
+
+ @AfterEach
+ public void tearDown() throws IOException {
+ fs.delete(walRootDir, true);
+ }
+
+ @Test
+ public void testLiveServerWithoutRollResultIsCappedBelowItsOldestWAL() throws IOException {
+ ServerName joined = ServerName.valueOf("joined", 16020, 1L);
+ createWAL(walDir(joined), joined, 500);
+ createWAL(walDir(joined), joined, 600);
+ Path metaWAL =
+ new Path(walDir(joined), walName(joined, 450) + AbstractFSWALProvider.META_WAL_PROVIDER_ID);
+ fs.create(metaWAL).close();
+
+ assertEquals(Map.of("joined:16020", 499L), computeLogBoundaries(Map.of(), Set.of(joined)));
+ }
+
+ @Test
+ public void testDeadServerGetsItsNewestWAL() throws IOException {
+ ServerName dead = ServerName.valueOf("dead", 16020, 2L);
+ createWAL(new Path(walDir(dead).toString() + AbstractFSWALProvider.SPLITTING_EXT), dead, 300);
+ createWAL(new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME), dead, 350);
+
+ assertEquals(Map.of("dead:16020", 350L), computeLogBoundaries(Map.of(), Set.of()));
+ }
+
+ @Test
+ public void testRestartedServerCoversOldInstanceOnly() throws IOException {
+ ServerName oldInstance = ServerName.valueOf("restarted", 16020, 3L);
+ ServerName newInstance = ServerName.valueOf("restarted", 16020, 4L);
+ createWAL(walDir(oldInstance), oldInstance, 700);
+ createWAL(walDir(newInstance), newInstance, 800);
+
+ assertEquals(Map.of("restarted:16020", 700L),
+ computeLogBoundaries(Map.of(), Set.of(newInstance)));
+ }
+
+ @Test
+ public void testRolledServerKeepsItsRollResult() throws IOException {
+ ServerName rolled = ServerName.valueOf("rolled", 16020, 5L);
+ createWAL(walDir(rolled), rolled, 900);
+ createWAL(new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME), rolled, 950);
+
+ assertEquals(Map.of("rolled:16020", 850L),
+ computeLogBoundaries(Map.of("rolled:16020", 850L), Set.of(rolled)));
+ }
+
+ @Test
+ public void testServerRegisteringAfterTheListingGetsNoBoundary() throws IOException {
+ ServerName joined = ServerName.valueOf("joined", 16020, 6L);
+ ServerName late = ServerName.valueOf("late", 16020, 7L);
+ createWAL(walDir(joined), joined, 500);
+ Admin admin = mock(Admin.class);
+ when(admin.getRegionServers()).thenAnswer(invocation -> {
+ createWAL(walDir(late), late, 600);
+ return List.of(joined);
+ });
+
+ assertEquals(Map.of("joined:16020", 499L),
+ FullTableBackupClient.computeLogBoundaries(fs, walRootDir, Map.of(), admin));
+ }
+
+ @Test
+ public void testIncompleteLiveServerListFailsTheBackup() throws IOException {
+ ServerName rolled = ServerName.valueOf("rolled", 16020, 8L);
+ ServerName joined = ServerName.valueOf("joined", 16020, 9L);
+ createWAL(walDir(rolled), rolled, 900);
+ createWAL(walDir(joined), joined, 500);
+
+ assertThrows(IOException.class,
+ () -> computeLogBoundaries(Map.of("rolled:16020", 850L), Set.of(joined)));
+ }
+
+ private Map<String, Long> computeLogBoundaries(Map<String, Long> rolledHosts,
+ Set<ServerName> liveServers) throws IOException {
+ Admin admin = mock(Admin.class);
+ when(admin.getRegionServers()).thenReturn(liveServers);
+ return FullTableBackupClient.computeLogBoundaries(fs, walRootDir, rolledHosts, admin);
+ }
+
+ private Path walDir(ServerName serverName) {
+ return new Path(walRootDir, AbstractFSWALProvider.getWALDirectoryName(serverName.toString()));
+ }
+
+ private void createWAL(Path dir, ServerName serverName, long ts) throws IOException {
+ fs.mkdirs(dir);
+ fs.create(new Path(dir, walName(serverName, ts))).close();
+ }
+
+ private static String walName(ServerName serverName, long ts) {
+ return serverName.toString().replace(",", "%2C") + "." + ts;
+ }
+}