/*
 * 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.util;

import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONF_CONTINUOUS_BACKUP_WAL_DIR;
import static org.apache.hadoop.hbase.backup.BackupRestoreConstants.CONTINUOUS_BACKUP_REPLICATION_PEER;
import static org.apache.hadoop.hbase.backup.replication.ContinuousBackupReplicationEndpoint.ONE_DAY_IN_MILLISECONDS;
import static org.apache.hadoop.hbase.backup.util.BackupFileSystemManager.WALS_DIR;
import static org.apache.hadoop.hbase.replication.regionserver.ReplicationMarkerChore.REPLICATION_MARKER_ENABLED_DEFAULT;
import static org.apache.hadoop.hbase.replication.regionserver.ReplicationMarkerChore.REPLICATION_MARKER_ENABLED_KEY;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URLDecoder;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.ZoneOffset;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.TimeZone;
import java.util.TreeSet;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.PathFilter;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.MetaTableAccessor;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.backup.BackupInfo;
import org.apache.hadoop.hbase.backup.BackupRestoreConstants;
import org.apache.hadoop.hbase.backup.HBackupFileSystem;
import org.apache.hadoop.hbase.backup.RestoreRequest;
import org.apache.hadoop.hbase.backup.impl.BackupManifest;
import org.apache.hadoop.hbase.backup.impl.BackupManifest.BackupImage;
import org.apache.hadoop.hbase.backup.impl.BackupSystemTable;
import org.apache.hadoop.hbase.backup.master.LogRollMasterProcedureManager;
import org.apache.hadoop.hbase.client.Admin;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.master.region.MasterRegionFactory;
import org.apache.hadoop.hbase.replication.ReplicationException;
import org.apache.hadoop.hbase.replication.ReplicationGroupOffset;
import org.apache.hadoop.hbase.replication.ReplicationQueueId;
import org.apache.hadoop.hbase.replication.ReplicationQueueStorage;
import org.apache.hadoop.hbase.replication.ReplicationStorageFactory;
import org.apache.hadoop.hbase.tool.BulkLoadHFiles;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.FSTableDescriptors;
import org.apache.hadoop.hbase.util.FSUtils;
import org.apache.hadoop.hbase.wal.AbstractFSWALProvider;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.base.Splitter;
import org.apache.hbase.thirdparty.com.google.common.base.Strings;
import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableMap;
import org.apache.hbase.thirdparty.com.google.common.collect.Iterables;
import org.apache.hbase.thirdparty.com.google.common.collect.Iterators;

/**
 * A collection for methods used by multiple classes to backup HBase tables.
 */
@InterfaceAudience.Private
public final class BackupUtils {
  private static final Logger LOG = LoggerFactory.getLogger(BackupUtils.class);
  public static final String LOGNAME_SEPARATOR = ".";
  public static final int MILLISEC_IN_HOUR = 3600000;
  public static final String DATE_FORMAT = "yyyy-MM-dd";

  private BackupUtils() {
    throw new AssertionError("Instantiating utility class...");
  }

  /**
   * Loop through the RS log timestamp map for the tables, for each RS, find the min timestamp value
   * for the RS among the tables.
   * @param rsLogTimestampMap timestamp map
   * @return the min timestamp of each RS
   */
  public static Map<String, Long>
    getRSLogTimestampMins(Map<TableName, Map<String, Long>> rsLogTimestampMap) {
    return rsLogTimestampMap.values().stream().flatMap(map -> map.entrySet().stream())
      .collect(Collectors.toMap(Entry::getKey, Entry::getValue, Math::min));
  }

  /**
   * copy out Table RegionInfo into incremental backup image need to consider move this logic into
   * HBackupFileSystem
   * @param conn       connection
   * @param backupInfo backup info
   * @param conf       configuration
   * @throws IOException exception
   */
  public static void copyTableRegionInfo(Connection conn, BackupInfo backupInfo, Configuration conf)
    throws IOException {
    Path rootDir = CommonFSUtils.getRootDir(conf);
    FileSystem fs = rootDir.getFileSystem(conf);

    // for each table in the table set, copy out the table info and region
    // info files in the correct directory structure
    try (Admin admin = conn.getAdmin()) {
      for (TableName table : backupInfo.getTables()) {
        if (!admin.tableExists(table)) {
          LOG.warn("Table " + table + " does not exists, skipping it.");
          continue;
        }
        TableDescriptor orig = FSTableDescriptors.getTableDescriptorFromFs(fs, rootDir, table);

        // write a copy of descriptor to the target directory
        Path target = new Path(backupInfo.getTableBackupDir(table));
        FileSystem targetFs = target.getFileSystem(conf);
        try (FSTableDescriptors descriptors =
          new FSTableDescriptors(targetFs, CommonFSUtils.getRootDir(conf))) {
          descriptors.createTableDescriptorForTableDirectory(target, orig, false);
        }
        LOG.debug("Attempting to copy table info for:" + table + " target: " + target
          + " descriptor: " + orig);
        LOG.debug("Finished copying tableinfo.");
        List<RegionInfo> regions = MetaTableAccessor.getTableRegions(conn, table);
        // For each region, write the region info to disk
        LOG.debug("Starting to write region info for table " + table);
        for (RegionInfo regionInfo : regions) {
          Path regionDir = FSUtils
            .getRegionDirFromTableDir(new Path(backupInfo.getTableBackupDir(table)), regionInfo);
          regionDir = new Path(backupInfo.getTableBackupDir(table), regionDir.getName());
          writeRegioninfoOnFilesystem(conf, targetFs, regionDir, regionInfo);
        }
        LOG.debug("Finished writing region info for table " + table);
      }
    }
  }

