@@ -662,8 +662,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
 
@@ -36,13 +36,21 @@
 import static org.apache.hadoop.hbase.replication.ReplicationUtils.OFFSET_UPDATE_INTERVAL_MS_KEY;
 import static org.apache.hadoop.hbase.replication.ReplicationUtils.OFFSET_UPDATE_SIZE_THRESHOLD_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.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.ServerName;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.backup.BackupCopyJob;
 import org.apache.hadoop.hbase.backup.BackupInfo;
@@ -60,7 +68,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,7 +218,12 @@ private void handleContinuousBackup(Admin admin) throws IOException {
   }
 
   private void handleNonContinuousBackup(Admin admin) throws IOException {
-    performLogRoll();
+    Map<String, Long> previousLogRollsByHost = backupManager.readRegionServerLastLogRollResult();
+    Map<String, Long> latestLogRollsByHost = performLogRoll();
+    Path walRootDir = CommonFSUtils.getWALRootDir(conf);
+    newTimestamps = computeLogBoundaries(walRootDir.getFileSystem(conf), walRootDir,
+      BackupUtils.getRolledHosts(previousLogRollsByHost, latestLogRollsByHost), admin);
+
     performBackupSnapshots(admin);
     backupManager.addIncrementalBackupTableSet(backupInfo.getTables());
 
@@ -219,15 +234,15 @@ private void handleNonContinuousBackup(Admin admin) throws IOException {
     updateBackupMetadata();
   }
 
-  private void performLogRoll() throws IOException {
+  private Map<String, Long> performLogRoll() throws IOException {
     // We roll log here before we do the snapshot. It is possible there is duplicate data
     // in the log that is already in the snapshot. But if we do it after the snapshot, we
     // could have data loss.
     // A better approach is to do the roll log on each RS in the same global procedure as
     // the snapshot.
     LOG.info("Execute roll log procedure for full backup ...");
     BackupUtils.logRoll(conn, backupInfo.getBackupRootDir(), conf);
-    newTimestamps = backupManager.readRegionServerLastLogRollResult();
+    return backupManager.readRegionServerLastLogRollResult();
   }
 
   private void performBackupSnapshots(Admin admin) throws IOException {
@@ -248,6 +263,64 @@ private void performSnapshots(Admin admin) throws IOException {
     }
   }
 
+  /**
+   * Computes the per-host log boundaries stored by this full backup, using
+   * {@link BackupUtils#computeLogBoundaries}. A host that took part in this backup's log roll gets
+   * its roll result: WALs created after the roll are included in the next incremental backup, even
+   * though some of their edits may already be in the snapshot, which is safe because deletes are
+   * replayed along with the puts. For any other host, its archived WALs and the WALs of dead
+   * servers are covered by the snapshot, so the next incremental backup does not replay them. The
+   * WALs of a live server that did not take part in the roll (for example one that started during
+   * it) are pending, because they can still receive edits that are not in the snapshot. The live
+   * servers are read only after the WAL directories are listed: a region server creates its WAL
+   * directory only after the master registers it, so every live server whose directory was listed
+   * is found. It must run right after the log roll, before the snapshot is taken, so that a host
+   * starting later gets no boundary and has all of its WALs included in the next incremental
+   * backup.
+   */
+  @RestrictedApi(
+      explanation = "Package-private for test visibility only. Do not use outside tests.",
+      link = "",
+      allowedOnPath = "(.*/src/test/.*|.*/org/apache/hadoop/hbase/backup/impl/FullTableBackupClient.java)")
+  static Map<String, Long> computeLogBoundaries(FileSystem fs, Path walRootDir,
+    Map<String, Long> rolledHosts, Admin admin) throws IOException {
+    Path logDir = new Path(walRootDir, HConstants.HREGION_LOGDIR_NAME);
+    Path oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
+
+    Map<ServerName, List<String>> logsByServer = new HashMap<>();
+    for (FileStatus serverLogDir : fs.listStatus(logDir)) {
+      ServerName serverName =
+        AbstractFSWALProvider.getServerNameFromWALDirectoryName(serverLogDir.getPath());
+      if (serverName == null) {
+        continue;
+      }
+      List<String> logs = logsByServer.computeIfAbsent(serverName, k -> new ArrayList<>());
+      for (FileStatus log : fs.listStatus(serverLogDir.getPath())) {
+        if (!AbstractFSWALProvider.isMetaFile(log.getPath())) {
+          logs.add(log.getPath().toString());
+        }
+      }
+    }
+
+    Set<ServerName> live = new HashSet<>(admin.getRegionServers());
+    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);
+  }
+
   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.
 
@@ -54,12 +54,13 @@ 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, and the logs
+   *         held back for a later backup (for example, logs still being split) are pending.
    * @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 +70,27 @@ 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, logList, selection.heldBack());
+    LOG.debug("Log boundaries for incremental backup {}: {}", backupInfo.getBackupId(),
+      newTimestamps);
     return newTimestamps;
   }
 
+  private record LogFileSelection(List<String> included, List<String> heldBack) {
+  }
+
   private List<String> excludeProcV2WALs(List<String> logList) {
     List<String> list = new ArrayList<>();
     for (int i = 0; i < logList.size(); i++) {
@@ -97,15 +107,16 @@ 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, and the log files held back for a later backup
    * @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);
@@ -217,18 +228,14 @@ private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
         resultLogFiles.add(currentLogFile);
       }
 
-      // It is possible that a host in .oldlogs is an obsolete region server
-      // so newestTimestamps.get(host) here can be null.
-      // Even if these logs belong to a obsolete region server, we still need
-      // to include they to avoid loss of edits for backup.
       Long newTimestamp = newestTimestamps.get(host);
-      if (newTimestamp == null || currentLogTS > newTimestamp) {
+      if (newTimestamp != null && currentLogTS > newTimestamp) {
         newestLogs.add(currentLogFile);
       }
     }
     // remove newest log per host because they are still in use
     resultLogFiles.removeAll(newestLogs);
-    return resultLogFiles;
+    return new LogFileSelection(resultLogFiles, newestLogs);
   }
 
   static class NewestLogFilter implements PathFilter {
 
@@ -31,6 +31,7 @@
 import java.text.SimpleDateFormat;
 import java.time.ZoneOffset;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.Date;
@@ -206,6 +207,62 @@ 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. 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. A host without covered
+   * or pending logs gets no boundary.</li>
+   * </ul>
+   * Capping is needed because a host's logs can still be pending while some of its newer logs are
+   * already covered, for example when a dead server's logs are split and archived out of order. A
+   * host is also kept while any of its logs are pending, even if none are covered, because without
+   * a boundary the next backup would treat it as unknown and skip its older logs.
+   * @param rolledHosts roll results of the hosts that took part in the log roll
+   * @param coveredLogs logs whose edits are covered by this backup
+   * @param pendingLogs logs whose edits might not be covered by this backup
+   * @return boundaries by host
+   */
+  public static Map<String, Long> computeLogBoundaries(Map<String, Long> rolledHosts,
+    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);
+    return boundaries;
+  }
+
   /**
    * Returns WAL file name
    * @param walFileName WAL file name