HBASE-26322 add rsgroup support for backup

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/BackupAdmin.java

@@ -44,7 +44,7 @@ public interface BackupAdmin extends Closeable {
    * @return the backup Id
    */
 
-  String backupTables(final BackupRequest userRequest) throws IOException;
+  BackupInfo backupTables(final BackupRequest userRequest) throws IOException;
 
   /**
    * Restore backup

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

@@ -521,7 +521,7 @@ public void restore(RestoreRequest request) throws IOException {
   }
 
   @Override
-  public String backupTables(BackupRequest request) throws IOException {
+  public BackupInfo backupTables(BackupRequest request) throws IOException {
     BackupType type = request.getBackupType();
     String targetRootDir = request.getTargetRootDir();
     List<TableName> tableList = request.getTableList();
@@ -604,7 +604,7 @@ public String backupTables(BackupRequest request) throws IOException {
 
     client.execute();
 
-    return backupId;
+    return client.backupInfo;
   }
 
   private List<TableName> excludeNonExistingTables(List<TableName> tableList,

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

@@ -40,11 +40,9 @@
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_WORKERS_DESC;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_YARN_QUEUE_NAME;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.OPTION_YARN_QUEUE_NAME_DESC;
-
 import java.io.IOException;
 import java.net.URI;
 import java.util.List;
-
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.conf.Configured;
@@ -66,7 +64,6 @@
 import org.apache.hadoop.hbase.client.ConnectionFactory;
 import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.yetus.audience.InterfaceAudience;
-
 import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
 import org.apache.hbase.thirdparty.org.apache.commons.cli.CommandLine;
 import org.apache.hbase.thirdparty.org.apache.commons.cli.HelpFormatter;
@@ -345,8 +342,9 @@ public void execute() throws IOException {
                   tables != null ? Lists.newArrayList(BackupUtils.parseTableNames(tables)) : null)
                 .withTargetRootDir(targetBackupDir).withTotalTasks(workers)
                 .withBandwidthPerTasks(bandwidth).withBackupSetName(setName).build();
-        String backupId = admin.backupTables(request);
-        System.out.println("Backup session " + backupId + " finished. Status: SUCCESS");
+        BackupInfo backupInfo = admin.backupTables(request);
+        System.out
+          .println("Backup session " + backupInfo.getBackupId() + " finished. Status: SUCCESS");
       } catch (IOException e) {
         System.out.println("Backup session finished. Status: FAILURE");
         throw e;

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

@@ -21,8 +21,11 @@
 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.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
@@ -34,7 +37,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.net.Address;
 import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
+import org.apache.hadoop.hbase.rsgroup.RSGroupInfo;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
 import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
 import org.apache.yetus.audience.InterfaceAudience;
@@ -94,13 +99,36 @@ public Map<String, Long> getIncrBackupLogFileMap() throws IOException {
     }
     newTimestamps = readRegionServerLastLogRollResult();
 
-    logList = getLogFilesForNewBackup(previousTimestampMins, newTimestamps, conf, savedStartCode);
+    logList = getLogFilesForNewBackup(previousTimestampMins, newTimestamps, conf, savedStartCode,
+      getParticipatingServerNames(backupInfo.getTables()));
     logList = excludeProcV2WALs(logList);
     backupInfo.setIncrBackupFileList(logList);
 
     return newTimestamps;
   }
 
+  private Set<String> getParticipatingServerNames(Set<TableName> tables) throws IOException {
+    Set<Address> participatingServers = new HashSet<>();
+    boolean flag = false;
+    for (TableName table : tables) {
+      RSGroupInfo rsGroupInfo = conn.getAdmin().getRSGroup(table);
+      if (rsGroupInfo != null && !rsGroupInfo.getServers().isEmpty()) {
+        LOG.info("Participating servers for table {}, rsgroup Name: {} are: {}", table,
+          rsGroupInfo.getName(), rsGroupInfo.getServers());
+        participatingServers.addAll(rsGroupInfo.getServers());
+      } else {
+        LOG.warn(
+          "Rsgroup isn't available for table {}, all servers in the cluster will be participating ",
+          table);
+        flag = true;
+      }
+    }
+
+    return flag ?
+      new HashSet<>() :
+      participatingServers.stream().map(a -> a.toString()).collect(Collectors.toSet());
+  }
+
   private List<String> excludeProcV2WALs(List<String> logList) {
     List<String> list = new ArrayList<>();
     for (int i=0; i < logList.size(); i++) {
@@ -127,8 +155,8 @@ private List<String> excludeProcV2WALs(List<String> logList) {
    * @throws IOException exception
    */
   private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
-      Map<String, Long> newestTimestamps, Configuration conf, String savedStartCode)
-      throws IOException {
+      Map<String, Long> newestTimestamps, Configuration conf, String savedStartCode,
+      Set<String> servers) throws IOException {
     LOG.debug("In getLogFilesForNewBackup()\n" + "olderTimestamps: " + olderTimestamps
         + "\n newestTimestamps: " + newestTimestamps);
 
@@ -161,7 +189,7 @@ private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
     for (FileStatus rs : rss) {
       p = rs.getPath();
       host = BackupUtils.parseHostNameFromLogFile(p);
-      if (host == null) {
+      if (host == null || (!servers.isEmpty() && !servers.contains(host))) {
         continue;
       }
       FileStatus[] logs;
@@ -216,7 +244,7 @@ private List<String> getLogFilesForNewBackup(Map<String, Long> olderTimestamps,
         continue;
       }
       host = BackupUtils.parseHostFromOldLog(p);
-      if (host == null) {
+      if (host == null || (!servers.isEmpty() && !servers.contains(host))) {
         continue;
       }
       currentLogTS = BackupUtils.getCreationTime(p);

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/master/BackupLogCleaner.java

@@ -22,8 +22,11 @@
 import java.util.ArrayList;
 import java.util.Collections;
 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.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
@@ -39,6 +42,7 @@
 import org.apache.hadoop.hbase.master.cleaner.BaseLogCleanerDelegate;
 import org.apache.hadoop.hbase.net.Address;
 import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
+import org.apache.hadoop.hbase.rsgroup.RSGroupInfo;
 import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
 import org.apache.yetus.audience.InterfaceAudience;
 import org.slf4j.Logger;
@@ -85,14 +89,31 @@ private Map<Address, Long> getServersToOldestBackupMapping(List<BackupInfo> back
     Map<Address, Long> serverAddressToLastBackupMap = new HashMap<>();
 
     Map<TableName, Long> tableNameBackupInfoMap = new HashMap<>();
+    Set<Address> servers = new HashSet<>();
+    for (BackupInfo backupInfo : backups) {
+      for (TableName table : backupInfo.getTables()) {
+        RSGroupInfo rsGroupInfo = conn.getAdmin().getRSGroup(table);
+        if (rsGroupInfo != null && rsGroupInfo.getServers() != null && !rsGroupInfo.getServers()
+          .isEmpty()) {
+          servers.addAll(rsGroupInfo.getServers());
+        } else {
+          servers.addAll(conn.getAdmin().getRegionServers().stream().map(s -> s.getAddress())
+            .collect(Collectors.toList()));
+        }
+      }
+    }
+
     for (BackupInfo backupInfo : backups) {
       for (TableName table : backupInfo.getTables()) {
         tableNameBackupInfoMap.putIfAbsent(table, backupInfo.getStartTs());
         if (tableNameBackupInfoMap.get(table) <= backupInfo.getStartTs()) {
           tableNameBackupInfoMap.put(table, backupInfo.getStartTs());
           for (Map.Entry<String, Long> entry : backupInfo.getTableSetTimestampMap().get(table)
             .entrySet()) {
-            serverAddressToLastBackupMap.put(Address.fromString(entry.getKey()), entry.getValue());
+            if (servers.contains(Address.fromString(entry.getKey()))) {
+              serverAddressToLastBackupMap
+                .put(Address.fromString(entry.getKey()), entry.getValue());
+            }
           }
         }
       }