  /**
   * Write the .regioninfo file on-disk.
   */
  public static void writeRegioninfoOnFilesystem(final Configuration conf, final FileSystem fs,
    final Path regionInfoDir, RegionInfo regionInfo) throws IOException {
    final byte[] content = RegionInfo.toDelimitedByteArray(regionInfo);
    Path regionInfoFile = new Path(regionInfoDir, "." + HConstants.REGIONINFO_QUALIFIER_STR);
    // First check to get the permissions
    FsPermission perms = CommonFSUtils.getFilePermissions(fs, conf, HConstants.DATA_FILE_UMASK_KEY);
    // Write the RegionInfo file content
    FSDataOutputStream out = FSUtils.create(conf, fs, regionInfoFile, perms, null);
    try {
      out.write(content);
    } finally {
      out.close();
    }
  }

  /**
   * Parses hostname:port from WAL file path
   * @param p path to WAL file
   * @return hostname:port
   */
  public static String parseHostNameFromLogFile(Path p) {
    try {
      if (AbstractFSWALProvider.isArchivedLogFile(p)) {
        return BackupUtils.parseHostFromOldLog(p);
      } else {
        ServerName sname = AbstractFSWALProvider.getServerNameFromWALDirectoryName(p);
        if (sname != null) {
          return sname.getAddress().toString();
        } else {
          LOG.error("Skip log file (can't parse): " + p);
          return null;
        }
      }
    } catch (Exception e) {
      LOG.error("Skip log file (can't parse): " + p, e);
      return null;
    }
  }

  /**
   * Returns WAL file name
   * @param walFileName WAL file name
   * @return WAL file name
   */
  public static String getUniqueWALFileNamePart(String walFileName) {
    return getUniqueWALFileNamePart(new Path(walFileName));
  }

  /**
   * Returns WAL file name
   * @param p WAL file path
   * @return WAL file name
   */
  public static String getUniqueWALFileNamePart(Path p) {
    return p.getName();
  }

  /**
   * Get the total length of files under the given directory recursively.
   * @param fs  The hadoop file system
   * @param dir The target directory
   * @return the total length of files
   * @throws IOException exception
   */
  public static long getFilesLength(FileSystem fs, Path dir) throws IOException {
    long totalLength = 0;
    FileStatus[] files = CommonFSUtils.listStatus(fs, dir);
    if (files != null) {
      for (FileStatus fileStatus : files) {
        if (fileStatus.isDirectory()) {
          totalLength += getFilesLength(fs, fileStatus.getPath());
        } else {
          totalLength += fileStatus.getLen();
        }
      }
    }
    return totalLength;
  }

  /**
   * Get list of all old WAL files (WALs and archive)
   * @param c                configuration
   * @param hostTimestampMap {host,timestamp} map
   * @return list of WAL files
   * @throws IOException exception
   */
  public static List<String> getWALFilesOlderThan(final Configuration c,
    final HashMap<String, Long> hostTimestampMap) throws IOException {
    Path walRootDir = CommonFSUtils.getWALRootDir(c);
    Path logDir = new Path(walRootDir, HConstants.HREGION_LOGDIR_NAME);
    Path oldLogDir = new Path(walRootDir, HConstants.HREGION_OLDLOGDIR_NAME);
    List<String> logFiles = new ArrayList<>();

    PathFilter filter = p -> {
      try {
        if (AbstractFSWALProvider.isMetaFile(p)) {
          return false;
        }
        String host = parseHostNameFromLogFile(p);
        if (host == null) {
          return false;
        }
        Long oldTimestamp = hostTimestampMap.get(host);
        Long currentLogTS = BackupUtils.getCreationTime(p);
        return currentLogTS <= oldTimestamp;
      } catch (Exception e) {
        LOG.warn("Can not parse" + p, e);
        return false;
      }
    };
    FileSystem walFs = CommonFSUtils.getWALFileSystem(c);
    logFiles = BackupUtils.getFiles(walFs, logDir, logFiles, filter);
    logFiles = BackupUtils.getFiles(walFs, oldLogDir, logFiles, filter);
    return logFiles;
  }

