HBASE-29519 Copy Bulkloaded Files in Continuous Backup

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hbase.backup.impl;
 
+import static org.apache.hadoop.hbase.HConstants.REPLICATION_BULKLOAD_ENABLE_KEY;
 import static org.apache.hadoop.hbase.HConstants.REPLICATION_SCOPE_GLOBAL;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.BACKUP_ATTEMPTS_PAUSE_MS_KEY;
 import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.BACKUP_MAX_ATTEMPTS_KEY;
@@ -190,6 +191,15 @@ private void handleContinuousBackup(Admin admin) throws IOException {
     // set overall backup status: complete. Here we make sure to complete the backup.
     // After this checkpoint, even if entering cancel process, will let the backup finished
     backupInfo.setState(BackupState.COMPLETE);
+
+    if (!conf.getBoolean(REPLICATION_BULKLOAD_ENABLE_KEY, false)) {
+      System.out.println("WARNING: Bulkload replication is not enabled. "
+        + "Since continuous backup is using HBase replication, bulk loaded files won't be backed up as part of continuous backup. "
+        + "To ensure bulk-loaded files are backed up, enable bulkload replication "
+        + "(hbase.replication.bulkload.enabled=true) and configure a unique cluster ID using "
+        + "hbase.replication.cluster.id. This cluster ID is required by the replication framework "
+        + "to uniquely identify clusters, even if continuous backup itself does not directly rely on it.");
+    }
   }
 
   private void handleNonContinuousBackup(Admin admin) throws IOException {

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/replication/BackupFileSystemManager.java

@@ -26,25 +26,28 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * Initializes and organizes backup directories for continuous Write-Ahead Logs (WALs) files within
- * the specified backup root directory.
+ * Initializes and organizes backup directories for continuous Write-Ahead Logs (WALs) and
+ * bulk-loaded files within the specified backup root directory.
  */
 @InterfaceAudience.Private
 public class BackupFileSystemManager {
   private static final Logger LOG = LoggerFactory.getLogger(BackupFileSystemManager.class);
 
   public static final String WALS_DIR = "WALs";
+  public static final String BULKLOAD_FILES_DIR = "bulk-load-files";
   private final String peerId;
   private final FileSystem backupFs;
   private final Path backupRootDir;
   private final Path walsDir;
+  private final Path bulkLoadFilesDir;
 
   public BackupFileSystemManager(String peerId, Configuration conf, String backupRootDirStr)
     throws IOException {
     this.peerId = peerId;
     this.backupRootDir = new Path(backupRootDirStr);
     this.backupFs = FileSystem.get(backupRootDir.toUri(), conf);
     this.walsDir = createDirectory(WALS_DIR);
+    this.bulkLoadFilesDir = createDirectory(BULKLOAD_FILES_DIR);
   }
 
   private Path createDirectory(String dirName) throws IOException {
@@ -58,6 +61,10 @@ public Path getWalsDir() {
     return walsDir;
   }
 
+  public Path getBulkLoadFilesDir() {
+    return bulkLoadFilesDir;
+  }
+
   public FileSystem getBackupFs() {
     return backupFs;
   }

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/replication/BulkLoadProcessor.java

@@ -0,0 +1,96 @@
+/*
+ * 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.replication;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.wal.WAL;
+import org.apache.hadoop.hbase.wal.WALEdit;
+import org.apache.yetus.audience.InterfaceAudience;
+
+import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
+
+/**
+ * Processes bulk load files from Write-Ahead Log (WAL) entries for HBase replication.
+ * <p>
+ * This utility class extracts and constructs the file paths of bulk-loaded files based on WAL
+ * entries. It processes bulk load descriptors and their associated store descriptors to generate
+ * the paths for each bulk-loaded file.
+ * <p>
+ * The class is designed for scenarios where replicable bulk load operations need to be parsed and
+ * their file paths need to be determined programmatically.
+ * </p>
+ */
+@InterfaceAudience.Private
+public final class BulkLoadProcessor {
+  private BulkLoadProcessor() {
+  }
+
+  public static List<Path> processBulkLoadFiles(List<WAL.Entry> walEntries) throws IOException {
+    List<Path> bulkLoadFilePaths = new ArrayList<>();
+
+    for (WAL.Entry entry : walEntries) {
+      WALEdit edit = entry.getEdit();
+      for (Cell cell : edit.getCells()) {
+        if (CellUtil.matchingQualifier(cell, WALEdit.BULK_LOAD)) {
+          TableName tableName = entry.getKey().getTableName();
+          String namespace = tableName.getNamespaceAsString();
+          String table = tableName.getQualifierAsString();
+          bulkLoadFilePaths.addAll(processBulkLoadDescriptor(cell, namespace, table));
+        }
+      }
+    }
+    return bulkLoadFilePaths;
+  }
+
+  private static List<Path> processBulkLoadDescriptor(Cell cell, String namespace, String table)
+    throws IOException {
+    List<Path> bulkLoadFilePaths = new ArrayList<>();
+    WALProtos.BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell);
+
+    if (bld == null || !bld.getReplicate() || bld.getEncodedRegionName() == null) {
+      return bulkLoadFilePaths; // Skip if not replicable
+    }
+
+    String regionName = bld.getEncodedRegionName().toStringUtf8();
+    for (WALProtos.StoreDescriptor storeDescriptor : bld.getStoresList()) {
+      bulkLoadFilePaths
+        .addAll(processStoreDescriptor(storeDescriptor, namespace, table, regionName));
+    }
+
+    return bulkLoadFilePaths;
+  }
+
+  private static List<Path> processStoreDescriptor(WALProtos.StoreDescriptor storeDescriptor,
+    String namespace, String table, String regionName) {
+    List<Path> paths = new ArrayList<>();
+    String columnFamily = storeDescriptor.getFamilyName().toStringUtf8();
+
+    for (String storeFile : storeDescriptor.getStoreFileList()) {
+      paths.add(new Path(namespace,
+        new Path(table, new Path(regionName, new Path(columnFamily, storeFile)))));
+    }
+
+    return paths;
+  }
+}

hbase-backup/src/main/java/org/apache/hadoop/hbase/backup/replication/BulkLoadUploadException.java

@@ -0,0 +1,32 @@
+/*
+ * 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.replication;
+
+import java.io.IOException;
+import org.apache.yetus.audience.InterfaceAudience;
+
+@InterfaceAudience.Private
+public class BulkLoadUploadException extends IOException {
+  public BulkLoadUploadException(String message) {
+    super(message);
+  }
+
+  public BulkLoadUploadException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}