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

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

import alluxio.AlluxioURI;
import alluxio.AuthenticatedUserRule;
import alluxio.ClientContext;
import alluxio.Constants;
import alluxio.annotation.dora.DoraTestTodoItem;
import alluxio.client.block.BlockMasterClient;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemTestUtils;
import alluxio.client.file.FileSystemUtils;
import alluxio.client.file.URIStatus;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AlluxioException;
import alluxio.exception.FileDoesNotExistException;
import alluxio.grpc.CreateDirectoryPOptions;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.DeletePOptions;
import alluxio.grpc.ExistsPOptions;
import alluxio.grpc.FileSystemMasterCommonPOptions;
import alluxio.grpc.FreePOptions;
import alluxio.grpc.GetStatusPOptions;
import alluxio.grpc.ListStatusPOptions;
import alluxio.grpc.LoadMetadataPType;
import alluxio.grpc.RenamePOptions;
import alluxio.grpc.SetAttributePOptions;
import alluxio.grpc.WritePType;
import alluxio.master.MasterClientContext;
import alluxio.security.authorization.Mode;
import alluxio.testutils.BaseIntegrationTest;
import alluxio.testutils.LocalAlluxioClusterResource;
import alluxio.underfs.UnderFileSystem;
import alluxio.util.CommonUtils;
import alluxio.util.FileSystemOptionsUtils;
import alluxio.util.WaitForOptions;
import alluxio.util.io.FileUtils;
import alluxio.util.io.PathUtils;

import com.google.common.collect.Sets;
import com.google.common.io.Files;
import io.grpc.Context;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;

import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.nio.file.attribute.PosixFilePermissions;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * Tests the loading of metadata and the available options.
 */
@DoraTestTodoItem(action = DoraTestTodoItem.Action.FIX, owner = "jiacheng",
    comment = "check if this feature is still relevant")
@Ignore
public class UfsSyncIntegrationTest extends BaseIntegrationTest {
  private static final long INTERVAL_MS = 100;
  private static final long LARGE_INTERVAL_MS = 1000;
  private static final FileSystemMasterCommonPOptions PSYNC_NEVER =
      FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(-1).build();
  private static final FileSystemMasterCommonPOptions PSYNC_ALWAYS =
      FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build();
  private static final FileSystemMasterCommonPOptions PSYNC_INTERVAL =
      FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(INTERVAL_MS).build();
  private static final FileSystemMasterCommonPOptions PSYNC_LARGE_INTERVAL =
      FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(LARGE_INTERVAL_MS).build();

  private static final String ROOT_DIR = "/";
  private static final String EXISTING_DIR = "/dir_exist";
  private static final String NEW_FILE_UNDER_DIR = "/dir_exist/file_new";
  private static final String EXISTING_FILE = "/file_exist";
  private static final String NEW_DIR = "/dir_new";
  private static final String NEW_FILE = "/file_new";
  private static final String NEW_NESTED_FILE = "/a/b/file_new";

  private FileSystem mFileSystem;
  private String mLocalUfsPath = Files.createTempDir().getAbsolutePath();

  @Rule
  public AuthenticatedUserRule mAuthenticatedUser = new AuthenticatedUserRule("test",
      Configuration.global());

  @Rule
  public LocalAlluxioClusterResource mLocalAlluxioClusterResource =
      new LocalAlluxioClusterResource.Builder()
          // use a smaller block size so files can have multiple blocks
          .setProperty(PropertyKey.USER_BLOCK_SIZE_BYTES_DEFAULT, 5)
          .setProperty(PropertyKey.MASTER_METADATA_SYNC_CONCURRENCY_LEVEL, 10)
          .setProperty(PropertyKey.MASTER_METADATA_SYNC_EXECUTOR_POOL_SIZE, 10).build();

  @After
  public void after() throws Exception {
    Configuration.reloadProperties();
  }

  @Before
  public void before() throws Exception {
    mFileSystem = FileSystem.Factory.create();
    mFileSystem.mount(new AlluxioURI("/mnt/"), new AlluxioURI(mLocalUfsPath));

    new File(ufsPath(EXISTING_DIR)).mkdirs();
    writeUfsFile(ufsPath(EXISTING_FILE), 1);
  }

  @Test
  public void getStatusNoSync() throws Exception {
    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_NEVER).build();
    checkGetStatus(EXISTING_DIR, options, false);
    checkGetStatus(EXISTING_FILE, options, false);

    // Create new ufs paths.
    new File(ufsPath(NEW_DIR)).mkdirs();
    writeUfsFile(ufsPath(NEW_FILE), 2);