  public static TableName[] parseTableNames(String tables) {
    if (tables == null) {
      return null;
    }
    return Splitter.on(BackupRestoreConstants.TABLENAME_DELIMITER_IN_COMMAND).splitToStream(tables)
      .map(TableName::valueOf).toArray(TableName[]::new);
  }

  /**
   * Check whether the backup path exist
   * @param backupStr backup
   * @param conf      configuration
   * @return Yes if path exists
   * @throws IOException exception
   */
  public static boolean checkPathExist(String backupStr, Configuration conf) throws IOException {
    boolean isExist = false;
    Path backupPath = new Path(backupStr);
    FileSystem fileSys = backupPath.getFileSystem(conf);
    String targetFsScheme = fileSys.getUri().getScheme();
    if (LOG.isTraceEnabled()) {
      LOG.trace("Schema of given url: " + backupStr + " is: " + targetFsScheme);
    }
    if (fileSys.exists(backupPath)) {
      isExist = true;
    }
    return isExist;
  }

  /**
   * Check target path first, confirm it doesn't exist before backup
   * @param backupRootPath backup destination path
   * @param conf           configuration
   * @throws IOException exception
   */
  public static void checkTargetDir(String backupRootPath, Configuration conf) throws IOException {
    boolean targetExists;
    try {
      targetExists = checkPathExist(backupRootPath, conf);
    } catch (IOException e) {
      String expMsg = e.getMessage();
      String newMsg = null;
      if (expMsg.contains("No FileSystem for scheme")) {
        newMsg =
          "Unsupported filesystem scheme found in the backup target url. Error Message: " + expMsg;
        LOG.error(newMsg);
        throw new IOException(newMsg);
      } else {
        throw e;
      }
    }

    if (targetExists) {
      LOG.info("Using existing backup root dir: " + backupRootPath);
    } else {
      LOG.info("Backup root dir " + backupRootPath + " does not exist. Will be created.");
    }
  }

  /**
   * Get the min value for all the Values a map.
   * @param map map
   * @return the min value
   */
  public static <T> Long getMinValue(Map<T, Long> map) {
    Long minTimestamp = null;
    if (map != null) {
      ArrayList<Long> timestampList = new ArrayList<>(map.values());
      Collections.sort(timestampList);
      // The min among all the RS log timestamps will be kept in backup system table table.
      minTimestamp = timestampList.get(0);
    }
    return minTimestamp;
  }

