HBASE-29984 Support separate old WAL directories in backup

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

@@ -184,10 +184,10 @@ private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
     }
 
     // Include the .oldlogs files too.
-    FileStatus[] oldlogs = fs.listStatus(oldLogDir);
-    for (FileStatus oldlog : oldlogs) {
-      p = oldlog.getPath();
-      currentLogFile = p.toString();
+    List<String> oldlogs = BackupUtils.getFiles(fs, oldLogDir, new ArrayList<>(), path -> true);
+    for (String oldlog : oldlogs) {
+      p = new Path(oldlog);
+      currentLogFile = oldlog;
       if (AbstractFSWALProvider.isMetaFile(p)) {
         if (LOG.isDebugEnabled()) {
           LOG.debug("Skip .meta log file: " + currentLogFile);

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/util/BackupUtils.java

@@ -332,6 +332,10 @@ public static String parseHostFromOldLog(Path p) {
     if (p.getName().endsWith(MasterRegionFactory.ARCHIVED_WAL_SUFFIX)) {
       return null;
     }
+    Path parent = p.getParent();
+    if (parent != null && ServerName.isFullServerName(parent.getName())) {
+      return ServerName.valueOf(parent.getName()).getAddress().toString();
+    }
     try {
       String urlDecodedName = URLDecoder.decode(p.getName(), "UTF8");
       Iterable<String> nameSplitsOnComma = Splitter.on(",").split(urlDecodedName);

hbase-backup/src/test/java/org/apache/hadoop/hbase/backup/TestIncrementalBackupManager.java

@@ -0,0 +1,90 @@
+/*
+ * 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.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+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.testclassification.LargeTests;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
+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();
+    conf1.setBoolean(AbstractFSWALProvider.SEPARATE_OLDLOGDIR, true);
+    autoRestoreOnFailure = true;
+    useSecondCluster = false;
+    setUpHelper();
+  }
+
+  @Test
+  public void testCollectWALFilesFromRegionServerDirectories() throws Exception {
+    List<TableName> tables = List.of(table1);
+    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));
+
+      try (IncrementalBackupManager manager = new IncrementalBackupManager(conn, conf1)) {
+        BackupInfo backupInfo = manager.createBackupInfo("backup_test", BackupType.INCREMENTAL,
+          tables, BACKUP_ROOT_DIR, -1, -1, false);
+        Map<String, Long> previousTimestamps =
+          BackupUtils.getRSLogTimestampMins(manager.readLogTimestampMap());
+        ServerName serverName = TEST_UTIL.getMiniHBaseCluster().getRegionServer(0).getServerName();
+        Long previousTimestamp = previousTimestamps.get(serverName.getAddress().toString());
+        assertNotNull(previousTimestamp);
+
+        TEST_UTIL.waitFor(30_000,
+          () -> EnvironmentEdgeManager.currentTime() > previousTimestamp + 1);
+        Path walRootDir = CommonFSUtils.getWALRootDir(conf1);
+        Path archiveDir = new Path(walRootDir,
+          AbstractFSWALProvider.getWALArchiveDirectoryName(conf1, serverName.toString()));
+        Path archivedWAL = new Path(archiveDir, "wal." + (previousTimestamp + 1));
+        FileSystem fs = walRootDir.getFileSystem(conf1);
+        fs.mkdirs(archiveDir);
+        fs.create(archivedWAL).close();
+
+        manager.getIncrBackupLogFileMap();
+
+        assertTrue(backupInfo.getIncrBackupFileList().contains(archivedWAL.toString()));
+      }
+    }
+  }
+}