    checkGetStatus(NEW_DIR, options, false);
    checkGetStatus(NEW_FILE, options, false);
  }

  @Test
  public void listStatusNoSync() throws Exception {
    ListStatusPOptions options = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_NEVER).build();
    checkListStatus(EXISTING_DIR, options, false);
    checkListStatus(EXISTING_FILE, options, false);

    // Create new ufs paths.
    new File(ufsPath(NEW_DIR)).mkdirs();
    writeUfsFile(ufsPath(NEW_FILE), 2);

    checkListStatus(NEW_DIR, options, false);
    checkListStatus(NEW_FILE, options, false);
  }

  @Test
  public void getStatusFileSync() throws Exception {
    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    checkGetStatus(EXISTING_FILE, options, true);
  }

  @Test
  public void getStatusDirSync() throws Exception {
    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    checkGetStatus(EXISTING_DIR, options, true);
  }

  // https://github.com/Alluxio/alluxio/issues/12372
  @Test
  public void getStatusDirSyncOnlyTouchingChildren() throws Exception {
    String dir1 = PathUtils.concatPath(EXISTING_DIR, "dir_should_sync");
    String dir2 = PathUtils.concatPath(dir1, "dir_should_not_sync");
    new File(ufsPath(dir1)).mkdirs();
    new File(ufsPath(dir2)).mkdirs();
    GetStatusPOptions optionsAlways = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER)
        .setCommonOptions(PSYNC_ALWAYS).build();
    checkGetStatus(EXISTING_DIR, optionsAlways, true);
    ListStatusPOptions optionsNever = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_NEVER)
        .setRecursive(false).build();
    checkListStatus(dir2, optionsNever, false);
  }

  @Test
  public void listDirSync() throws Exception {
    ListStatusPOptions options = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    checkListStatus(ROOT_DIR, options, true);

    // Create new ufs paths.
    new File(ufsPath(NEW_DIR)).mkdirs();
    writeUfsFile(ufsPath(NEW_FILE), 2);

    checkListStatus(ROOT_DIR, options, true);
  }

  // https://github.com/Alluxio/alluxio/issues/12372
  @Test
  public void listDirSyncOnlyTouchingChildren() throws Exception {
    String dir1 = PathUtils.concatPath(EXISTING_DIR, "dir_should_sync");
    String dir2 = PathUtils.concatPath(dir1, "dir_should_not_sync");
    new File(ufsPath(dir1)).mkdirs();
    new File(ufsPath(dir2)).mkdirs();
    ListStatusPOptions optionsAlways = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setRecursive(false)
        .setCommonOptions(PSYNC_ALWAYS).build();
    checkListStatus(EXISTING_DIR, optionsAlways, true);
    ListStatusPOptions optionsNever = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_NEVER)
        .setRecursive(false).build();
    checkListStatus(dir2, optionsNever, false);
  }

  @Test
  public void getStatusFileSyncInterval() throws Exception {
    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_INTERVAL).build();
    long startMs = System.currentTimeMillis();
    URIStatus status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
    long startLength = status.getLength();
    int sizeFactor = 10;
    while (true) {
      writeUfsFile(ufsPath(EXISTING_FILE), sizeFactor);
      status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
      if (status.getLength() != startLength) {
        break;
      }
      sizeFactor++;
    }
    long endMs = System.currentTimeMillis();

    assertTrue((endMs - startMs) >= INTERVAL_MS);
  }

  @Test(timeout = 10000)
  public void listDirSyncInterval() throws Exception {
    ListStatusPOptions options = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_INTERVAL).build();
    long startMs = System.currentTimeMillis();
    List<URIStatus> statusList =
        mFileSystem.listStatus(new AlluxioURI(alluxioPath(ROOT_DIR)), options);

    long startSize = statusList.size();
    int index = 10;
    while (true) {
      writeUfsFile(ufsPath(NEW_FILE + index), 1);
      statusList = mFileSystem.listStatus(new AlluxioURI(alluxioPath(ROOT_DIR)), options);
      if (statusList.size() != startSize) {
        break;
      }
      index++;
    }
    long endMs = System.currentTimeMillis();

    assertTrue((endMs - startMs) >= INTERVAL_MS);
  }

  @Test
  public void deleteFileNoSync() throws Exception {
    DeletePOptions options = DeletePOptions.newBuilder().setCommonOptions(PSYNC_NEVER).build();
    try {
      mFileSystem.delete(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
      Assert.fail("Delete expected to fail: " + alluxioPath(EXISTING_FILE));
    } catch (FileDoesNotExistException e) {
      // expected
    }
  }

  @Test
  public void deleteFileSync() throws Exception {
    DeletePOptions options = DeletePOptions.newBuilder().setCommonOptions(PSYNC_ALWAYS).build();
    mFileSystem.delete(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
  }

  @Test
  public void renameFileNoSync() throws Exception {
    RenamePOptions options = RenamePOptions.newBuilder().setCommonOptions(PSYNC_NEVER).build();
    try {
      mFileSystem
          .rename(new AlluxioURI(alluxioPath(EXISTING_FILE)), new AlluxioURI(alluxioPath(NEW_FILE)),
              options);
      Assert.fail("Rename expected to fail.");
    } catch (FileDoesNotExistException e) {
      // expected
    }
  }

  @Test
  public void renameFileSync() throws Exception {
    RenamePOptions options = RenamePOptions.newBuilder().setCommonOptions(PSYNC_ALWAYS).build();
    mFileSystem
        .rename(new AlluxioURI(alluxioPath(EXISTING_FILE)), new AlluxioURI(alluxioPath(NEW_FILE)),
            options);
  }

  @Test(timeout = 10000)
  public void lastModifiedLocalFileSync() throws Exception {
    int sizefactor = 10;
    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    writeUfsFile(ufsPath(EXISTING_FILE), sizefactor);
    URIStatus status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
    String startFingerprint = status.getUfsFingerprint();
    long startLength = status.getLength();

    long modTime = new File(ufsPath(EXISTING_FILE)).lastModified();
    long newModTime = modTime;
    while (newModTime == modTime) {
      // Sleep for a bit for a different last mod time.
      // This could be flaky, so
      long delay = 100;
      CommonUtils.sleepMs(delay);

      // Write the same file, but with a different last mod time.
      writeUfsFile(ufsPath(EXISTING_FILE), sizefactor);
      newModTime = new File(ufsPath(EXISTING_FILE)).lastModified();
    }

    status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);

    // Verify the sizes are the same, but the fingerprints are different.
    // Will only work with local ufs, since local ufs uses the last mod time for the content hash.
    assertTrue(status.getLength() == startLength);
    Assert.assertNotEquals(startFingerprint, status.getUfsFingerprint());
  }

  @Test
  public void mountPoint() throws Exception {
    ListStatusPOptions options = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_NEVER).build();
    List<String> rootListing =
        mFileSystem.listStatus(new AlluxioURI("/"), options).stream().map(URIStatus::getName)
            .collect(Collectors.toList());
    assertEquals(1, rootListing.size());
    assertEquals("mnt", rootListing.get(0));

    options = ListStatusPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.NEVER)
        .setCommonOptions(PSYNC_ALWAYS).build();
    rootListing =
        mFileSystem.listStatus(new AlluxioURI("/"), options).stream().map(URIStatus::getName)
            .collect(Collectors.toList());
    assertEquals(1, rootListing.size());
    assertEquals("mnt", rootListing.get(0));
  }

  @Test
  public void mountPointConflict() throws Exception {
    GetStatusPOptions gsOptions = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    URIStatus status = mFileSystem.getStatus(new AlluxioURI("/"), gsOptions);

    // add a UFS dir which conflicts with a mount point.
    String fromRootUfs = status.getUfsPath() + "/mnt";
    assertTrue(new File(fromRootUfs).mkdirs());

    ListStatusPOptions options = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    List<URIStatus> rootListing = mFileSystem.listStatus(new AlluxioURI("/"), options);
    assertEquals(1, rootListing.size());
    assertEquals("mnt", rootListing.get(0).getName());
    Assert.assertNotEquals(fromRootUfs, rootListing.get(0).getUfsPath());
  }

  @Test
  public void mountPointNested() throws Exception {
    String ufsPath = Files.createTempDir().getAbsolutePath();
    mFileSystem.createDirectory(new AlluxioURI("/nested/mnt/"), CreateDirectoryPOptions.newBuilder()
        .setRecursive(true).setWriteType(WritePType.CACHE_THROUGH).build());
    mFileSystem.mount(new AlluxioURI("/nested/mnt/ufs"), new AlluxioURI(ufsPath));

    // recursively sync (setAttribute enables recursive sync)
    mFileSystem.setAttribute(new AlluxioURI("/"), SetAttributePOptions.newBuilder()
        .setCommonOptions(PSYNC_ALWAYS.toBuilder().setTtl(55555)).setRecursive(true)
        .build());

    // Verify /nested/mnt/ dir has 1 mount point
    ListStatusPOptions options = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_NEVER).build();
    List<URIStatus> listing = mFileSystem.listStatus(new AlluxioURI("/nested/mnt/"), options);
    assertEquals(1, listing.size());
    assertEquals("ufs", listing.get(0).getName());

    // Remove a directory in the parent UFS, which has a mount point descendant
    URIStatus status =
        mFileSystem.getStatus(new AlluxioURI("/nested/mnt/"), GetStatusPOptions.newBuilder()
            .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_NEVER).build());
    assertTrue(new File(status.getUfsPath()).delete());

    // recursively sync (setAttribute enables recursive sync)
    mFileSystem.setAttribute(new AlluxioURI("/"), SetAttributePOptions.newBuilder()
        .setCommonOptions(PSYNC_ALWAYS.toBuilder().setTtl(44444)).setRecursive(true).build());

    // Verify /nested/mnt/ dir has 1 mount point
    listing = mFileSystem.listStatus(new AlluxioURI("/nested/mnt/"), options);
    assertEquals(1, listing.size());
    assertEquals("ufs", listing.get(0).getName());

    // Remove a directory in the parent UFS, which has a mount point descendant
    status = mFileSystem.getStatus(new AlluxioURI("/nested/"), GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_NEVER).build());
    assertTrue(new File(status.getUfsPath()).delete());

    // recursively sync (setAttribute enables recursive sync)
    mFileSystem.setAttribute(new AlluxioURI("/"), SetAttributePOptions.newBuilder()
        .setCommonOptions(PSYNC_ALWAYS.toBuilder().setTtl(44444)).setRecursive(true).build());

    // Verify /nested/mnt/ dir has 1 mount point
    listing = mFileSystem.listStatus(new AlluxioURI("/nested/mnt/"), options);
    assertEquals(1, listing.size());
    assertEquals("ufs", listing.get(0).getName());

    // adding a file into the nested mount point
    writeUfsFile(ufsPath + "/nestedufs", 1);

    // recursively sync (setAttribute enables recursive sync)
    mFileSystem.setAttribute(new AlluxioURI("/"), SetAttributePOptions.newBuilder()
        .setCommonOptions(PSYNC_ALWAYS.toBuilder().setTtl(44444)).setRecursive(true).build());
    // Verify /nested/mnt/ufs dir has 1 file
    listing = mFileSystem.listStatus(new AlluxioURI("/nested/mnt/ufs"), options);
    assertEquals(1, listing.size());
    assertEquals("nestedufs", listing.get(0).getName());
  }

  @Test
  public void ufsModeSync() throws Exception {
    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    writeUfsFile(ufsPath(EXISTING_FILE), 10);
    // Set the ufs permissions
    File ufsFile = new File(ufsPath(EXISTING_FILE));
    assertTrue(ufsFile.setReadable(true, false));
    assertTrue(ufsFile.setWritable(true, false));
    assertTrue(ufsFile.setExecutable(true, false));

    URIStatus status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
    String startFingerprint = status.getUfsFingerprint();

    // Update ufs permissions
    assertTrue(ufsFile.setExecutable(false, false));

    status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);

    // Verify the fingerprints are different.
    Assert.assertNotEquals(startFingerprint, status.getUfsFingerprint());
  }

  @Test
  public void alluxioModeFingerprintUpdate() throws Exception {
    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.ONCE).setCommonOptions(PSYNC_NEVER).build();
    writeUfsFile(ufsPath(EXISTING_FILE), 10);
    Assert
        .assertNotNull(mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options));

    // Set initial alluxio permissions
    mFileSystem.setAttribute(new AlluxioURI(alluxioPath(EXISTING_FILE)),
        SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto()).build());

    URIStatus status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
    String startFingerprint = status.getUfsFingerprint();

    // Change alluxio permissions
    mFileSystem.setAttribute(new AlluxioURI(alluxioPath(EXISTING_FILE)),
        SetAttributePOptions.newBuilder().setMode(new Mode((short) 0655).toProto()).build());

    status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
    String endFingerprint = status.getUfsFingerprint();

    // Verify the fingerprints are different.
    Assert.assertNotEquals(startFingerprint, endFingerprint);
  }

  @Test
  public void ufsDirUpdatePermissions() throws Exception {
    new File(ufsPath("/dir1")).mkdirs();
    new File(ufsPath("/dir1/dir2")).mkdirs();
    String fileA = "/dir1/dir2/fileA";
    writeUfsFile(ufsPath(fileA), 1);

    // Set the mode for the directory
    FileUtils.changeLocalFilePermission(ufsPath("/dir1"), "rwxrwxrwx");

    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    URIStatus status = mFileSystem.getStatus(new AlluxioURI(alluxioPath("/dir1")), options);
    Assert.assertNotNull(status);

    assertEquals(
        FileUtils.translatePosixPermissionToMode(PosixFilePermissions.fromString("rwxrwxrwx")),
        status.getMode());

    // Change the mode for the directory
    FileUtils.changeLocalFilePermission(ufsPath("/dir1"), "rwxr-xr-x");

    status = mFileSystem.getStatus(new AlluxioURI(alluxioPath("/dir1")), options);
    Assert.assertNotNull(status);

    assertEquals(
        FileUtils.translatePosixPermissionToMode(PosixFilePermissions.fromString("rwxr-xr-x")),
        status.getMode());
  }

  @Test
  public void ufsMetadataContentChange() throws Exception {
    FileSystemTestUtils.loadFile(mFileSystem, alluxioPath(EXISTING_FILE));

    GetStatusPOptions options = GetStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(PSYNC_ALWAYS).build();
    URIStatus status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
    Assert.assertNotNull(status);
    long prevFileid = status.getFileId();

    // Set the mode for the file
    FileUtils.changeLocalFilePermission(ufsPath(EXISTING_FILE), "rwxrwxrwx");

    status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
    Assert.assertNotNull(status);

    // Make sure the mode is correctly updated with a metadata change only
    assertEquals(
        FileUtils.translatePosixPermissionToMode(PosixFilePermissions.fromString("rwxrwxrwx")),
        status.getMode());

    // Change the permission of the file and the file id should not change
    assertEquals(prevFileid, status.getFileId());

    // Change the content of the file and the file id should change as a result because it is
    // deleted and reloaded.
    writeUfsFile(ufsPath(EXISTING_FILE), 2);

    status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE)), options);
    Assert.assertNotNull(status);
    Assert.assertNotEquals(prevFileid, status.getFileId());
  }

  @Test
  public void ufsDeleteSync() throws Exception {
    FileSystemTestUtils.loadFile(mFileSystem, alluxioPath(EXISTING_FILE));
    FileSystemUtils
        .waitForAlluxioPercentage(mFileSystem, new AlluxioURI(alluxioPath(EXISTING_FILE)), 100);
    new File(ufsPath(EXISTING_FILE)).delete();
    assertFalse(mFileSystem.exists(new AlluxioURI(alluxioPath(EXISTING_FILE)),
        ExistsPOptions.newBuilder().setCommonOptions(PSYNC_ALWAYS).build()));
    mFileSystem.free(new AlluxioURI("/"), FreePOptions.newBuilder().setRecursive(true).build());
    BlockMasterClient blockClient =
        BlockMasterClient.Factory.create(MasterClientContext
            .newBuilder(ClientContext.create(Configuration.global())).build());
    CommonUtils.waitFor("data to be freed", () -> {
      try {
        return blockClient.getUsedBytes() == 0;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, WaitForOptions.defaults().setTimeoutMs(30 * Constants.SECOND_MS));
  }

  @Test
  public void ufsDeleteChildrenSync() throws Exception {
    // Create a UFS directory with several files
    // Force each file to have many blocks
    String baseDir = "/base_dir";
    int numChildren = 20;
    new File(ufsPath(baseDir)).mkdirs();
    for (int i = 0; i < numChildren; i++) {
      writeUfsFile(ufsPath(baseDir + "/child" + i), 10000);
    }

    ListStatusPOptions lsOptions =
        ListStatusPOptions.newBuilder().setCommonOptions(PSYNC_ALWAYS).build();

    // initial sync
    List<URIStatus> statuses =
        mFileSystem.listStatus(new AlluxioURI(alluxioPath(baseDir)), lsOptions);
    assertEquals(numChildren, statuses.size());

    // Delete all the children from UFS only
    for (int i = 0; i < numChildren; i++) {
      new File(ufsPath(baseDir + "/child" + i)).delete();
    }

    // this is a parallel sync, check to see if BlockDeletionContext is thread-safe.
    statuses =
        mFileSystem.listStatus(new AlluxioURI(alluxioPath(baseDir)), lsOptions);
    assertTrue(statuses.isEmpty());
  }

  @Test
  public void createNestedFileSync() throws Exception {
    Configuration.set(PropertyKey.USER_FILE_METADATA_SYNC_INTERVAL, "0");

    mFileSystem.createFile(new AlluxioURI(alluxioPath(NEW_NESTED_FILE)),
        CreateFilePOptions.newBuilder().setWriteType(WritePType.CACHE_THROUGH)
            .setCommonOptions(PSYNC_ALWAYS).setRecursive(true).build())
        .close();

    // Make sure we can create the nested file.
    Assert.assertNotNull(mFileSystem.getStatus(new AlluxioURI(alluxioPath(EXISTING_FILE))));
  }

  @Test
  public void clusterRestartSync() throws Exception {
    ListStatusPOptions listStatusPOptions = ListStatusPOptions.newBuilder()
        .setLoadMetadataType(LoadMetadataPType.NEVER)
        .setCommonOptions(PSYNC_LARGE_INTERVAL).build();

    List<URIStatus> statusList =
        mFileSystem.listStatus(new AlluxioURI(alluxioPath(EXISTING_DIR)), listStatusPOptions);
    Assert.assertNotNull(statusList);
    assertEquals(0, statusList.size());
    mLocalAlluxioClusterResource.get().stopMasters();
    mLocalAlluxioClusterResource.get().startMasters();

    List<URIStatus> statusListAfterRestart =
        mFileSystem.listStatus(new AlluxioURI(alluxioPath(EXISTING_DIR)), listStatusPOptions);
    Assert.assertNotNull(statusListAfterRestart);
    assertEquals(0, statusListAfterRestart.size());

    writeUfsFile(ufsPath(NEW_FILE_UNDER_DIR), 1);

    List<URIStatus> statusListAgain =
        mFileSystem.listStatus(new AlluxioURI(alluxioPath(EXISTING_DIR)), listStatusPOptions);
    Assert.assertNotNull(statusListAgain);
    assertEquals(0, statusListAgain.size());

    Thread.sleep(LARGE_INTERVAL_MS);

    List<URIStatus> statusListAfterSleeping =
        mFileSystem.listStatus(new AlluxioURI(alluxioPath(EXISTING_DIR)), listStatusPOptions);
    Assert.assertNotNull(statusListAfterSleeping);
    assertEquals(1, statusListAfterSleeping.size());
  }

  /** This is a timing based test and may become flaky.
   *  The goal is to simulate a user interrupted listStatus call.
   *
   *  In this case, the user's listStatus should have synced the first level directory but have
   *  not completed the second level directory sync. Thus resulting in a partial sync.
   */
  @LocalAlluxioClusterResource.Config(
      confParams = {
          PropertyKey.Name.USER_FILE_METADATA_LOAD_TYPE, "NEVER"
      })
  @Test
  public void interruptSync() throws Exception {
    // make large nested directories/files in UFS
    for (int i = 0; i < 100; i++) {
      new File(ufsPath("/dir" + i)).mkdirs();
      for (int j = 0; j < 100; j++) {
        new File(ufsPath("/dir" + i + "/dir" + j)).mkdirs();
        writeUfsFile(ufsPath("/dir" + i + "/dir" + j + "/file"), 1);
      }
    }
    List<URIStatus> status;
    try (Context.CancellableContext c = Context.current()
        .withDeadlineAfter(1, TimeUnit.MILLISECONDS,
            Executors.newScheduledThreadPool(1))) {
      Context toRestore = c.attach();
      try {
        status = Context.current().withCancellation().call(() -> {
          try {
            return mFileSystem.listStatus(new AlluxioURI("/"), ListStatusPOptions.newBuilder()
                .setRecursive(true)
                .setCommonOptions(FileSystemOptionsUtils.commonDefaults(
                    mFileSystem.getConf()).toBuilder().setSyncIntervalMs(0).build()).build());
          } catch (Exception e) {
            return Collections.<URIStatus>emptyList();
          }
        });
        Thread.sleep(5);
        c.cancel(new AlluxioException("test exception"));
      } finally {
        c.detach(toRestore);
      }
    }
    // cancelled call should not return any results
    assertEquals(0, status.size());
    status = mFileSystem.listStatus(new AlluxioURI("/"), ListStatusPOptions.newBuilder()
        .setRecursive(true)
        .setCommonOptions(FileSystemOptionsUtils.commonDefaults(
            mFileSystem.getConf()).toBuilder().setSyncIntervalMs(-1).build()).build());
    final int TOTAL_FILE_COUNT = 20103;
    // verify that the previous sync did not complete
    assertTrue(status.size() < TOTAL_FILE_COUNT);
    for (URIStatus stat : status) {
      assertTrue(stat.isCompleted());
    }
    status = mFileSystem.listStatus(new AlluxioURI("/"), ListStatusPOptions.newBuilder()
        .setRecursive(true).setCommonOptions(PSYNC_ALWAYS).build());
    assertEquals(TOTAL_FILE_COUNT, status.size());
  }

  @LocalAlluxioClusterResource.Config(
      confParams = {
          PropertyKey.Name.USER_FILE_METADATA_LOAD_TYPE, "NEVER"
      })
  @Test
  public void recursiveSync() throws Exception {

    // make nested directories/files in UFS
    new File(ufsPath("/dir1")).mkdirs();
    new File(ufsPath("/dir1/dir2")).mkdirs();
    new File(ufsPath("/dir1/dir2/dir3")).mkdirs();
    String fileA = "/dir1/dir2/fileA";
    String fileB = "/dir1/dir2/fileB";
    String fileC = "/dir1/dir2/fileC";
    writeUfsFile(ufsPath(fileA), 1);
    writeUfsFile(ufsPath(fileB), 1);

    // Should not exist, since no loading or syncing
    assertFalse(mFileSystem.exists(new AlluxioURI(alluxioPath(fileA)), ExistsPOptions.newBuilder()
        .setCommonOptions(FileSystemOptionsUtils.commonDefaults(mFileSystem.getConf()).toBuilder()
            .setSyncIntervalMs(-1).build()).build()));

    try {
      mFileSystem.setAttribute(new AlluxioURI(alluxioPath("/dir1")),
          SetAttributePOptions.newBuilder().setRecursive(true).setCommonOptions(
              FileSystemMasterCommonPOptions.newBuilder().setTtl(55555).setSyncIntervalMs(-1))
              .build());
    } catch (FileDoesNotExistException e) {
      // expected, continue
    }

    // Enable UFS sync, before next recursive setAttribute.
    FileSystemMasterCommonPOptions ttlOption = FileSystemMasterCommonPOptions.newBuilder()
        .setTtl(123456789).setSyncIntervalMs(0).build();
    mFileSystem.setAttribute(new AlluxioURI(alluxioPath("/dir1")),
        SetAttributePOptions.newBuilder().setRecursive(true).setCommonOptions(ttlOption).build());

    // Verify recursive set TTL by getting info, without sync.
    ttlOption = ttlOption.toBuilder().setSyncIntervalMs(-1).build();
    URIStatus status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(fileA)));
    assertEquals(ttlOption.getTtl(), status.getTtl());

    // Add UFS fileC and remove existing UFS fileA.
    writeUfsFile(ufsPath(fileC), 1);
    assertTrue(new File(ufsPath(fileA)).delete());

    // Enable UFS sync, before next recursive setAttribute.
    ttlOption = FileSystemMasterCommonPOptions.newBuilder().setTtl(987654321)
        .setSyncIntervalMs(0).build();
    mFileSystem.setAttribute(new AlluxioURI(alluxioPath("/dir1")),
        SetAttributePOptions.newBuilder().setRecursive(true).setCommonOptions(ttlOption).build());

    // Verify recursive set TTL by getting info, without sync.
    ttlOption =
        FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(-1).setTtl(987654321).build();
    status = mFileSystem.getStatus(new AlluxioURI(alluxioPath(fileB)));
    assertEquals(ttlOption.getTtl(), status.getTtl());

    // deleted UFS file should not exist.
    assertFalse(mFileSystem.exists(new AlluxioURI(alluxioPath(fileA))));
  }

  @LocalAlluxioClusterResource.Config(
      confParams = {
          PropertyKey.Name.USER_FILE_METADATA_LOAD_TYPE, "NEVER"
      })
  @Test
  public void recursiveSyncCacheDescendants() throws Exception {
    // make nested directories/files in UFS
    new File(ufsPath("/dir1")).mkdirs();
    new File(ufsPath("/dir1/dir2")).mkdirs();
    new File(ufsPath("/dir1/dir2/dir3")).mkdirs();
    String fileA = "/dir1/dir2/dir3/fileA";
    String fileB = "/dir1/dir2/dir3/fileB";
    String fileNew = "/dir1/dir2/dir3/fileNew";
    writeUfsFile(ufsPath(fileA), 1);
    writeUfsFile(ufsPath(fileB), 1);

    FileSystemMasterCommonPOptions longinterval =
        FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(Constants.HOUR_MS).build();

    // Should not exist, since no loading or syncing
    assertFalse(mFileSystem.exists(new AlluxioURI(alluxioPath(fileA)), ExistsPOptions.newBuilder()
        .setCommonOptions(PSYNC_NEVER).build()));

    try {
      mFileSystem.listStatus(new AlluxioURI(alluxioPath("/dir1")),
          ListStatusPOptions.newBuilder().setCommonOptions(PSYNC_NEVER).build());
      Assert.fail("paths are not expected to exist without sync");
    } catch (FileDoesNotExistException e) {
      // expected, continue
    }

    // recursively sync the top dir
    List<URIStatus> paths = mFileSystem.listStatus(new AlluxioURI(alluxioPath("/dir1")),
        ListStatusPOptions.newBuilder().setCommonOptions(PSYNC_ALWAYS).setRecursive(true)
            .build());
    assertEquals(4, paths.size());

    // write a new UFS file
    writeUfsFile(ufsPath(fileNew), 1);
    // the new UFS file should not exist, since the sync interval is 1 hour, and an ancestor
    // already synced recently.
    assertFalse(mFileSystem.exists(new AlluxioURI(alluxioPath(fileNew)), ExistsPOptions.newBuilder()
        .setCommonOptions(longinterval).build()));

    // newly created file should not exist
    paths = mFileSystem.listStatus(new AlluxioURI(alluxioPath("/dir1/dir2/dir3")),
        ListStatusPOptions.newBuilder().setCommonOptions(longinterval).build());
    assertEquals(2, paths.size());

    // create a new UFS dir
    new File(ufsPath("/dir1/dir2/dirNew")).mkdirs();
    // newly created dir should not exist, since sync interval is long, and an ancestor is
    // already synced
    assertFalse(mFileSystem.exists(new AlluxioURI(alluxioPath("/dir1/dir2/dirNew")),
        ExistsPOptions.newBuilder().setCommonOptions(longinterval).build()));
    // newly created dir should not exist
    paths = mFileSystem.listStatus(new AlluxioURI(alluxioPath("/dir1/dir2")),
        ListStatusPOptions.newBuilder().setCommonOptions(longinterval).build());
    assertEquals(1, paths.size());

    // check the original path, and verify no new files/dirs are picked up from UFS
    paths = mFileSystem.listStatus(new AlluxioURI(alluxioPath("/dir1")),
        ListStatusPOptions.newBuilder().setCommonOptions(longinterval).setRecursive(true)
            .build());
    assertEquals(4, paths.size());
  }

  @Test
  public void deleteUfsFileGetStatus() throws Exception {
    new File(ufsPath("/delete")).mkdirs();
    writeUfsFile(ufsPath("/delete/file"), 10);

    List<URIStatus> paths = mFileSystem.listStatus(new AlluxioURI(alluxioPath("/delete")),
        ListStatusPOptions.newBuilder().setRecursive(false).setCommonOptions(PSYNC_ALWAYS).build());

    assertEquals(1, paths.size());
    assertEquals("file", paths.get(0).getName());

    // delete the file and wait a bit
    new File(ufsPath("/delete/file")).delete();
    CommonUtils.sleepMs(3000);

    // getStatus (not listStatus) on the root, with a shorter interval than the sleep.
    // This will sync that directory. The sync interval has to be long enough for the internal
    // syncing process to finish within that time.
    mFileSystem.getStatus(new AlluxioURI(alluxioPath("/delete")), GetStatusPOptions.newBuilder()
        .setCommonOptions(
            FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(2000).build()).build());

    // a following list status should trigger a metadata sync even though the path was just synced,
    // because the descendant type is ONE this time, and it was NONE previously.
    mFileSystem.listStatus(new AlluxioURI(alluxioPath("/delete")),
        ListStatusPOptions.newBuilder().setRecursive(false).setCommonOptions(
            FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(2000).build()).build());

    // verify that the file is deleted, without syncing
    try {
      mFileSystem.getStatus(new AlluxioURI(alluxioPath("/delete/file")),
          GetStatusPOptions.newBuilder().setCommonOptions(PSYNC_NEVER).build());
      Assert.fail("the ufs deleted file is not expected to exist after sync via getStatus");
    } catch (FileDoesNotExistException e) {
      // expected
    }
  }

  private String ufsPath(String path) {
    return mLocalUfsPath + path;
  }

  private String alluxioPath(String path) {
    return "/mnt/" + path;
  }

  private void writeUfsFile(String path, int sizeFactor) throws IOException {
    FileWriter fileWriter = new FileWriter(path);
    for (int i = 0; i < sizeFactor; i++) {
      fileWriter.write("test");
    }
    fileWriter.close();
  }

  private void checkGetStatus(String path, GetStatusPOptions options, boolean expectExists)
      throws Exception {
    try {
      URIStatus uriStatus = mFileSystem.getStatus(new AlluxioURI(alluxioPath(path)), options);
      if (!expectExists) {
        Assert.fail("Path is not expected to exist: " + alluxioPath(path));
      }
      checkUriStatus(uriStatus);
    } catch (FileDoesNotExistException e) {
      if (expectExists) {
        throw e;
      }
    }
  }

  private void checkListStatus(String path, ListStatusPOptions options, boolean expectExists)
      throws Exception {
    try {
      List<URIStatus> statusList =
          mFileSystem.listStatus(new AlluxioURI(alluxioPath(path)), options);
      if (!expectExists) {
        Assert.fail("Path is not expected to exist: " + alluxioPath(path));
      }
      Assert.assertNotNull(statusList);
      for (URIStatus uriStatus : statusList) {
        checkUriStatus(uriStatus);
      }
      checkUfsListing(ufsPath(path), statusList);
    } catch (FileDoesNotExistException e) {
      if (expectExists) {
        throw e;
      }
    }
  }

  private void checkUriStatus(URIStatus uriStatus) {
    File ufsFile = new File(uriStatus.getUfsPath());
    if (uriStatus.isFolder() != ufsFile.isDirectory()) {
      Assert.fail("Directory mismatch (Alluxio isDir: " + uriStatus.isFolder() + ") (ufs isDir: "
          + ufsFile.isDirectory() + ") path: " + uriStatus.getPath());
    }
    if (!uriStatus.isFolder()) {
      long ufsLength = ufsFile.length();
      long alluxioLength = uriStatus.getLength();
      if (ufsLength != alluxioLength) {
        Assert.fail("Alluxio length (" + alluxioLength + ") and ufs length (" + ufsLength
            + ") are inconsistent. path: " + uriStatus.getPath());
      }
      // Check fingerprint.
      UnderFileSystem ufs = UnderFileSystem.Factory.create(uriStatus.getUfsPath(),
          Configuration.global());
      String ufsFingerprint = ufs.getParsedFingerprint(uriStatus.getUfsPath()).serialize();
      String alluxioFingerprint = uriStatus.getUfsFingerprint();
      if (!ufsFingerprint.equals(alluxioFingerprint)) {
        Assert.fail(
            "Alluxio fingerprint (" + alluxioFingerprint + ") and ufs fingerprint ("
                + ufsFingerprint + ") are inconsistent. path: " + uriStatus.getPath());
      }
    }
  }

  private void checkUfsListing(String ufsPath, List<URIStatus> statusList) {
    File ufsFile = new File(ufsPath);

    Set<String> ufsFiles = Sets.newHashSet(ufsFile.list());

    Set<String> alluxioFiles = new HashSet<>();
    for (URIStatus uriStatus : statusList) {
      alluxioFiles.add(uriStatus.getName());
    }

    ufsFiles.removeAll(alluxioFiles);
    if (!ufsFiles.isEmpty()) {
      Assert.fail("UFS files are missing from Alluxio: " + ufsFiles);
    }
  }
}
