Log filtering in IncrementalBackupManager can lead to data loss

Author: Hernan Gelaf-Romer

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java

@@ -38,11 +38,16 @@
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 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.TableName;
 import org.apache.hadoop.hbase.backup.BackupCopyJob;
 import org.apache.hadoop.hbase.backup.BackupInfo;
@@ -60,7 +65,9 @@
 import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
 import org.apache.hadoop.hbase.replication.ReplicationException;
 import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+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;
@@ -208,14 +215,17 @@ private void handleContinuousBackup(Admin admin) throws IOException {
   }
 
   private void handleNonContinuousBackup(Admin admin) throws IOException {
+    Map<String, Long> previousLogRollsByHost = backupManager.readRegionServerLastLogRollResult();
     performLogRoll();
+    Map<String, Long> latestLogRollsByHost = newTimestamps;
     performBackupSnapshots(admin);
     backupManager.addIncrementalBackupTableSet(backupInfo.getTables());
 
     // set overall backup status: complete. Here we make sure to complete the backup.
     // After this checkpoint, even if entering cancel process, will let the backup finished
     backupInfo.setState(BackupState.COMPLETE);
 
+    adjustTimestampsForOfflineHosts(previousLogRollsByHost, latestLogRollsByHost);
     updateBackupMetadata();
   }
 
@@ -248,6 +258,54 @@ private void performSnapshots(Admin admin) throws IOException {
     }
   }
 
+  /**
+   * Scan all WAL files to find offline/decommissioned hosts and record their max WAL timestamps in
+   * newTimestamps. This ensures subsequent incremental backups won't re-include WALs already covered
+   * by this full backup's snapshot.
+   */
+  private void adjustTimestampsForOfflineHosts(Map<String, Long> previousLogRollsByHost,
+    Map<String, Long> latestLogRollsByHost) throws IOException {
+    Path walRootDir = CommonFSUtils.getWALRootDir(conf);
+    Path logDir = new Path(walRootDir, HConstants.HREGION_LOGDIR_NAME);
+    Path oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
+    FileSystem fs = walRootDir.getFileSystem(conf);
+
+    List<FileStatus> allLogs = new ArrayList<>();
+    for (FileStatus hostLogDir : fs.listStatus(logDir)) {
+      String host = BackupUtils.parseHostNameFromLogFile(hostLogDir.getPath());
+      if (host == null) {
+        continue;
+      }
+      allLogs.addAll(Arrays.asList(fs.listStatus(hostLogDir.getPath())));
+    }
+    allLogs.addAll(Arrays.asList(fs.listStatus(oldLogDir)));
+
+    newTimestamps = new HashMap<>();
+
+    for (FileStatus log : allLogs) {
+      if (AbstractFSWALProvider.isMetaFile(log.getPath())) {
+        continue;
+      }
+      String host = BackupUtils.parseHostNameFromLogFile(log.getPath());
+      if (host == null) {
+        continue;
+      }
+      long timestamp = BackupUtils.getCreationTime(log.getPath());
+      Long previousLogRoll = previousLogRollsByHost.get(host);
+      Long latestLogRoll = latestLogRollsByHost.get(host);
+      boolean isInactive = latestLogRoll == null || latestLogRoll.equals(previousLogRoll);
+
+      if (isInactive) {
+        long currentTs = newTimestamps.getOrDefault(host, 0L);
+        if (timestamp > currentTs) {
+          newTimestamps.put(host, timestamp);
+        }
+      } else {
+        newTimestamps.put(host, latestLogRoll);
+      }
+    }
+  }
+
   private void updateBackupMetadata() throws IOException {
     // The table list in backupInfo is good for both full backup and incremental backup.
     // For incremental backup, it contains the incremental backup table set.

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/impl/IncrementalBackupManager.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.hadoop.conf.Configuration;
@@ -59,7 +60,6 @@ public IncrementalBackupManager(Connection conn, Configuration conf) throws IOEx
    */
   public Map<String, Long> getIncrBackupLogFileMap() throws IOException {
     List<String> logList;
-    Map<String, Long> newTimestamps;
     Map<String, Long> previousTimestampMins =
       BackupUtils.getRSLogTimestampMins(readLogTimestampMap());
 
@@ -72,12 +72,46 @@ public Map<String, Long> getIncrBackupLogFileMap() throws IOException {
     LOG.info("Execute roll log procedure for incremental backup ...");
     BackupUtils.logRoll(conn, backupInfo.getBackupRootDir(), conf);
 
-    newTimestamps = readRegionServerLastLogRollResult();
+    Map<String, Long> newTimestamps = new HashMap<>();
+    Map<String, Long> latestLogRollByHost = readRegionServerLastLogRollResult();
+    for (Map.Entry<String, Long> entry : latestLogRollByHost.entrySet()) {
+      String host = entry.getKey();
+      long latestLogRoll = entry.getValue();
+      Long earliestTimestampToIncludeInBackup = previousTimestampMins.get(host);
+
+      boolean isInactive = earliestTimestampToIncludeInBackup != null
+        && earliestTimestampToIncludeInBackup >= latestLogRoll;
+
+      if (isInactive) {
+        LOG.debug(
+          "Skipping inactive host {} from newTimestamps (boundary={} >= latestLogRoll={})", host,
+          earliestTimestampToIncludeInBackup, latestLogRoll);
+      } else {
+        newTimestamps.put(host, latestLogRoll);
+      }
+    }
 
     logList = getLogFilesForNewBackup(previousTimestampMins, newTimestamps, conf);
     logList = excludeProcV2WALs(logList);
     backupInfo.setIncrBackupFileList(logList);
 
+    // Update boundaries based on WALs that will be backed up
+    for (String logFile : logList) {
+      Path logPath = new Path(logFile);
+      String logHost = BackupUtils.parseHostFromOldLog(logPath);
+      if (logHost == null) {
+        logHost = BackupUtils.parseHostNameFromLogFile(logPath.getParent());
+      }
+      if (logHost != null) {
+        long logTs = BackupUtils.getCreationTime(logPath);
+        Long latestTimestampToIncludeInBackup = newTimestamps.get(logHost);
+        if (latestTimestampToIncludeInBackup == null || logTs > latestTimestampToIncludeInBackup) {
+          LOG.info("Updating backup boundary for inactive host {}: timestamp={}", logHost, logTs);
+          newTimestamps.put(logHost, logTs);
+        }
+      }
+    }
+
     return newTimestamps;
   }
 
@@ -217,14 +251,6 @@ 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) {
-        newestLogs.add(currentLogFile);
-      }
     }
     // remove newest log per host because they are still in use
     resultLogFiles.removeAll(newestLogs);