/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.server.ft.journal;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

import alluxio.AlluxioTestDirectory;
import alluxio.AlluxioURI;
import alluxio.ClientContext;
import alluxio.ConfigurationRule;
import alluxio.Constants;
import alluxio.client.WriteType;
import alluxio.client.block.BlockMasterClient;
import alluxio.client.block.RetryHandlingBlockMasterClient;
import alluxio.client.file.FileInStream;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemTestUtils;
import alluxio.client.file.URIStatus;
import alluxio.client.meta.MetaMasterClient;
import alluxio.client.meta.RetryHandlingMetaMasterClient;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.BackupAbortedException;
import alluxio.exception.status.FailedPreconditionException;
import alluxio.grpc.BackupPOptions;
import alluxio.grpc.BackupPRequest;
import alluxio.grpc.BackupState;
import alluxio.grpc.CreateDirectoryPOptions;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.ListStatusPOptions;
import alluxio.grpc.LoadMetadataPType;
import alluxio.grpc.WritePType;
import alluxio.master.MasterClientContext;
import alluxio.master.NoopMaster;
import alluxio.master.journal.JournalReader;
import alluxio.master.journal.JournalType;
import alluxio.master.journal.ufs.UfsJournal;
import alluxio.master.journal.ufs.UfsJournalLogWriter;
import alluxio.master.journal.ufs.UfsJournalReader;
import alluxio.master.metastore.MetastoreType;
import alluxio.multi.process.MultiProcessCluster;
import alluxio.multi.process.PortCoordination;
import alluxio.proto.journal.Journal;
import alluxio.testutils.AlluxioOperationThread;
import alluxio.testutils.BaseIntegrationTest;
import alluxio.util.CommonUtils;
import alluxio.util.URIUtils;
import alluxio.util.WaitForOptions;
import alluxio.wire.BackupStatus;

import org.junit.After;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

/**
 * Integration test for backing up and restoring alluxio master.
 */
@Ignore("In Dora, Client does not use Master/Journal services.")
public final class JournalBackupIntegrationTest extends BaseIntegrationTest {
  public MultiProcessCluster mCluster;
  private static final int GET_PRIMARY_INDEX_TIMEOUT_MS = 30000;
  private static final int PRIMARY_KILL_TIMEOUT_MS = 30000;
  private static final int WAIT_NODES_REGISTERED_MS = 30000;

  @Rule
  public ConfigurationRule mConf = new ConfigurationRule(new HashMap<PropertyKey, Object>() {
    {
      put(PropertyKey.USER_METRICS_COLLECTION_ENABLED, false);
    }
  }, Configuration.modifiableGlobal());

  @After
  public void after() throws Exception {
    if (mCluster != null) {
      mCluster.destroy();
    }
  }

