@@ -639,8 +639,9 @@ public List<BackupInfo> getBackupHistory(Order order, int n, BackupInfo.Filter..
 
   /**
    * 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
 
@@ -23,10 +23,20 @@
 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.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 void execute() throws IOException {
       // 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 @@ protected void snapshotTable(Admin admin, TableName tableName, String snapshotNa
     }
     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);
+  }
 }
@@ -20,8 +20,10 @@
 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;
@@ -54,12 +56,14 @@ public IncrementalBackupManager(Connection conn, Configuration conf) throws IOEx
   /**
    * 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());
 
@@ -69,18 +73,28 @@ public Map<String, Long> getIncrBackupLogFileMap() throws IOException {
         + "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++) {
@@ -97,15 +111,17 @@ private List<String> excludeProcV2WALs(List<String> logList) {
   }
 
   /**
-   * 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);
@@ -119,6 +135,7 @@ private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
 
     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
@@ -143,6 +160,7 @@ private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
       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
@@ -165,17 +183,15 @@ private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
         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);
@@ -198,6 +214,7 @@ private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
       if (host == null) {
         continue;
       }
+      hostsWithLogs.add(host);
       currentLogTS = BackupUtils.getCreationTime(p);
       oldTimeStamp = olderTimestamps.get(host);
       /*
@@ -217,18 +234,15 @@ 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
+    // 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 {
 
@@ -21,12 +21,14 @@
 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 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. 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