  /**
   * Parses host name:port from archived WAL path
   * @param p path
   * @return host name
   */
  public static String parseHostFromOldLog(Path p) {
    // Skip master wals
    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);
      String host = Iterables.get(nameSplitsOnComma, 0);
      String port = Iterables.get(nameSplitsOnComma, 1);
      return host + ":" + port;
    } catch (Exception e) {
      LOG.warn("Skip log file (can't parse): {}", p);
      return null;
    }
  }

  /**
   * Given the log file, parse the timestamp from the file name. The timestamp is the last number.
   * @param p a path to the log file
   * @return the timestamp
   * @throws IOException exception
   */
  public static Long getCreationTime(Path p) throws IOException {
    int idx = p.getName().lastIndexOf(LOGNAME_SEPARATOR);
    if (idx < 0) {
      throw new IOException("Cannot parse timestamp from path " + p);
    }
    String ts = p.getName().substring(idx + 1);
    return Long.parseLong(ts);
  }

  public static List<String> getFiles(FileSystem fs, Path rootDir, List<String> files,
    PathFilter filter) throws IOException {
    RemoteIterator<LocatedFileStatus> it = fs.listFiles(rootDir, true);

    while (it.hasNext()) {
      LocatedFileStatus lfs = it.next();
      if (lfs.isDirectory()) {
        continue;
      }
      // apply filter
      if (filter.accept(lfs.getPath())) {
        files.add(lfs.getPath().toString());
      }
    }
    return files;
  }

  public static void cleanupBackupData(BackupInfo context, Configuration conf) throws IOException {
    cleanupHLogDir(context, conf);
    cleanupTargetDir(context, conf);
  }

  /**
   * Clean up directories which are generated when DistCp copying hlogs
   * @param backupInfo backup info
   * @param conf       configuration
   * @throws IOException exception
   */
  private static void cleanupHLogDir(BackupInfo backupInfo, Configuration conf) throws IOException {
    String logDir = backupInfo.getHLogTargetDir();
    if (logDir == null) {
      LOG.warn("No log directory specified for " + backupInfo.getBackupId());
      return;
    }

    Path rootPath = new Path(logDir).getParent();
    FileSystem fs = FileSystem.get(rootPath.toUri(), conf);
    FileStatus[] files = listStatus(fs, rootPath, null);
    if (files == null) {
      return;
    }
    for (FileStatus file : files) {
      LOG.debug("Delete log files: " + file.getPath().getName());
      fs.delete(file.getPath(), true);
    }
  }

  private static void cleanupTargetDir(BackupInfo backupInfo, Configuration conf) {
    try {
      // clean up the data at target directory
      LOG.debug("Trying to cleanup up target dir : " + backupInfo.getBackupId());
      String targetDir = backupInfo.getBackupRootDir();
      if (targetDir == null) {
        LOG.warn("No target directory specified for " + backupInfo.getBackupId());
        return;
      }

      FileSystem outputFs = FileSystem.get(new Path(backupInfo.getBackupRootDir()).toUri(), conf);

      for (TableName table : backupInfo.getTables()) {
        Path targetDirPath = new Path(
          getTableBackupDir(backupInfo.getBackupRootDir(), backupInfo.getBackupId(), table));
        if (outputFs.delete(targetDirPath, true)) {
          LOG.info("Cleaning up backup data at " + targetDirPath.toString() + " done.");
        } else {
          LOG.info("No data has been found in " + targetDirPath.toString() + ".");
        }

        Path tableDir = targetDirPath.getParent();
        FileStatus[] backups = listStatus(outputFs, tableDir, null);
        if (backups == null || backups.length == 0) {
          outputFs.delete(tableDir, true);
          LOG.debug(tableDir.toString() + " is empty, remove it.");
        }
      }
      outputFs.delete(new Path(targetDir, backupInfo.getBackupId()), true);
    } catch (IOException e1) {
      LOG.error("Cleaning up backup data of " + backupInfo.getBackupId() + " at "
        + backupInfo.getBackupRootDir() + " failed due to " + e1.getMessage() + ".");
    }
  }

  /**
   * Given the backup root dir, backup id and the table name, return the backup image location,
   * which is also where the backup manifest file is. return value look like:
   * "hdfs://backup.hbase.org:9000/user/biadmin/backup1/backup_1396650096738/default/t1_dn/"
   * @param backupRootDir backup root directory
   * @param backupId      backup id
   * @param tableName     table name
   * @return backupPath String for the particular table
   */
  public static String getTableBackupDir(String backupRootDir, String backupId,
    TableName tableName) {
    return backupRootDir + Path.SEPARATOR + backupId + Path.SEPARATOR
      + tableName.getNamespaceAsString() + Path.SEPARATOR + tableName.getQualifierAsString()
      + Path.SEPARATOR;
  }

  /**
   * Calls fs.listStatus() and treats FileNotFoundException as non-fatal This accommodates
   * differences between hadoop versions, where hadoop 1 does not throw a FileNotFoundException, and
   * return an empty FileStatus[] while Hadoop 2 will throw FileNotFoundException.
   * @param fs     file system
   * @param dir    directory
   * @param filter path filter
   * @return null if dir is empty or doesn't exist, otherwise FileStatus array
   */
  public static FileStatus[] listStatus(final FileSystem fs, final Path dir,
    final PathFilter filter) throws IOException {
    FileStatus[] status = null;
    try {
      status = filter == null ? fs.listStatus(dir) : fs.listStatus(dir, filter);
    } catch (FileNotFoundException fnfe) {
      // if directory doesn't exist, return null
      if (LOG.isTraceEnabled()) {
        LOG.trace(dir + " doesn't exist");
      }
    }

    if (status == null || status.length < 1) {
      return null;
    }

    return status;
  }

  /**
   * Return the 'path' component of a Path. In Hadoop, Path is a URI. This method returns the 'path'
   * component of a Path's URI: e.g. If a Path is
   * <code>hdfs://example.org:9000/hbase_trunk/TestTable/compaction.dir</code>, this method returns
   * <code>/hbase_trunk/TestTable/compaction.dir</code>. This method is useful if you want to print
   * out a Path without qualifying Filesystem instance.
   * @param p file system Path whose 'path' component we are to return.
   * @return Path portion of the Filesystem
   */
  public static String getPath(Path p) {
    return p.toUri().getPath();
  }

  /**
   * Given the backup root dir and the backup id, return the log file location for an incremental
   * backup.
   * @param backupRootDir backup root directory
   * @param backupId      backup id
   * @return logBackupDir: ".../user/biadmin/backup1/WALs/backup_1396650096738"
   */
  public static String getLogBackupDir(String backupRootDir, String backupId) {
    return backupRootDir + Path.SEPARATOR + backupId + Path.SEPARATOR
      + HConstants.HREGION_LOGDIR_NAME;
  }

  /**
   * Loads all backup history as stored in files on the given backup root path.
   * @return all backup history, from newest (most recent) to oldest (least recent)
   */
  private static List<BackupInfo> getHistory(Configuration conf, Path backupRootPath)
    throws IOException {
    // Get all (n) history from backup root destination

    FileSystem fs = FileSystem.get(backupRootPath.toUri(), conf);
    RemoteIterator<LocatedFileStatus> it;
    try {
      it = fs.listLocatedStatus(backupRootPath);
    } catch (FileNotFoundException e) {
      return Collections.emptyList();
    }

    List<BackupInfo> infos = new ArrayList<>();
    while (it.hasNext()) {
      LocatedFileStatus lfs = it.next();

      if (!lfs.isDirectory()) {
        continue;
      }

      String backupId = lfs.getPath().getName();
      try {
        BackupInfo info = loadBackupInfo(backupRootPath, backupId, fs);
        infos.add(info);
      } catch (IOException e) {
        LOG.error("Can not load backup info from: " + lfs.getPath(), e);
      }
    }
    // Sort
    infos.sort(Comparator.<BackupInfo> naturalOrder().reversed());
    return infos;
  }

  /**
   * Loads all backup history as stored in files on the given backup root path, and returns the
   * first n entries matching all given filters.
   * @return (subset of) backup history, from newest (most recent) to oldest (least recent)
   */
  public static List<BackupInfo> getHistory(Configuration conf, int n, Path backupRootPath,
    BackupInfo.Filter... filters) throws IOException {
    List<BackupInfo> infos = getHistory(conf, backupRootPath);

    Predicate<BackupInfo> combinedPredicate = Stream.of(filters)
      .map(filter -> (Predicate<BackupInfo>) filter).reduce(Predicate::and).orElse(x -> true);

    return infos.stream().filter(combinedPredicate).limit(n).toList();
  }

  public static BackupInfo loadBackupInfo(Path backupRootPath, String backupId, FileSystem fs)
    throws IOException {
    Path backupPath = new Path(backupRootPath, backupId);

    RemoteIterator<LocatedFileStatus> it = fs.listFiles(backupPath, true);
    while (it.hasNext()) {
      LocatedFileStatus lfs = it.next();
      if (lfs.getPath().getName().equals(BackupManifest.MANIFEST_FILE_NAME)) {
        // Load BackupManifest
        BackupManifest manifest = new BackupManifest(fs, lfs.getPath().getParent());
        BackupInfo info = manifest.toBackupInfo();
        return info;
      }
    }
    return null;
  }

  /**
   * Create restore request.
   * @param backupRootDir backup root dir
   * @param backupId      backup id
   * @param check         check only
   * @param fromTables    table list from
   * @param toTables      table list to
   * @param isOverwrite   overwrite data
   * @return request obkect
   */
  public static RestoreRequest createRestoreRequest(String backupRootDir, String backupId,
    boolean check, TableName[] fromTables, TableName[] toTables, boolean isOverwrite) {
    return createRestoreRequest(backupRootDir, backupId, check, fromTables, toTables, isOverwrite,
      false);
  }

  public static RestoreRequest createRestoreRequest(String backupRootDir, String backupId,
    boolean check, TableName[] fromTables, TableName[] toTables, boolean isOverwrite,
    boolean isKeepOriginalSplits) {
    RestoreRequest.Builder builder = new RestoreRequest.Builder();
    RestoreRequest request = builder.withBackupRootDir(backupRootDir).withBackupId(backupId)
      .withCheck(check).withFromTables(fromTables).withToTables(toTables).withOverwrite(isOverwrite)
      .withKeepOriginalSplits(isKeepOriginalSplits).build();
    return request;
  }

  public static boolean validate(List<TableName> tables, BackupManifest backupManifest,
    Configuration conf) throws IOException {
    boolean isValid = true;

    for (TableName table : tables) {
      TreeSet<BackupImage> imageSet = new TreeSet<>();

      ArrayList<BackupImage> depList = backupManifest.getDependentListByTable(table);
      if (depList != null && !depList.isEmpty()) {
        imageSet.addAll(depList);
      }

      LOG.info("Dependent image(s) from old to new:");
      for (BackupImage image : imageSet) {
        String imageDir =
          HBackupFileSystem.getTableBackupDir(image.getRootDir(), image.getBackupId(), table);
        if (!BackupUtils.checkPathExist(imageDir, conf)) {
          LOG.error("ERROR: backup image does not exist: " + imageDir);
          isValid = false;
          break;
        }
        LOG.info("Backup image: " + image.getBackupId() + " for '" + table + "' is available");
      }
    }
    return isValid;
  }

  public static Path getBulkOutputDir(Path restoreRootDir, String tableName, Configuration conf,
    boolean deleteOnExit) throws IOException {
    FileSystem fs = restoreRootDir.getFileSystem(conf);
    Path path = new Path(restoreRootDir,
      "bulk_output-" + tableName + "-" + EnvironmentEdgeManager.currentTime());
    if (deleteOnExit) {
      fs.deleteOnExit(path);
    }
    return path;
  }

  public static Path getBulkOutputDir(Path restoreRootDir, String tableName, Configuration conf)
    throws IOException {
    return getBulkOutputDir(restoreRootDir, tableName, conf, true);
  }

  public static Path getBulkOutputDir(String tableName, Configuration conf, boolean deleteOnExit)
    throws IOException {
    FileSystem fs = FileSystem.get(conf);
    return getBulkOutputDir(getTmpRestoreOutputDir(fs, conf), tableName, conf, deleteOnExit);
  }

  /**
   * Build temporary output path
   * @param fs   filesystem for default output dir
   * @param conf configuration
   * @return output path
   */
  public static Path getTmpRestoreOutputDir(FileSystem fs, Configuration conf) {
    String tmp =
      conf.get(HConstants.TEMPORARY_FS_DIRECTORY_KEY, fs.getHomeDirectory() + "/hbase-staging");
    return new Path(tmp);
  }

  public static String getFileNameCompatibleString(TableName table) {
    return table.getNamespaceAsString() + "-" + table.getQualifierAsString();
  }

  public static boolean failed(int result) {
    return result != 0;
  }

  public static boolean succeeded(int result) {
    return result == 0;
  }

  public static BulkLoadHFiles createLoader(Configuration config) {
    // set configuration for restore:
    // LoadIncrementalHFile needs more time
    // <name>hbase.rpc.timeout</name> <value>600000</value>
    // calculates
    Configuration conf = new Configuration(config);
    conf.setInt(HConstants.HBASE_RPC_TIMEOUT_KEY, MILLISEC_IN_HOUR);

    // By default, it is 32 and loader will fail if # of files in any region exceed this
    // limit. Bad for snapshot restore.
    conf.setInt(BulkLoadHFiles.MAX_FILES_PER_REGION_PER_FAMILY, Integer.MAX_VALUE);
    conf.set(BulkLoadHFiles.IGNORE_UNMATCHED_CF_CONF_KEY, "yes");
    return BulkLoadHFiles.create(conf);
  }

  public static String findMostRecentBackupId(String[] backupIds) {
    long recentTimestamp = Long.MIN_VALUE;
    for (String backupId : backupIds) {
      long ts = Long.parseLong(Iterators.get(Splitter.on('_').split(backupId).iterator(), 1));
      if (ts > recentTimestamp) {
        recentTimestamp = ts;
      }
    }
    return BackupRestoreConstants.BACKUPID_PREFIX + recentTimestamp;
  }

  /**
   * roll WAL writer for all region servers and record the newest log roll result
   */
  public static void logRoll(Connection conn, String backupRootDir, Configuration conf)
    throws IOException {
    boolean legacy = conf.getBoolean("hbase.backup.logroll.legacy.used", false);
    if (legacy) {
      logRollV1(conn, backupRootDir);
    } else {
      logRollV2(conn, backupRootDir);
    }
  }

  private static void logRollV1(Connection conn, String backupRootDir) throws IOException {
    try (Admin admin = conn.getAdmin()) {
      admin.execProcedure(LogRollMasterProcedureManager.ROLLLOG_PROCEDURE_SIGNATURE,
        LogRollMasterProcedureManager.ROLLLOG_PROCEDURE_NAME,
        ImmutableMap.of("backupRoot", backupRootDir));
    }
  }

  private static void logRollV2(Connection conn, String backupRootDir) throws IOException {
    BackupSystemTable backupSystemTable = new BackupSystemTable(conn);
    HashMap<String, Long> lastLogRollResult =
      backupSystemTable.readRegionServerLastLogRollResult(backupRootDir);
    try (Admin admin = conn.getAdmin()) {
      Map<ServerName, Long> newLogRollResult = admin.rollAllWALWriters();

      for (Map.Entry<ServerName, Long> entry : newLogRollResult.entrySet()) {
        ServerName serverName = entry.getKey();
        long newHighestWALFilenum = entry.getValue();

        String address = serverName.getAddress().toString();
        Long lastHighestWALFilenum = lastLogRollResult.get(address);
        if (lastHighestWALFilenum != null && lastHighestWALFilenum > newHighestWALFilenum) {
          LOG.warn("Won't update last roll log result for server {}: current = {}, new = {}",
            serverName, lastHighestWALFilenum, newHighestWALFilenum);
        } else {
          backupSystemTable.writeRegionServerLastLogRollResult(address, newHighestWALFilenum,
            backupRootDir);
          if (LOG.isDebugEnabled()) {
            LOG.debug("updated last roll log result for {} from {} to {}", serverName,
              lastHighestWALFilenum, newHighestWALFilenum);
          }
        }
      }
    }
  }

  /**
   * Calculates the replication checkpoint timestamp used for continuous backup.
   * <p>
   * A replication checkpoint is the earliest timestamp across all region servers such that every
   * WAL entry before that point is known to be replicated to the target system. This is essential
   * for features like Point-in-Time Restore (PITR) and incremental backups, where we want to
   * confidently restore data to a consistent state without missing updates.
   * <p>
   * The checkpoint is calculated using a combination of:
   * <ul>
   * <li>The start timestamps of WAL files currently being replicated for each server.</li>
   * <li>The latest successfully replicated timestamp recorded by the replication marker chore.</li>
   * </ul>
   * <p>
   * We combine these two sources to handle the following challenges:
   * <ul>
   * <li><b>Stale WAL start times:</b> If replication traffic is low or WALs are long-lived, the
   * replication offset may point to the same WAL for a long time, resulting in stale timestamps
   * that underestimate progress. This could delay PITR unnecessarily.</li>
   * <li><b>Limitations of marker-only tracking:</b> The replication marker chore stores the last
   * successfully replicated timestamp per region server in a system table. However, this data may
   * become stale if the server goes offline or region ownership changes. For example, if a region
   * initially belonged to rs1 and was later moved to rs4 due to re-balancing, rs1’s marker would
   * persist even though it no longer holds any regions. Relying solely on these stale markers could
   * lead to incorrect or outdated checkpoints.</li>
   * </ul>
   * <p>
   * To handle these limitations, the method:
   * <ol>
   * <li>Verifies that the continuous backup peer exists to ensure replication is enabled.</li>
   * <li>Retrieves WAL replication queue information for the peer, collecting WAL start times per
   * region server. This gives us a lower bound for replication progress.</li>
   * <li>Reads the marker chore's replicated timestamps from the backup system table.</li>
   * <li>For servers found in both sources, if the marker timestamp is more recent than the WAL's
   * start timestamp, we use the marker (since replication has progressed beyond the WAL).</li>
   * <li>We discard marker entries for region servers that are not present in WAL queues, assuming
   * those servers are no longer relevant (e.g., decommissioned or reassigned).</li>
   * <li>The checkpoint is the minimum of all chosen timestamps — i.e., the slowest replicating
   * region server.</li>
   * <li>Finally, we persist the updated marker information to include any newly participating
   * region servers.</li>
   * </ol>
   * <p>
   * Note: If the replication marker chore is disabled, we fall back to using only the WAL start
   * times. This ensures correctness but may lead to conservative checkpoint estimates during idle
   * periods.
   * @param conn the HBase connection
   * @return the calculated replication checkpoint timestamp
   * @throws IOException if reading replication queues or updating the backup system table fails
   */
  public static long getReplicationCheckpoint(Connection conn) throws IOException {
    Configuration conf = conn.getConfiguration();
    long checkpoint = EnvironmentEdgeManager.getDelegate().currentTime();

    // Step 1: Ensure the continuous backup replication peer exists
    if (!continuousBackupReplicationPeerExists(conn.getAdmin())) {
      String msg = "Replication peer '" + CONTINUOUS_BACKUP_REPLICATION_PEER
        + "' not found. Continuous backup not enabled.";
      LOG.error(msg);
      throw new IOException(msg);
    }

    // Step 2: Get all replication queues for the continuous backup peer
    ReplicationQueueStorage queueStorage =
      ReplicationStorageFactory.getReplicationQueueStorage(conn, conf);

    List<ReplicationQueueId> queueIds;
    try {
      queueIds = queueStorage.listAllQueueIds(CONTINUOUS_BACKUP_REPLICATION_PEER);
    } catch (ReplicationException e) {
      String msg = "Failed to retrieve replication queue IDs for peer '"
        + CONTINUOUS_BACKUP_REPLICATION_PEER + "'";
      LOG.error(msg, e);
      throw new IOException(msg, e);
    }

    if (queueIds.isEmpty()) {
      String msg = "Replication peer '" + CONTINUOUS_BACKUP_REPLICATION_PEER + "' has no queues. "
        + "This may indicate that continuous backup replication is not initialized correctly.";
      LOG.error(msg);
      throw new IOException(msg);
    }

    // Step 3: Build a map of ServerName -> WAL start timestamp (lowest seen per server)
    Map<ServerName, Long> serverToCheckpoint = new HashMap<>();
    for (ReplicationQueueId queueId : queueIds) {
      Map<String, ReplicationGroupOffset> offsets;
      try {
        offsets = queueStorage.getOffsets(queueId);
      } catch (ReplicationException e) {
        String msg = "Failed to fetch WAL offsets for replication queue: " + queueId;
        LOG.error(msg, e);
        throw new IOException(msg, e);
      }

      for (ReplicationGroupOffset offset : offsets.values()) {
        String walFile = offset.getWal();
        long ts = AbstractFSWALProvider.getTimestamp(walFile); // WAL creation time
        ServerName server = queueId.getServerName();
        // Store the minimum timestamp per server (ts - 1 to avoid edge boundary issues)
        serverToCheckpoint.merge(server, ts - 1, Math::min);
      }
    }

    // Step 4: If replication markers are enabled, overlay fresher timestamps from backup system
    // table
    boolean replicationMarkerEnabled =
      conf.getBoolean(REPLICATION_MARKER_ENABLED_KEY, REPLICATION_MARKER_ENABLED_DEFAULT);
    if (replicationMarkerEnabled) {
      try (BackupSystemTable backupSystemTable = new BackupSystemTable(conn)) {
        Map<ServerName, Long> markerTimestamps = backupSystemTable.getBackupCheckpointTimestamps();

        for (Map.Entry<ServerName, Long> entry : markerTimestamps.entrySet()) {
          ServerName server = entry.getKey();
          long markerTs = entry.getValue();

          // If marker timestamp is newer, override
          if (serverToCheckpoint.containsKey(server)) {
            long current = serverToCheckpoint.get(server);
            if (markerTs > current) {
              serverToCheckpoint.put(server, markerTs);
            }
          } else {
            // This server is no longer active (e.g., RS moved or removed); skip
            if (LOG.isDebugEnabled()) {
              LOG.debug("Skipping replication marker timestamp for inactive server: {}", server);
            }
          }
        }

        // Step 5: Persist current server timestamps into backup system table
        for (Map.Entry<ServerName, Long> entry : serverToCheckpoint.entrySet()) {
          backupSystemTable.updateBackupCheckpointTimestamp(entry.getKey(), entry.getValue());
        }
      }
    } else {
      LOG.warn(
        "Replication marker chore is disabled. Using WAL-based timestamps only for checkpoint calculation.");
    }

    // Step 6: Calculate final checkpoint as minimum timestamp across all active servers
    for (long ts : serverToCheckpoint.values()) {
      checkpoint = Math.min(checkpoint, ts);
    }

    return checkpoint;
  }

  private static boolean continuousBackupReplicationPeerExists(Admin admin) throws IOException {
    return admin.listReplicationPeers().stream()
      .anyMatch(peer -> peer.getPeerId().equals(CONTINUOUS_BACKUP_REPLICATION_PEER));
  }

  /**
   * Convert dayInMillis to "yyyy-MM-dd" format
   */
  public static String formatToDateString(long dayInMillis) {
    SimpleDateFormat dateFormat = new SimpleDateFormat(DATE_FORMAT);
    dateFormat.setTimeZone(TimeZone.getTimeZone("UTC"));
    return dateFormat.format(new Date(dayInMillis));
  }

  /**
   * Fetches bulkload filepaths based on the given time range from backup WAL directory.
   */
  public static List<Path> collectBulkFiles(Connection conn, TableName sourceTable,
    TableName targetTable, long startTime, long endTime, Path restoreRootDir, List<String> walDirs)
    throws IOException {

    if (walDirs.isEmpty()) {
      String walBackupDir = conn.getConfiguration().get(CONF_CONTINUOUS_BACKUP_WAL_DIR);
      if (Strings.isNullOrEmpty(walBackupDir)) {
        throw new IOException(
          "WAL backup directory is not configured " + CONF_CONTINUOUS_BACKUP_WAL_DIR);
      }
      Path walDirPath = new Path(walBackupDir);
      walDirs =
        BackupUtils.getValidWalDirs(conn.getConfiguration(), walDirPath, startTime, endTime);
    }

    if (walDirs.isEmpty()) {
      LOG.warn("No valid WAL directories found for range {} - {}. Skipping bulk-file collection.",
        startTime, endTime);
      return Collections.emptyList();
    }

    LOG.info(
      "Starting WAL bulk-file collection for source: {}, target: {}, time range: {} - {}, WAL "
        + "backup dir: {}, restore root: {}",
      sourceTable, targetTable, startTime, endTime, walDirs, restoreRootDir);
    String walDirsCsv = String.join(",", walDirs);

    return BulkFilesCollector.collectFromWalDirs(HBaseConfiguration.create(conn.getConfiguration()),
      walDirsCsv, restoreRootDir, sourceTable, targetTable, startTime, endTime);
  }

  /**
   * Fetches valid WAL directories based on the given time range.
   */
  public static List<String> getValidWalDirs(Configuration conf, Path walBackupDir, long startTime,
    long endTime) throws IOException {
    FileSystem backupFs = FileSystem.get(walBackupDir.toUri(), conf);
    FileStatus[] dayDirs = backupFs.listStatus(new Path(walBackupDir, WALS_DIR));

    List<String> validDirs = new ArrayList<>();
    SimpleDateFormat dateFormat = new SimpleDateFormat(DATE_FORMAT);
    dateFormat.setTimeZone(TimeZone.getTimeZone(ZoneOffset.UTC));

    for (FileStatus dayDir : dayDirs) {
      if (!dayDir.isDirectory()) {
        continue; // Skip files, only process directories
      }

      String dirName = dayDir.getPath().getName();
      try {
        Date dirDate = dateFormat.parse(dirName);
        long dirStartTime = dirDate.getTime(); // Start of that day (00:00:00)
        long dirEndTime = dirStartTime + ONE_DAY_IN_MILLISECONDS - 1; // End time of day (23:59:59)

        // Check if this day's WAL files overlap with the required time range
        if (dirEndTime >= startTime && dirStartTime <= endTime) {
          validDirs.add(dayDir.getPath().toString());
        }
      } catch (ParseException e) {
        LOG.warn("Skipping invalid directory name: {}", dirName, e);
      }
    }
    return validDirs;
  }
}