  // This test needs to stop and start master many times, so it can take up to a minute to complete.
  @Test
  public void backupRestoreZk() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_RESTORE_ZK)
        .setClusterName("backupRestoreZk")
        .setNumMasters(3)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Masters become primary faster
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "3sec")
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    backupRestoreTest(true);
  }

  @Test
  public void backupRestoreMetastore_Heap() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_RESTORE_METASSTORE_HEAP)
        .setClusterName("backupRestoreMetastore_Heap")
        .setNumMasters(1)
        .setNumWorkers(1)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Masters become primary faster
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
        .addProperty(PropertyKey.MASTER_METASTORE, MetastoreType.HEAP)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    backupRestoreMetaStoreTest();
  }

  @Test
  public void backupRestoreMetastore_Rocks() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_RESTORE_METASSTORE_ROCKS)
        .setClusterName("backupRestoreMetastore_Rocks")
        .setNumMasters(1)
        .setNumWorkers(1)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Masters become primary faster
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
        .addProperty(PropertyKey.MASTER_METASTORE, MetastoreType.ROCKS)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    backupRestoreMetaStoreTest();
  }

  @Test
  public void backupRestoreMetastore_InodeRocksBlockHeap() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_RESTORE_METASSTORE_ROCKS)
        .setClusterName("backupRestoreMetastore_InodeRocksBlockHeap")
        .setNumMasters(1)
        .setNumWorkers(1)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Masters become primary faster
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
        .addProperty(PropertyKey.MASTER_INODE_METASTORE, MetastoreType.ROCKS)
        .addProperty(PropertyKey.MASTER_BLOCK_METASTORE, MetastoreType.HEAP)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    backupRestoreMetaStoreTest();
  }

  @Test
  public void backupRestoreMetastore_InodeHeapBlockRocks() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_RESTORE_METASSTORE_ROCKS)
        .setClusterName("backupRestoreMetastore_InodeHeapBlockRocks")
        .setNumMasters(1)
        .setNumWorkers(1)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Masters become primary faster
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
        .addProperty(PropertyKey.MASTER_INODE_METASTORE, MetastoreType.HEAP)
        .addProperty(PropertyKey.MASTER_BLOCK_METASTORE, MetastoreType.ROCKS)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    backupRestoreMetaStoreTest();
  }

  @Test
  public void emergencyBackup() throws Exception {
    emergencyBackupCore(1);
  }

  @Test
  public void emergencyBackupHA() throws Exception {
    emergencyBackupCore(3);
  }

  private void emergencyBackupCore(int numMasters) throws Exception {
    TemporaryFolder backupFolder = new TemporaryFolder();
    backupFolder.create();
    List<PortCoordination.ReservedPort> ports1 = numMasters > 1
        ? PortCoordination.BACKUP_EMERGENCY_HA_1 : PortCoordination.BACKUP_EMERGENCY_1;
    String clusterName1 = numMasters > 1 ? "emergencyBackup_HA_1" : "emergencyBackup_1";
    mCluster = MultiProcessCluster.newBuilder(ports1)
        .setClusterName(clusterName1)
        .setNumMasters(numMasters)
        .setNumWorkers(0)
        // Masters become primary faster, will be ignored in non HA case
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        .addProperty(PropertyKey.MASTER_METASTORE, MetastoreType.ROCKS)
        .addProperty(PropertyKey.MASTER_BACKUP_DIRECTORY, backupFolder.getRoot())
        .addProperty(PropertyKey.MASTER_JOURNAL_BACKUP_WHEN_CORRUPTED, true)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    mCluster.start();
    final int numFiles = 10;
    // create normal uncorrupted journal
    for (int i = 0; i < numFiles; i++) {
      mCluster.getFileSystemClient().createFile(new AlluxioURI("/normal-file-" + i));
    }
    mCluster.stopMasters();
    // corrupt journal
    URI journalLocation = new URI(mCluster.getJournalDir());
    UfsJournal fsMaster =
        new UfsJournal(URIUtils.appendPathOrDie(journalLocation, "FileSystemMaster"),
            new NoopMaster(), 0, Collections::emptySet);
    fsMaster.start();
    fsMaster.gainPrimacy();
    long nextSN = 0;
    try (UfsJournalReader reader = new UfsJournalReader(fsMaster, true)) {
      while (reader.advance() != JournalReader.State.DONE) {
        nextSN++;
      }
    }
    try (UfsJournalLogWriter writer = new UfsJournalLogWriter(fsMaster, nextSN)) {
      Journal.JournalEntry entry = Journal.JournalEntry.newBuilder()
          .setSequenceNumber(nextSN)
          .setDeleteFile(alluxio.proto.journal.File.DeleteFileEntry.newBuilder()
              .setId(4563728) // random non-zero ID number (zero would delete the root)
              .setPath("/nonexistant")
              .build())
          .build();
      writer.write(entry);
      writer.flush();
    }
    // this should fail and create backup(s)
    mCluster.startMasters();
    // wait for backup file(s) to be created
    // successful backups leave behind one .gz file and one .gz.complete file
    CommonUtils.waitFor("backup file(s) to be created automatically",
        () -> 2 * numMasters == Objects.requireNonNull(backupFolder.getRoot().list()).length,
        WaitForOptions.defaults().setInterval(500).setTimeoutMs(30_000));
    List<String> backupFiles = Arrays.stream(Objects.requireNonNull(backupFolder.getRoot().list()))
            .filter(s -> s.endsWith(".gz")).collect(Collectors.toList());
    assertEquals(numMasters, backupFiles.size());
    // create new cluster
    List<PortCoordination.ReservedPort> ports2 = numMasters > 1
        ? PortCoordination.BACKUP_EMERGENCY_HA_2 : PortCoordination.BACKUP_EMERGENCY_2;
    String clusterName2 = numMasters > 1 ? "emergencyBackup_HA_2" : "emergencyBackup_2";
    // verify that every backup contains all the entries
    for (String backupFile : backupFiles) {
      mCluster = MultiProcessCluster.newBuilder(ports2)
          .setClusterName(String.format("%s_%s", clusterName2, backupFile))
          .setNumMasters(numMasters)
          .setNumWorkers(0)
          .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
          .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
          // change metastore type to ensure backup is independent of metastore type
          .addProperty(PropertyKey.MASTER_METASTORE, MetastoreType.HEAP)
          .addProperty(PropertyKey.MASTER_JOURNAL_INIT_FROM_BACKUP,
              Paths.get(backupFolder.getRoot().toString(), backupFile))
          .build();
      mCluster.start();
      // test that the files were restored from the backup properly
      for (int i = 0; i < numFiles; i++) {
        boolean exists = mCluster.getFileSystemClient().exists(new AlluxioURI("/normal-file-" + i));
        assertTrue(exists);
      }
      mCluster.stopMasters();
      mCluster.notifySuccess();
    }
    backupFolder.delete();
  }

  @Test
  public void syncRootOnBackupRestore() throws Exception {
    syncLsTestCore(true);
  }

  @Test
  public void doNotSyncRootOnBackupRestore() throws Exception {
    syncLsTestCore(false);
  }

  private void syncLsTestCore(boolean syncRootOnRestore) throws Exception {
    TemporaryFolder temporaryFolder = new TemporaryFolder();
    temporaryFolder.create();
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_SYNC_ON_RESTORE)
        .setClusterName("syncRootOnBackupRestore")
        .setNumMasters(1)
        .setNumWorkers(1)
        .addProperty(PropertyKey.MASTER_BACKUP_DIRECTORY, temporaryFolder.getRoot())
        .addProperty(PropertyKey.USER_FILE_WRITE_TYPE_DEFAULT, WriteType.CACHE_THROUGH)
        .addProperty(PropertyKey.MASTER_JOURNAL_SYNC_ROOT_AFTER_INIT_FROM_BACKUP, syncRootOnRestore)
        // this test uses NEVER as the metadata load type to ensure that the UFS sync is
        // performed due to the invalidation associated with restoring a backup, as opposed to
        // performed automatically under some other metadata load types
        .addProperty(PropertyKey.USER_FILE_METADATA_LOAD_TYPE, LoadMetadataPType.NEVER)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    mCluster.start();

    mCluster.getFileSystemClient().createDirectory(new AlluxioURI("/in_backup"));
    BackupStatus backup =
        mCluster.getMetaMasterClient().backup(BackupPRequest.getDefaultInstance());
    UUID id = backup.getBackupId();
    while (backup.getState() != BackupState.Completed) {
      backup = mCluster.getMetaMasterClient().getBackupStatus(id);
    }
    mCluster.getFileSystemClient().createDirectory(new AlluxioURI("/NOT_in_backup"));
    mCluster.stopMasters();
    // give it an empty journal and start from backup
    mCluster.updateMasterConf(PropertyKey.MASTER_JOURNAL_FOLDER,
        temporaryFolder.newFolder().getAbsolutePath());
    mCluster.updateMasterConf(PropertyKey.MASTER_METASTORE_DIR,
        temporaryFolder.newFolder().getAbsolutePath());
    mCluster.updateMasterConf(PropertyKey.MASTER_JOURNAL_INIT_FROM_BACKUP,
        backup.getBackupUri().getPath());
    mCluster.startMasters();
    List<URIStatus> statuses = mCluster.getFileSystemClient().listStatus(new AlluxioURI("/"));
    int expected = syncRootOnRestore ? 2 : 1;
    assertEquals(expected, statuses.size());
    mCluster.notifySuccess();
    temporaryFolder.delete();
  }

  @Test
  public void syncContentsOnBackupRestore() throws Exception {
    syncContentsTestCore(true);
  }

  @Test
  public void doNotSyncContentsOnBackupRestore() throws Exception {
    syncContentsTestCore(false);
  }

  private void syncContentsTestCore(boolean syncRootOnRestore) throws Exception {
    TemporaryFolder temporaryFolder = new TemporaryFolder();
    temporaryFolder.create();
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_CONTENT_ON_RESTORE)
        .setClusterName("syncContentOnBackupRestore")
        .setNumMasters(1)
        .setNumWorkers(1)
        .addProperty(PropertyKey.MASTER_BACKUP_DIRECTORY, temporaryFolder.getRoot())
        .addProperty(PropertyKey.USER_FILE_WRITE_TYPE_DEFAULT, WriteType.CACHE_THROUGH)
        .addProperty(PropertyKey.MASTER_JOURNAL_SYNC_ROOT_AFTER_INIT_FROM_BACKUP, syncRootOnRestore)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    mCluster.start();

    AlluxioURI f = new AlluxioURI("/in_backup");
    String originalData = "data";
    try (FileOutStream inBackup = mCluster.getFileSystemClient().createFile(f)) {
      inBackup.write(originalData.getBytes());
    }

    BackupStatus backup =
        mCluster.getMetaMasterClient().backup(BackupPRequest.getDefaultInstance());
    UUID id = backup.getBackupId();
    while (backup.getState() != BackupState.Completed) {
      backup = mCluster.getMetaMasterClient().getBackupStatus(id);
    }

    mCluster.getFileSystemClient().delete(f);
    String modifiedData = "modified data";
    try (FileOutStream overwriteInBackup = mCluster.getFileSystemClient().createFile(f)) {
      overwriteInBackup.write(modifiedData.getBytes());
    }
    mCluster.stopMasters();
    // give it an empty journal and start from backup
    mCluster.updateMasterConf(PropertyKey.MASTER_JOURNAL_FOLDER,
        temporaryFolder.newFolder().getAbsolutePath());
    mCluster.updateMasterConf(PropertyKey.MASTER_METASTORE_DIR,
        temporaryFolder.newFolder().getAbsolutePath());
    mCluster.updateMasterConf(PropertyKey.MASTER_JOURNAL_INIT_FROM_BACKUP,
        backup.getBackupUri().getPath());
    mCluster.startMasters();
    mCluster.waitForAllNodesRegistered(WAIT_NODES_REGISTERED_MS);

    try (FileInStream inStream = mCluster.getFileSystemClient().openFile(f)) {
      byte[] bytes = new byte[modifiedData.length()];
      int read = inStream.read(bytes);
      // if the invalidation is not set during the backup restore only the length of the old
      // contents ("data") will be read (instead of reading the new contents "modified data")
      int expectedLength = syncRootOnRestore ? modifiedData.length() : originalData.length();
      String expectedStart = syncRootOnRestore ? modifiedData :
          modifiedData.substring(0, originalData.length());
      assertEquals(expectedLength, read);
      assertTrue(new String(bytes).startsWith(expectedStart));
    }
    mCluster.notifySuccess();
    temporaryFolder.delete();
  }

  // This test needs to stop and start master many times, so it can take up to a minute to complete.
  @Test
  public void backupRestoreEmbedded() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_RESTORE_EMBEDDED)
        .setClusterName("backupRestoreEmbedded")
        .setNumMasters(3)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.EMBEDDED)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    backupRestoreTest(true);
  }

  @Test
  public void backupRestoreSingleMaster() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_RESTORE_SINGLE)
        .setClusterName("backupRestoreSingle")
        .setNumMasters(1)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Disable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, false)
        .build();
    backupRestoreTest(false);
  }

  // Tests various protocols and configurations for backup delegation.
  @Test
  public void backupDelegationProtocol() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_DELEGATION_PROTOCOL)
        .setClusterName("backupDelegationProtocol")
        .setNumMasters(3)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Masters become primary faster
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
        // For faster backup role handshake.
        .addProperty(PropertyKey.MASTER_BACKUP_CONNECT_INTERVAL_MIN, "100ms")
        .addProperty(PropertyKey.MASTER_BACKUP_CONNECT_INTERVAL_MAX, "100ms")
        // Enable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, true)
        .build();

    File backups = AlluxioTestDirectory.createTemporaryDirectory("backups");
    mCluster.start();

    // Validate backup works with delegation.
    waitForBackup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
        .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build());

    // Kill the primary.
    int primaryIdx = mCluster.getPrimaryMasterIndex(GET_PRIMARY_INDEX_TIMEOUT_MS);
    mCluster.waitForAndKillPrimaryMaster(PRIMARY_KILL_TIMEOUT_MS);

    // Validate backup works again after leader fail-over.
    waitForBackup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
        .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build());

    // Continue testing with 2 masters...
    // Find standby master index.
    int newPrimaryIdx = mCluster.getPrimaryMasterIndex(GET_PRIMARY_INDEX_TIMEOUT_MS);
    int followerIdx = (newPrimaryIdx + 1) % 2;
    if (followerIdx == primaryIdx) {
      followerIdx = (followerIdx + 1) % 2;
    }

    // Kill the follower. (only leader remains).
    mCluster.stopMaster(followerIdx);

    // Wait for a second for process to terminate properly.
    // This is so that backup request don't get delegated to follower before termination.
    Thread.sleep(1000);

    // Validate backup delegation fails.
    try {
      mCluster.getMetaMasterClient()
          .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
              .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build());
      Assert.fail("Cannot delegate backup with no followers.");
    } catch (FailedPreconditionException e) {
      // Expected to fail since there is only single master.
    }

    // Should work with "AllowLeader" backup.
    mCluster.getMetaMasterClient()
        .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
            .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false).setAllowLeader(true))
            .build());

    // Restart the follower. (1 leader 1 follower remains).
    mCluster.startMaster(followerIdx);

    // Validate backup works again.
    waitForBackup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
        .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build());

    // Schedule async backup.
    UUID backupId = mCluster.getMetaMasterClient()
        .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
            .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false).setRunAsync(true))
            .build())
        .getBackupId();
    // Wait until backup is complete.
    CommonUtils.waitFor("Backup completed.", () -> {
      try {
        return mCluster.getMetaMasterClient().getBackupStatus(backupId).getState()
            .equals(BackupState.Completed);
      } catch (Exception e) {
        throw new RuntimeException(
            String.format("Unexpected error while getting backup status: %s", e));
      }
    });

    // Schedule a local backup to overwrite latest backup Id in the current leader.
    mCluster.getMetaMasterClient()
        .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
            .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false).setAllowLeader(true))
            .build());

    // Validate old backup can still be queried.
    mCluster.getMetaMasterClient().getBackupStatus(backupId);

    mCluster.notifySuccess();
  }

  // Tests various protocols and configurations for backup delegation during fail-overs.
  @Test
  public void backupDelegationFailoverProtocol() throws Exception {
    mCluster = MultiProcessCluster.newBuilder(PortCoordination.BACKUP_DELEGATION_FAILOVER_PROTOCOL)
        .setClusterName("backupDelegationFailoverProtocol")
        .setNumMasters(2)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Masters become primary faster
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
        // For faster backup role handshake.
        .addProperty(PropertyKey.MASTER_BACKUP_CONNECT_INTERVAL_MIN, "100ms")
        .addProperty(PropertyKey.MASTER_BACKUP_CONNECT_INTERVAL_MAX, "100ms")
        // Enable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, true)
        // Set backup timeout to be shorter
        .addProperty(PropertyKey.MASTER_BACKUP_ABANDON_TIMEOUT, "3sec")
        .build();

    File backups = AlluxioTestDirectory.createTemporaryDirectory("backups");
    mCluster.start();

    // Validate backup works with delegation.
    waitForBackup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
        .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build());

    // Find standby master index.
    int primaryIdx = mCluster.getPrimaryMasterIndex(GET_PRIMARY_INDEX_TIMEOUT_MS);
    int followerIdx = (primaryIdx + 1) % 2;

    // Schedule async backup.
    UUID backupId = mCluster.getMetaMasterClient()
        .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
            .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false).setRunAsync(true))
            .build())
        .getBackupId();
    // Kill follower immediately before it sends the next heartbeat to leader.
    mCluster.stopMaster(followerIdx);
    // Wait until backup is abandoned.
    CommonUtils.waitForResult("Backup abandoned.", () -> {
      try {
        return mCluster.getMetaMasterClient().getBackupStatus(backupId);
      } catch (Exception e) {
        throw new RuntimeException(
            String.format("Unexpected error while getting backup status: %s", e));
      }
    }, (backupStatus) -> backupStatus.getError() instanceof BackupAbortedException);

    // Restart follower to restore HA.
    mCluster.startMaster(followerIdx);

    // Validate delegated backup works again.
    waitForBackup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
        .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build());

    // Schedule async backup.
    mCluster.getMetaMasterClient()
        .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
            .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false).setRunAsync(true))
            .build())
        .getBackupId();

    // Kill leader immediately before it receives the next heartbeat from backup-worker.
    mCluster.waitForAndKillPrimaryMaster(PRIMARY_KILL_TIMEOUT_MS);
    // Wait until follower steps up.
    assertEquals(mCluster.getPrimaryMasterIndex(GET_PRIMARY_INDEX_TIMEOUT_MS), followerIdx);

    // Follower should step-up without problem and accept backup requests.
    mCluster.getMetaMasterClient()
        .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
            .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false).setAllowLeader(true))
            .build());

    mCluster.notifySuccess();
  }

  @Test
  public void backupDelegationZk() throws Exception {
    MultiProcessCluster.Builder clusterBuilder = MultiProcessCluster
        .newBuilder(PortCoordination.BACKUP_DELEGATION_ZK)
        .setClusterName("backupDelegationZk")
        .setNumMasters(2)
        .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
        // Masters become primary faster
        .addProperty(PropertyKey.ZOOKEEPER_SESSION_TIMEOUT, "1sec")
        // Enable backup delegation
        .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, true)
        // For faster backup role handshake.
        .addProperty(PropertyKey.MASTER_BACKUP_CONNECT_INTERVAL_MIN, "100ms")
        .addProperty(PropertyKey.MASTER_BACKUP_CONNECT_INTERVAL_MAX, "100ms");
    backupDelegationTest(clusterBuilder);
  }

  @Test
  public void backupDelegationEmbedded() throws Exception {
    MultiProcessCluster.Builder clusterBuilder =
        MultiProcessCluster.newBuilder(PortCoordination.BACKUP_DELEGATION_EMBEDDED)
            .setClusterName("backupDelegationEmbedded")
            .setNumMasters(2)
            .addProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.EMBEDDED)
            // Enable backup delegation
            .addProperty(PropertyKey.MASTER_BACKUP_DELEGATION_ENABLED, true)
            // For faster backup role handshake.
            .addProperty(PropertyKey.MASTER_BACKUP_CONNECT_INTERVAL_MIN, "100ms")
            .addProperty(PropertyKey.MASTER_BACKUP_CONNECT_INTERVAL_MAX, "100ms");
    backupDelegationTest(clusterBuilder);
  }

  /**
   * Used to wait until successful backup.
   *
   * It tolerates {@link FailedPreconditionException} that could be thrown when backup-workers have
   * not established connection with the backup-leader yet.
   *
   * @param backupRequest the backup request
   * @return backup uri
   */
  private AlluxioURI waitForBackup(BackupPRequest backupRequest) throws Exception {
    AtomicReference<AlluxioURI> backupUriRef = new AtomicReference<>(null);
    CommonUtils.waitFor("Backup delegation to succeed.", () -> {
      try {
        backupUriRef.set(mCluster.getMetaMasterClient().backup(backupRequest).getBackupUri());
        return true;
      } catch (FailedPreconditionException e1) {
        // Expected to fail with this until backup-workers connect with backup-leader.
        return false;
      } catch (Exception e2) {
        throw new RuntimeException(
            String.format("Backup failed with unexpected error: %s", e2));
      }
    }, WaitForOptions.defaults());
    return backupUriRef.get();
  }

  private void backupRestoreTest(boolean testFailover) throws Exception {
    File backups = AlluxioTestDirectory.createTemporaryDirectory("backups");
    mCluster.start();
    List<Thread> opThreads = new ArrayList<>();
    // Run background threads to perform metadata operations while the journal backups and restores
    // are happening.
    for (int i = 0; i < 10; i++) {
      AlluxioOperationThread thread = new AlluxioOperationThread(mCluster.getFileSystemClient());
      thread.start();
      opThreads.add(thread);
    }
    try {
      FileSystem fs = mCluster.getFileSystemClient();
      MetaMasterClient metaClient = getMetaClient(mCluster);

      AlluxioURI dir1 = new AlluxioURI("/dir1");
      fs.createDirectory(dir1,
          CreateDirectoryPOptions.newBuilder().build());
      AlluxioURI backup1 = metaClient
          .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
              .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build())
          .getBackupUri();
      AlluxioURI dir2 = new AlluxioURI("/dir2");
      fs.createDirectory(dir2,
          CreateDirectoryPOptions.newBuilder().build());
      AlluxioURI backup2 = metaClient
          .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
              .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build())
          .getBackupUri();

      restartMastersFromBackup(backup2);
      assertTrue(fs.exists(dir1));
      assertTrue(fs.exists(dir2));

      restartMastersFromBackup(backup1);
      assertTrue(fs.exists(dir1));
      assertFalse(fs.exists(dir2));

      // Restart normally and make sure we remember the state from backup 1.
      mCluster.stopMasters();
      mCluster.startMasters();
      assertTrue(fs.exists(dir1));
      assertFalse(fs.exists(dir2));

      if (testFailover) {
        // Verify that failover works correctly.
        mCluster.waitForAndKillPrimaryMaster(30 * Constants.SECOND_MS);
        assertTrue(fs.exists(dir1));
        assertFalse(fs.exists(dir2));
      }

      mCluster.notifySuccess();
    } finally {
      opThreads.forEach(Thread::interrupt);
    }
  }

  private void backupDelegationTest(MultiProcessCluster.Builder clusterBuilder) throws Exception {
    // Update configuration for each master to backup to a specific folder.
    Map<Integer, Map<PropertyKey, String>> masterProps = new HashMap<>();
    File backupsParent0 = AlluxioTestDirectory.createTemporaryDirectory("backups0");
    File backupsParent1 = AlluxioTestDirectory.createTemporaryDirectory("backups1");
    Map<PropertyKey, String> master0Props = new HashMap<>();
    master0Props.put(PropertyKey.MASTER_BACKUP_DIRECTORY, backupsParent0.getAbsolutePath());
    Map<PropertyKey, String> master1Props = new HashMap<>();
    master1Props.put(PropertyKey.MASTER_BACKUP_DIRECTORY, backupsParent1.getAbsolutePath());
    masterProps.put(0, master0Props);
    masterProps.put(1, master1Props);
    clusterBuilder.setMasterProperties(masterProps);
    // Start cluster.
    mCluster = clusterBuilder.build();
    mCluster.start();

    // Delegation test can work with 2 masters only.
    assertEquals(2, mCluster.getMasterAddresses().size());

    FileSystem fs = mCluster.getFileSystemClient();

    AlluxioURI dir1 = new AlluxioURI("/dir1");
    mCluster.getFileSystemClient().createDirectory(dir1,
        CreateDirectoryPOptions.newBuilder().build());

    AlluxioURI backupUri = waitForBackup(BackupPRequest.newBuilder()
        .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(true)).build());

    int primaryIndex = mCluster.getPrimaryMasterIndex(GET_PRIMARY_INDEX_TIMEOUT_MS);
    int followerIndex = (primaryIndex + 1) % 2;

    // Validate backup is taken on follower's local path.
    assertTrue(backupUri.toString()
        .contains(masterProps.get(followerIndex).get(PropertyKey.MASTER_BACKUP_DIRECTORY)));

    // Validate backup is valid.
    restartMastersFromBackup(backupUri);
    assertTrue(fs.exists(dir1));

    mCluster.notifySuccess();
  }

  private void backupRestoreMetaStoreTest() throws Exception {
    // Directory for backups.
    File backups = AlluxioTestDirectory.createTemporaryDirectory("backups");
    mCluster.start();

    // Needs workers to be up before the test.
    mCluster.waitForAllNodesRegistered(WAIT_NODES_REGISTERED_MS);

    // Acquire clients.
    FileSystem fs = mCluster.getFileSystemClient();
    MetaMasterClient metaClient = getMetaClient(mCluster);
    BlockMasterClient blockClient = getBlockClient(mCluster);

    // Create single test file.
    String testFilePath = "/file";
    AlluxioURI testFileUri = new AlluxioURI(testFilePath);

    // Create file THROUGH.
    FileSystemTestUtils.createByteFile(fs, testFilePath, 100,
        CreateFilePOptions.newBuilder().setWriteType(WritePType.THROUGH).build());
    // Delete it from Alluxio namespace.
    fs.delete(testFileUri, DeletePOptions.newBuilder().setAlluxioOnly(true).build());
    // List status on root to bring the file's meta back.
    fs.listStatus(new AlluxioURI("/"), ListStatusPOptions.newBuilder().setRecursive(true)
        .setLoadMetadataType(LoadMetadataPType.ONCE).build());
    // Verify that file's meta is in Alluxio.
    assertNotNull(fs.getStatus(testFileUri));

    // Take a backup.
    AlluxioURI backup1 = metaClient
        .backup(BackupPRequest.newBuilder().setTargetDirectory(backups.getAbsolutePath())
            .setOptions(BackupPOptions.newBuilder().setLocalFileSystem(false)).build())
        .getBackupUri();
    // Restart with backup.
    restartMastersFromBackup(backup1);

    // Verify that file and its blocks are in Alluxio after restore.
    URIStatus fileStatus = fs.getStatus(testFileUri);
    assertNotNull(fileStatus);
    for (long blockId : fileStatus.getBlockIds()) {
      assertNotNull(blockClient.getBlockInfo(blockId));
    }
  }

  private void restartMastersFromBackup(AlluxioURI backup) throws IOException {
    mCluster.stopMasters();
    mCluster.formatJournal();
    mCluster.updateMasterConf(PropertyKey.MASTER_JOURNAL_INIT_FROM_BACKUP, backup.toString());
    mCluster.startMasters();
    mCluster.updateMasterConf(PropertyKey.MASTER_JOURNAL_INIT_FROM_BACKUP, null);
  }

  private MetaMasterClient getMetaClient(MultiProcessCluster cluster) {
    return new RetryHandlingMetaMasterClient(
        MasterClientContext.newBuilder(ClientContext.create(Configuration.global()))
            .setMasterInquireClient(cluster.getMasterInquireClient())
            .build());
  }

  private BlockMasterClient getBlockClient(MultiProcessCluster cluster) {
    return new RetryHandlingBlockMasterClient(
        MasterClientContext.newBuilder(ClientContext.create(Configuration.global()))
            .setMasterInquireClient(cluster.getMasterInquireClient()).build());
  }
}
