@@ -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,82 @@ private void performSnapshots(Admin admin) throws IOException {
     }
   }
 
+  /**
+   * 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);
+  }
+
   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.
 
@@ -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 {