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

import static org.hamcrest.CoreMatchers.containsString;
import static org.hamcrest.CoreMatchers.hasItem;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;

import alluxio.AlluxioURI;
import alluxio.AuthenticatedClientUserResource;
import alluxio.AuthenticatedUserRule;
import alluxio.Constants;
import alluxio.client.WriteType;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AccessControlException;
import alluxio.exception.ExceptionMessage;
import alluxio.exception.FileDoesNotExistException;
import alluxio.exception.InvalidPathException;
import alluxio.exception.UnexpectedAlluxioException;
import alluxio.grpc.Command;
import alluxio.grpc.CommandType;
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.ListStatusPOptions;
import alluxio.grpc.LoadMetadataPType;
import alluxio.grpc.MountPOptions;
import alluxio.grpc.SetAclAction;
import alluxio.grpc.SetAclPOptions;
import alluxio.grpc.SetAttributePOptions;
import alluxio.grpc.StorageList;
import alluxio.grpc.XAttrPropagationStrategy;
import alluxio.heartbeat.HeartbeatContext;
import alluxio.heartbeat.HeartbeatScheduler;
import alluxio.master.file.contexts.CreateDirectoryContext;
import alluxio.master.file.contexts.CreateFileContext;
import alluxio.master.file.contexts.DeleteContext;
import alluxio.master.file.contexts.ExistsContext;
import alluxio.master.file.contexts.FreeContext;
import alluxio.master.file.contexts.GetStatusContext;
import alluxio.master.file.contexts.ListStatusContext;
import alluxio.master.file.contexts.MountContext;
import alluxio.master.file.contexts.RenameContext;
import alluxio.master.file.contexts.ScheduleAsyncPersistenceContext;
import alluxio.master.file.contexts.SetAclContext;
import alluxio.master.file.contexts.SetAttributeContext;
import alluxio.master.file.contexts.WorkerHeartbeatContext;
import alluxio.master.file.meta.PersistenceState;
import alluxio.master.journal.JournalContext;
import alluxio.proto.journal.Journal;
import alluxio.security.authorization.AclEntry;
import alluxio.security.authorization.Mode;
import alluxio.util.FileSystemOptionsUtils;
import alluxio.util.IdUtils;
import alluxio.util.io.FileUtils;
import alluxio.wire.FileBlockInfo;
import alluxio.wire.FileInfo;
import alluxio.wire.FileSystemCommand;
import alluxio.wire.UfsInfo;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.common.math.IntMath;
import com.google.protobuf.ByteString;
import org.junit.Assert;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * Unit tests for {@link FileSystemMaster}.
 */
@RunWith(Parameterized.class)
public final class FileSystemMasterTest extends FileSystemMasterTestBase {
  private static final Logger LOG = LoggerFactory.getLogger(FileSystemMasterTest.class);

  @Parameterized.Parameters
  public static Collection<Object[]> data() {
    return Arrays.asList(new Object[][] {
        {new ImmutableMap.Builder<PropertyKey, Object>()
            .put(PropertyKey.MASTER_FILE_SYSTEM_MERGE_INODE_JOURNALS, false)
            .build()},
        {new ImmutableMap.Builder<PropertyKey, Object>()
            .put(PropertyKey.MASTER_FILE_SYSTEM_MERGE_INODE_JOURNALS, true)
            .put(PropertyKey.MASTER_RECURSIVE_OPERATION_JOURNAL_FORCE_FLUSH_MAX_ENTRIES, 0)
            .build()},
    });
  }

  @Parameterized.Parameter
  public ImmutableMap<PropertyKey, Object> mConfigMap;

  @Override
  public void before() throws Exception {
    for (Map.Entry<PropertyKey, Object> entry : mConfigMap.entrySet()) {
      Configuration.set(entry.getKey(), entry.getValue());
    }
    super.before();
  }

  @Test
  public void createPathWithWhiteSpaces() throws Exception {
    String[] paths = new String[]{
        "/ ",
        "/  ",
        "/ path",
        "/path ",
        "/pa th",
        "/ pa th ",
        "/pa/ th",
        "/pa / th",
        "/ pa / th ",
    };
    for (String path : paths) {
      AlluxioURI uri = new AlluxioURI(path);
      long id = mFileSystemMaster.createFile(uri, CreateFileContext.mergeFrom(
          CreateFilePOptions.newBuilder().setRecursive(true))).getFileId();
      Assert.assertEquals(id, mFileSystemMaster.getFileId(uri));
      mFileSystemMaster.delete(uri, DeleteContext.defaults());
      id = mFileSystemMaster.createDirectory(uri, CreateDirectoryContext
          .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)));
      Assert.assertEquals(id, mFileSystemMaster.getFileId(uri));
    }
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a directory with persistent entries with a sync check.
   */
  @Test
  public void deleteSyncedPersistedDirectoryWithCheck() throws Exception {
    deleteSyncedPersistedDirectory(1, false);
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a directory with persistent entries without a sync check.
   */
  @Test
  public void deleteSyncedPersistedDirectoryWithoutCheck() throws Exception {
    deleteSyncedPersistedDirectory(1, true);
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a multi-level directory with persistent entries with a sync check.
   */
  @Test
  public void deleteSyncedPersistedMultilevelDirectoryWithCheck() throws Exception {
    deleteSyncedPersistedDirectory(3, false);
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a multi-level directory with persistent entries without a sync check.
   */
  @Test
  public void deleteSyncedPersistedMultilevelDirectoryWithoutCheck() throws Exception {
    deleteSyncedPersistedDirectory(3, true);
  }

  // Helper method for deleteSynctedPersisted* tests.
  private void deleteSyncedPersistedDirectory(int levels, boolean unchecked) throws Exception {
    AlluxioURI ufsMount = createPersistedDirectories(levels);
    mountPersistedDirectories(ufsMount);
    loadPersistedDirectories(levels);
    // delete top-level directory
    mFileSystemMaster.delete(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL),
        DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)
            .setAlluxioOnly(false).setUnchecked(unchecked)));
    checkPersistedDirectoriesDeleted(levels, ufsMount, Collections.EMPTY_LIST);
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a directory with un-synced persistent entries with a sync check.
   */
  @Test
  public void deleteUnsyncedPersistedDirectoryWithCheck() throws Exception {
    AlluxioURI ufsMount = createPersistedDirectories(1);
    mountPersistedDirectories(ufsMount);
    loadPersistedDirectories(1);
    // Add a file to the UFS.
    Files.createFile(
        Paths.get(ufsMount.join(DIR_TOP_LEVEL).join(FILE_PREFIX + (DIR_WIDTH)).getPath()));
    // delete top-level directory
    try {
      mFileSystemMaster.delete(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL),
          DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)
              .setAlluxioOnly(false).setUnchecked(false)));
      fail();
    } catch (IOException e) {
      // Expected
    }
    // Check all that could be deleted.
    List<AlluxioURI> except = new ArrayList<>();
    except.add(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL));
    checkPersistedDirectoriesDeleted(1, ufsMount, except);
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a directory with un-synced persistent entries without a sync check.
   */
  @Test
  public void deleteUnsyncedPersistedDirectoryWithoutCheck() throws Exception {
    AlluxioURI ufsMount = createPersistedDirectories(1);
    mountPersistedDirectories(ufsMount);
    loadPersistedDirectories(1);
    // Add a file to the UFS.
    Files.createFile(
        Paths.get(ufsMount.join(DIR_TOP_LEVEL).join(FILE_PREFIX + (DIR_WIDTH)).getPath()));
    // delete top-level directory
    mFileSystemMaster.delete(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL), DeleteContext.mergeFrom(
        DeletePOptions.newBuilder().setRecursive(true).setAlluxioOnly(false).setUnchecked(true)));
    checkPersistedDirectoriesDeleted(1, ufsMount, Collections.EMPTY_LIST);
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a multi-level directory with un-synced persistent entries with a sync check.
   */
  @Test
  public void deleteUnsyncedPersistedMultilevelDirectoryWithCheck() throws Exception {
    AlluxioURI ufsMount = createPersistedDirectories(3);
    mountPersistedDirectories(ufsMount);
    loadPersistedDirectories(3);
    // Add a file to the UFS down the tree.
    Files.createFile(Paths.get(ufsMount.join(DIR_TOP_LEVEL).join(DIR_PREFIX + 0)
        .join(FILE_PREFIX + (DIR_WIDTH)).getPath()));
    mThrown.expect(IOException.class);
    // delete top-level directory
    mFileSystemMaster.delete(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL), DeleteContext.mergeFrom(
        DeletePOptions.newBuilder().setRecursive(true).setAlluxioOnly(false).setUnchecked(false)));
    // Check all that could be deleted.
    List<AlluxioURI> except = new ArrayList<>();
    except.add(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL));
    except.add(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL).join(DIR_PREFIX + 0));
    checkPersistedDirectoriesDeleted(3, ufsMount, except);
  }

  /**
   * Tests the {@link FileSystemMaster#delete(AlluxioURI, DeleteContext)} method for
   * a multi-level directory with un-synced persistent entries without a sync check.
   */
  @Test
  public void deleteUnsyncedPersistedMultilevelDirectoryWithoutCheck() throws Exception {
    AlluxioURI ufsMount = createPersistedDirectories(3);
    mountPersistedDirectories(ufsMount);
    loadPersistedDirectories(3);
    // Add a file to the UFS down the tree.
    Files.createFile(Paths.get(ufsMount.join(DIR_TOP_LEVEL).join(DIR_PREFIX + 0)
        .join(FILE_PREFIX + (DIR_WIDTH)).getPath()));
    // delete top-level directory
    mFileSystemMaster.delete(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL), DeleteContext.mergeFrom(
        DeletePOptions.newBuilder().setRecursive(true).setAlluxioOnly(false).setUnchecked(true)));
    checkPersistedDirectoriesDeleted(3, ufsMount, Collections.EMPTY_LIST);
  }

  /**
   * Tests the {@link FileSystemMaster#getNewBlockIdForFile(AlluxioURI)} method.
   */
  @Test
  public void getNewBlockIdForFile() throws Exception {
    mFileSystemMaster.createFile(NESTED_FILE_URI, mNestedFileContext);
    long blockId = mFileSystemMaster.getNewBlockIdForFile(NESTED_FILE_URI);
    FileInfo fileInfo = mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT);
    assertEquals(Lists.newArrayList(blockId), fileInfo.getBlockIds());
  }

  @Test
  public void getPath() throws Exception {
    AlluxioURI rootUri = new AlluxioURI("/");
    long rootId = mFileSystemMaster.getFileId(rootUri);
    assertEquals(rootUri, mFileSystemMaster.getPath(rootId));

    // get non-existent id
    try {
      mFileSystemMaster.getPath(rootId + 1234);
      fail("getPath() for a non-existent id should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
  }

  /**
   * Tests the {@link FileSystemMaster#getPersistenceState(long)} method.
   */
  @Test
  public void getPersistenceState() throws Exception {
    AlluxioURI rootUri = new AlluxioURI("/");
    long rootId = mFileSystemMaster.getFileId(rootUri);
    assertEquals(PersistenceState.PERSISTED, mFileSystemMaster.getPersistenceState(rootId));

    // get non-existent id
    try {
      mFileSystemMaster.getPersistenceState(rootId + 1234);
      fail("getPath() for a non-existent id should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
  }

  /**
   * Tests the {@link FileSystemMaster#getFileId(AlluxioURI)} method.
   */
  @Test
  public void getFileId() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);

    // These URIs exist.
    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(ROOT_URI));
    assertEquals(ROOT_URI, mFileSystemMaster.getPath(mFileSystemMaster.getFileId(ROOT_URI)));

    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_URI));
    assertEquals(NESTED_URI,
        mFileSystemMaster.getPath(mFileSystemMaster.getFileId(NESTED_URI)));

    assertNotEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(NESTED_FILE_URI));
    assertEquals(NESTED_FILE_URI,
        mFileSystemMaster.getPath(mFileSystemMaster.getFileId(NESTED_FILE_URI)));

    // These URIs do not exist.
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(ROOT_FILE_URI));
    assertEquals(IdUtils.INVALID_FILE_ID, mFileSystemMaster.getFileId(TEST_URI));
    assertEquals(IdUtils.INVALID_FILE_ID,
        mFileSystemMaster.getFileId(NESTED_FILE_URI.join("DNE")));
  }

  @Test
  public void getFileBlockInfoList() throws Exception {
    createFileWithSingleBlock(ROOT_FILE_URI);
    createFileWithSingleBlock(NESTED_FILE_URI);

    List<FileBlockInfo> blockInfo;

    blockInfo = mFileSystemMaster.getFileBlockInfoList(ROOT_FILE_URI);
    assertEquals(1, blockInfo.size());

    blockInfo = mFileSystemMaster.getFileBlockInfoList(NESTED_FILE_URI);
    assertEquals(1, blockInfo.size());

    // Test directory URI.
    try {
      mFileSystemMaster.getFileBlockInfoList(NESTED_URI);
      fail("getFileBlockInfoList() for a directory URI should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }

    // Test non-existent URI.
    try {
      mFileSystemMaster.getFileBlockInfoList(TEST_URI);
      fail("getFileBlockInfoList() for a non-existent URI should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
  }

  @Test
  public void mountUnmount() throws Exception {
    AlluxioURI ufsMount = new AlluxioURI(mTestFolder.newFolder().getAbsolutePath());
    mFileSystemMaster.createDirectory(new AlluxioURI("/mnt/"), CreateDirectoryContext.defaults());

    // Alluxio mount point should not exist before mounting.
    try {
      mFileSystemMaster.getFileInfo(new AlluxioURI("/mnt/local"), GET_STATUS_CONTEXT);
      fail("getFileInfo() for a non-existent URI (before mounting) should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }

    mFileSystemMaster.mount(new AlluxioURI("/mnt/local"), ufsMount,
        MountContext.defaults());
    // Alluxio mount point should exist after mounting.
    assertNotNull(
        mFileSystemMaster.getFileInfo(new AlluxioURI("/mnt/local"), GET_STATUS_CONTEXT));

    mFileSystemMaster.unmount(new AlluxioURI("/mnt/local"));

    // Alluxio mount point should not exist after unmounting.
    try {
      mFileSystemMaster.getFileInfo(new AlluxioURI("/mnt/local"), GET_STATUS_CONTEXT);
      fail("getFileInfo() for a non-existent URI (after mounting) should fail.");
    } catch (FileDoesNotExistException e) {
      // Expected case.
    }
  }

  @Test
  public void inheritExtendedDefaultAcl() throws Exception {
    AlluxioURI dir = new AlluxioURI("/dir");
    mFileSystemMaster.createDirectory(dir, CreateDirectoryContext.defaults());
    String aclString = "default:user:foo:-w-";
    mFileSystemMaster.setAcl(dir, SetAclAction.MODIFY,
        Arrays.asList(AclEntry.fromCliString(aclString)), SetAclContext.defaults());
    AlluxioURI inner = new AlluxioURI("/dir/inner");
    mFileSystemMaster.createDirectory(inner, CreateDirectoryContext.defaults());
    FileInfo fileInfo = mFileSystemMaster.getFileInfo(inner, GetStatusContext.defaults());
    List<String> accessEntries = fileInfo.getAcl().toStringEntries();
    assertTrue(accessEntries.toString(), accessEntries.contains("user:foo:-w-"));
    List<String> defaultEntries = fileInfo.getDefaultAcl().toStringEntries();
    assertTrue(defaultEntries.toString(), defaultEntries.contains(aclString));
  }

  @Test
  public void inheritNonExtendedDefaultAcl() throws Exception {
    AlluxioURI dir = new AlluxioURI("/dir");
    mFileSystemMaster.createDirectory(dir, CreateDirectoryContext.defaults());
    String aclString = "default:user::-w-";
    mFileSystemMaster.setAcl(dir, SetAclAction.MODIFY,
        Arrays.asList(AclEntry.fromCliString(aclString)), SetAclContext.defaults());
    AlluxioURI inner = new AlluxioURI("/dir/inner");
    mFileSystemMaster.createDirectory(inner, CreateDirectoryContext.defaults());
    FileInfo fileInfo = mFileSystemMaster.getFileInfo(inner, GetStatusContext.defaults());
    List<String> accessEntries = fileInfo.getAcl().toStringEntries();
    assertTrue(accessEntries.toString(), accessEntries.contains("user::-w-"));
    List<String> defaultEntries = fileInfo.getDefaultAcl().toStringEntries();
    assertTrue(defaultEntries.toString(), defaultEntries.contains(aclString));
  }

  @Test
  public void setAclWithoutOwner() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setMode(new Mode((short) 0777).toProto())));
    Set<String> entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertEquals(3, entries.size());

    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      Set<String> newEntries = Sets.newHashSet("user::rwx", "group::rwx", "other::rwx");
      mThrown.expect(AccessControlException.class);
      mFileSystemMaster.setAcl(NESTED_FILE_URI, SetAclAction.REPLACE,
          newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()),
          SetAclContext.defaults());
    }
  }

  @Test
  @Ignore
  public void setAclNestedWithoutOwner() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);
    // not supported to set owner with non-existed user in local ufs
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext.mergeFrom(SetAttributePOptions
        .newBuilder().setMode(new Mode((short) 0777).toProto()).setOwner("userA")));
    Set<String> entries = Sets.newHashSet(mFileSystemMaster
        .getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).convertAclToStringEntries());
    assertEquals(3, entries.size());
    // recursive setAcl should fail if one of the child is not owned by the user
    mThrown.expect(AccessControlException.class);
    try (AuthenticatedClientUserResource userA = new AuthenticatedClientUserResource("userA",
        Configuration.global())) {
      Set<String> newEntries = Sets.newHashSet("user::rwx", "group::rwx", "other::rwx");
      mFileSystemMaster.setAcl(NESTED_URI, SetAclAction.REPLACE,
          newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()),
          SetAclContext.mergeFrom(SetAclPOptions.newBuilder().setRecursive(true)));
      entries = Sets.newHashSet(mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT)
          .convertAclToStringEntries());
      assertEquals(newEntries, entries);
    }
  }

  @Test
  public void removeExtendedAclMask() throws Exception {
    mFileSystemMaster.createDirectory(NESTED_URI, CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)));
    AclEntry newAcl = AclEntry.fromCliString("user:newuser:rwx");
    // Add an ACL
    addAcl(NESTED_URI, newAcl);
    assertThat(getInfo(NESTED_URI).getAcl().getEntries(), hasItem(newAcl));

    // Attempt to remove the ACL mask
    AclEntry maskEntry = AclEntry.fromCliString("mask::rwx");
    assertThat(getInfo(NESTED_URI).getAcl().getEntries(), hasItem(maskEntry));
    try {
      removeAcl(NESTED_URI, maskEntry);
      fail("Expected removing the mask from an extended ACL to fail");
    } catch (IOException e) {
      assertThat(e.getMessage(), containsString("mask"));
    }

    // Remove the extended ACL
    removeAcl(NESTED_URI, newAcl);
    // Now we can add and remove a mask
    addAcl(NESTED_URI, maskEntry);
    removeAcl(NESTED_URI, maskEntry);
  }

  @Test
  public void removeExtendedDefaultAclMask() throws Exception {
    mFileSystemMaster.createDirectory(NESTED_URI, CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)));
    AclEntry newAcl = AclEntry.fromCliString("default:user:newuser:rwx");
    // Add an ACL
    addAcl(NESTED_URI, newAcl);
    assertThat(getInfo(NESTED_URI).getDefaultAcl().getEntries(), hasItem(newAcl));

    // Attempt to remove the ACL mask
    AclEntry maskEntry = AclEntry.fromCliString("default:mask::rwx");
    assertThat(getInfo(NESTED_URI).getDefaultAcl().getEntries(), hasItem(maskEntry));
    try {
      removeAcl(NESTED_URI, maskEntry);
      fail("Expected removing the mask from an extended ACL to fail");
    } catch (IOException e) {
      assertThat(e.getMessage(), containsString("mask"));
    }

    // Remove the extended ACL
    removeAcl(NESTED_URI, newAcl);
    // Now we can add and remove a mask
    addAcl(NESTED_URI, maskEntry);
    removeAcl(NESTED_URI, maskEntry);
  }

  private void addAcl(AlluxioURI uri, AclEntry acl) throws Exception {
    mFileSystemMaster.setAcl(uri, SetAclAction.MODIFY, Arrays.asList(acl),
        SetAclContext.defaults());
  }

  private void removeAcl(AlluxioURI uri, AclEntry acl) throws Exception {
    mFileSystemMaster.setAcl(uri, SetAclAction.REMOVE, Arrays.asList(acl),
        SetAclContext.defaults());
  }

  private FileInfo getInfo(AlluxioURI uri) throws Exception {
    return mFileSystemMaster.getFileInfo(uri, GetStatusContext.defaults());
  }

  /**
   * Tests that an exception is in the
   * {@link FileSystemMaster#createFile(AlluxioURI, CreateFileContext)} with a
   * TTL set in the {@link CreateFileContext} after the TTL check was done once.
   */
  @Test
  public void ttlFileDelete() throws Exception {
    CreateFileContext context = CreateFileContext.defaults();
    context.getOptions().setBlockSizeBytes(Constants.KB);
    context.getOptions().setRecursive(true);
    context.getOptions().setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0));
    long fileId = mFileSystemMaster.createFile(NESTED_FILE_URI, context).getFileId();
    FileInfo fileInfo = mFileSystemMaster.getFileInfo(fileId);
    assertEquals(fileInfo.getFileId(), fileId);
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    mThrown.expect(FileDoesNotExistException.class);
    mFileSystemMaster.getFileInfo(fileId);
  }

  /**
   * Tests that TTL delete of a file is not forgotten across restarts.
   */
  @Test
  public void ttlFileDeleteReplay() throws Exception {
    CreateFileContext context = CreateFileContext.defaults();
    context.getOptions().setBlockSizeBytes(Constants.KB);
    context.getOptions().setRecursive(true);
    context.getOptions().setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0));
    long fileId = mFileSystemMaster.createFile(NESTED_FILE_URI, context).getFileId();

    // Simulate restart.
    stopServices();
    startServices();

    FileInfo fileInfo = mFileSystemMaster.getFileInfo(fileId);
    assertEquals(fileInfo.getFileId(), fileId);
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    mThrown.expect(FileDoesNotExistException.class);
    mFileSystemMaster.getFileInfo(fileId);
  }

  /**
   * Tests that an exception is in the
   * {@literal FileSystemMaster#createDirectory(AlluxioURI, CreateDirectoryOptions)}
   * with a TTL set in the {@link CreateDirectoryContext} after the TTL check was done once.
   */
  @Test
  public void ttlDirectoryDelete() throws Exception {
    CreateDirectoryContext context =
        CreateDirectoryContext.mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0)));
    long dirId = mFileSystemMaster.createDirectory(NESTED_DIR_URI, context);
    FileInfo fileInfo = mFileSystemMaster.getFileInfo(dirId);
    assertEquals(fileInfo.getFileId(), dirId);
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // TTL is set to 0, the directory should have been freed during last TTL check.
    assertEquals(0,
            mFileSystemMaster.getFileInfo(NESTED_DIR_URI, GET_STATUS_CONTEXT)
                    .getInAlluxioPercentage());
    mFileSystemMaster.getFileInfo(dirId);
  }

  /**
   * Tests that TTL delete of a directory is not forgotten across restarts.
   */
  @Test
  public void ttlDirectoryDeleteReplay() throws Exception {
    CreateDirectoryContext context =
        CreateDirectoryContext.mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0)));
    long dirId = mFileSystemMaster.createDirectory(NESTED_DIR_URI, context);

    // Simulate restart.
    stopServices();
    startServices();

    FileInfo fileInfo = mFileSystemMaster.getFileInfo(dirId);
    assertEquals(fileInfo.getFileId(), dirId);
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // TTL is set to 0, the directory should have been freed during last TTL check.
    assertEquals(0,
            mFileSystemMaster.getFileInfo(NESTED_DIR_URI, GET_STATUS_CONTEXT)
                    .getInAlluxioPercentage());
    mFileSystemMaster.getFileInfo(dirId);
  }

  /**
   * Tests that file information is still present after it has been freed after the TTL has been set
   * to 0.
   */
  @Test
  public void ttlFileFree() throws Exception {
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    assertEquals(1, mBlockMaster.getBlockInfo(blockId).getLocations().size());
    // Set ttl & operation.
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext.mergeFrom(
        SetAttributePOptions.newBuilder().setCommonOptions(FileSystemOptionsUtils
            .commonDefaults(Configuration.global()).toBuilder().setTtl(0)
            .setTtlAction(alluxio.grpc.TtlAction.FREE))));
    Command heartbeat = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat);
    assertEquals(0, mBlockMaster.getBlockInfo(blockId).getLocations().size());
  }

  /**
   * Tests that TTL free of a file is not forgotten across restarts.
   */
  @Test
  public void ttlFileFreeReplay() throws Exception {
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    assertEquals(1, mBlockMaster.getBlockInfo(blockId).getLocations().size());
    // Set ttl & operation.
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext.mergeFrom(
        SetAttributePOptions.newBuilder().setCommonOptions(FileSystemOptionsUtils
            .commonDefaults(Configuration.global()).toBuilder().setTtl(0)
            .setTtlAction(alluxio.grpc.TtlAction.FREE))));
    // Simulate restart.
    stopServices();
    startServices();

    Command heartbeat = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.<String, StorageList>of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat);
    assertEquals(0, mBlockMaster.getBlockInfo(blockId).getLocations().size());
  }

  /**
   * Tests that file information is still present after it has been freed after the parent
   * directory's TTL has been set to 0.
   */
  @Test
  public void ttlDirectoryFree() throws Exception {
    CreateDirectoryContext directoryContext = CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true));
    mFileSystemMaster.createDirectory(NESTED_URI, directoryContext);
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    assertEquals(1, mBlockMaster.getBlockInfo(blockId).getLocations().size());
    // Set ttl & operation.
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext.mergeFrom(
        SetAttributePOptions.newBuilder().setCommonOptions(FileSystemMasterCommonPOptions
            .newBuilder().setTtl(0).setTtlAction(alluxio.grpc.TtlAction.FREE))));
    Command heartbeat = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.<String, StorageList>of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat);
    assertEquals(0, mBlockMaster.getBlockInfo(blockId).getLocations().size());
  }

  /**
   * Tests that TTL free of a directory is not forgotten across restarts.
   */
  @Test
  public void ttlDirectoryFreeReplay() throws Exception {
    CreateDirectoryContext directoryContext = CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true));
    mFileSystemMaster.createDirectory(NESTED_URI, directoryContext);
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    assertEquals(1, mBlockMaster.getBlockInfo(blockId).getLocations().size());
    // Set ttl & operation.
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext.mergeFrom(
        SetAttributePOptions.newBuilder().setCommonOptions(FileSystemOptionsUtils
            .commonDefaults(Configuration.global()).toBuilder().setTtl(0)
            .setTtlAction(alluxio.grpc.TtlAction.FREE))));

    // Simulate restart.
    stopServices();
    startServices();

    Command heartbeat = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat);
    assertEquals(0, mBlockMaster.getBlockInfo(blockId).getLocations().size());
  }

  /**
   * Tests that an exception is thrown when trying to get information about a file after it
   * has been deleted because of a TTL of 0.
   */
  @Test
  public void setTtlForFileWithNoTtl() throws Exception {
    CreateFileContext context = CreateFileContext.mergeFrom(
        CreateFilePOptions.newBuilder().setBlockSizeBytes(Constants.KB).setRecursive(true));
    long fileId = mFileSystemMaster.createFile(NESTED_FILE_URI, context).getFileId();
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // Since no TTL is set, the file should not be deleted.
    assertEquals(fileId,
        mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).getFileId());

    mFileSystemMaster.setAttribute(NESTED_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0))));
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // TTL is set to 0, the file should have been freed during last TTL check.
    assertEquals(0,
            mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT)
                    .getInAlluxioPercentage());
    mFileSystemMaster.getFileInfo(fileId);
  }

  /**
   * Tests that an exception is thrown when trying to get information about a Directory after
   * it has been deleted because of a TTL of 0.
   */
  @Test
  public void setTtlForDirectoryWithNoTtl() throws Exception {
    CreateDirectoryContext directoryContext = CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true));
    mFileSystemMaster.createDirectory(NESTED_URI, directoryContext);
    mFileSystemMaster.createDirectory(NESTED_DIR_URI, directoryContext);
    CreateFileContext createFileContext = CreateFileContext.mergeFrom(
        CreateFilePOptions.newBuilder().setBlockSizeBytes(Constants.KB).setRecursive(true));
    long fileId = mFileSystemMaster.createFile(NESTED_FILE_URI, createFileContext).getFileId();
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // Since no TTL is set, the file should not be deleted.
    assertEquals(fileId,
        mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).getFileId());
    // Set ttl.
    mFileSystemMaster.setAttribute(NESTED_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0))));
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // TTL is set to 0, the file should have been freed during last TTL check.
    assertEquals(0,
        mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT)
            .getInAlluxioPercentage());
    mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT);
    mFileSystemMaster.getFileInfo(NESTED_DIR_URI, GET_STATUS_CONTEXT);
    mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT);
  }

  /**
   * Tests that an exception is thrown when trying to get information about a file after it
   * has been deleted after the TTL has been set to 0.
   */
  @Test
  public void setSmallerTtlForFileWithTtl() throws Exception {
    CreateFileContext context = CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder()
        .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(Constants.HOUR_MS))
        .setBlockSizeBytes(Constants.KB).setRecursive(true));
    long fileId = mFileSystemMaster.createFile(NESTED_FILE_URI, context).getFileId();
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // Since TTL is 1 hour, the file won't be deleted during last TTL check.
    assertEquals(fileId,
        mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).getFileId());

    mFileSystemMaster.setAttribute(NESTED_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0))));
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // TTL is set to 0, the file should have been freed during last TTL check.
    assertEquals(0,
            mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT)
                    .getInAlluxioPercentage());
    mFileSystemMaster.getFileInfo(fileId);
  }

  /**
   * Tests that an exception is thrown when trying to get information about a Directory after
   * it has been deleted after the TTL has been set to 0.
   */
  @Test
  public void setSmallerTtlForDirectoryWithTtl() throws Exception {
    CreateDirectoryContext directoryContext =
        CreateDirectoryContext.mergeFrom(CreateDirectoryPOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(Constants.HOUR_MS))
            .setRecursive(true));
    mFileSystemMaster.createDirectory(NESTED_URI, directoryContext);
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    assertTrue(
        mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).getName() != null);
    mFileSystemMaster.setAttribute(NESTED_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0))));
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // TTL is set to 0, the file should have been freed during last TTL check.
    assertEquals(0,
            mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT)
                    .getInAlluxioPercentage());
    mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT);
  }

  /**
   * Tests that a file has not been deleted after the TTL has been reset to a valid value.
   */
  @Test
  public void setLargerTtlForFileWithTtl() throws Exception {
    mFileSystemMaster.createDirectory(NESTED_URI, CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)));
    CreateFileContext context = CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder()
        .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0))
        .setBlockSizeBytes(Constants.KB).setRecursive(true));
    long fileId = mFileSystemMaster.createFile(NESTED_FILE_URI, context).getFileId();
    assertEquals(fileId,
        mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).getFileId());

    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setCommonOptions(FileSystemMasterCommonPOptions
            .newBuilder().setTtl(Constants.HOUR_MS))));
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // TTL is reset to 1 hour, the file should not be deleted during last TTL check.
    assertEquals(fileId, mFileSystemMaster.getFileInfo(fileId).getFileId());
  }

  /**
   * Tests that a directory has not been deleted after the TTL has been reset to a valid value.
   */
  @Test
  public void setLargerTtlForDirectoryWithTtl() throws Exception {
    mFileSystemMaster.createDirectory(new AlluxioURI("/nested"),
        CreateDirectoryContext.mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)));
    mFileSystemMaster.createDirectory(NESTED_URI,
        CreateDirectoryContext.mergeFrom(CreateDirectoryPOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0))
            .setRecursive(true)));
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setCommonOptions(FileSystemMasterCommonPOptions
            .newBuilder().setTtl(Constants.HOUR_MS))));
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    // TTL is reset to 1 hour, the directory should not be deleted during last TTL check.
    assertEquals(NESTED_URI.getName(),
        mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).getName());
  }

  /**
   * Tests that the original TTL is removed after setting it to {@link Constants#NO_TTL} for a file.
   */
  @Test
  public void setNoTtlForFileWithTtl() throws Exception {
    mFileSystemMaster.createDirectory(NESTED_URI, CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)));
    CreateFileContext context = CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder()
        .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0))
        .setBlockSizeBytes(Constants.KB).setRecursive(true));
    long fileId = mFileSystemMaster.createFile(NESTED_FILE_URI, context).getFileId();
    // After setting TTL to NO_TTL, the original TTL will be removed, and the file will not be
    // deleted during next TTL check.
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setCommonOptions(FileSystemMasterCommonPOptions
            .newBuilder().setTtl(Constants.NO_TTL))));
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    assertEquals(fileId, mFileSystemMaster.getFileInfo(fileId).getFileId());
  }

  /**
   * Tests that the original TTL is removed after setting it to {@link Constants#NO_TTL} for
   * a directory.
   */
  @Test
  public void setNoTtlForDirectoryWithTtl() throws Exception {
    mFileSystemMaster.createDirectory(new AlluxioURI("/nested"),
        CreateDirectoryContext.mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)));
    mFileSystemMaster.createDirectory(NESTED_URI,
        CreateDirectoryContext.mergeFrom(CreateDirectoryPOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder().setTtl(0))
            .setRecursive(true)));
    // After setting TTL to NO_TTL, the original TTL will be removed, and the file will not be
    // deleted during next TTL check.
    mFileSystemMaster.setAttribute(NESTED_URI, SetAttributeContext
        .mergeFrom(SetAttributePOptions.newBuilder().setCommonOptions(FileSystemMasterCommonPOptions
            .newBuilder().setTtl(Constants.NO_TTL))));
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_TTL_CHECK);
    assertEquals(NESTED_URI.getName(),
        mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).getName());
  }

  /**
   * Tests the {@link FileSystemMaster#setAttribute(AlluxioURI, SetAttributeContext)} method and
   * that an exception is thrown when trying to set a TTL for a directory.
   */
  @Test
  public void setAttribute() throws Exception {
    mFileSystemMaster.createFile(NESTED_FILE_URI, mNestedFileContext);
    FileInfo fileInfo = mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT);
    assertFalse(fileInfo.isPinned());
    assertEquals(Constants.NO_TTL, fileInfo.getTtl());

    // No State.
    mFileSystemMaster.setAttribute(NESTED_FILE_URI, SetAttributeContext.defaults());
    fileInfo = mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT);
    assertFalse(fileInfo.isPinned());
    assertEquals(Constants.NO_TTL, fileInfo.getTtl());

    // Just set pinned flag.
    mFileSystemMaster.setAttribute(NESTED_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder().setPinned(true)));
    fileInfo = mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT);
    assertTrue(fileInfo.isPinned());
    assertEquals(Constants.NO_TTL, fileInfo.getTtl());

    // Both pinned flag and ttl value.
    mFileSystemMaster.setAttribute(NESTED_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder().setPinned(false)
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder()
                .setTtl(1))));
    fileInfo = mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT);
    assertFalse(fileInfo.isPinned());
    assertEquals(1, fileInfo.getTtl());

    mFileSystemMaster.setAttribute(NESTED_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .setCommonOptions(FileSystemMasterCommonPOptions.newBuilder()
                .setTtl(1))));
  }

  /**
   * Tests the permission bits are 0777 for directories and 0666 for files with UMASK 000.
   */
  @Test
  public void permission() throws Exception {
    mFileSystemMaster.createFile(NESTED_FILE_URI, mNestedFileContext);
    assertEquals(0777,
        mFileSystemMaster.getFileInfo(NESTED_URI, GET_STATUS_CONTEXT).getMode());
    assertEquals(0666,
        mFileSystemMaster.getFileInfo(NESTED_FILE_URI, GET_STATUS_CONTEXT).getMode());
  }

  /**
   * Tests {@link FileSystemMaster#free} on persisted file.
   */
  @Test
  public void free() throws Exception {
    mNestedFileContext.setWriteType(WriteType.CACHE_THROUGH);
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    assertEquals(1, mBlockMaster.getBlockInfo(blockId).getLocations().size());

    // free the file
    mFileSystemMaster.free(NESTED_FILE_URI, FreeContext.mergeFrom(FreePOptions.newBuilder()
        .setForced(false).setRecursive(false)));
    // Update the heartbeat of removedBlockId received from worker 1.
    Command heartbeat2 = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat2);
    assertEquals(0, mBlockMaster.getBlockInfo(blockId).getLocations().size());
  }

  /**
   * Tests {@link FileSystemMaster#free} on non-persisted file.
   */
  @Test
  @Ignore
  public void freeNonPersistedFile() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);
    mThrown.expect(UnexpectedAlluxioException.class);
    mThrown.expectMessage(ExceptionMessage.CANNOT_FREE_NON_PERSISTED_FILE
        .getMessage(NESTED_FILE_URI.getPath()));
    // cannot free a non-persisted file
    mFileSystemMaster.free(NESTED_FILE_URI, FreeContext.defaults());
  }

  /**
   * Tests {@link FileSystemMaster#free} on pinned file when forced flag is false.
   */
  @Test
  public void freePinnedFileWithoutForce() throws Exception {
    mNestedFileContext.setWriteType(WriteType.CACHE_THROUGH);
    createFileWithSingleBlock(NESTED_FILE_URI);
    mFileSystemMaster.setAttribute(NESTED_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder().setPinned(true)));
    mThrown.expect(UnexpectedAlluxioException.class);
    mThrown.expectMessage(ExceptionMessage.CANNOT_FREE_PINNED_FILE
        .getMessage(NESTED_FILE_URI.getPath()));
    // cannot free a pinned file without "forced"
    mFileSystemMaster.free(NESTED_FILE_URI, FreeContext.defaults());
  }

  /**
   * Tests {@link FileSystemMaster#free} on pinned file when forced flag is true.
   */
  @Test
  public void freePinnedFileWithForce() throws Exception {
    mNestedFileContext.setWriteType(WriteType.CACHE_THROUGH);
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    mFileSystemMaster.setAttribute(NESTED_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder().setPinned(true)));

    assertEquals(1, mBlockMaster.getBlockInfo(blockId).getLocations().size());

    // free the file
    mFileSystemMaster.free(NESTED_FILE_URI,
        FreeContext.mergeFrom(FreePOptions.newBuilder().setForced(true)));
    // Update the heartbeat of removedBlockId received from worker 1.
    Command heartbeat = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat);
    assertEquals(0, mBlockMaster.getBlockInfo(blockId).getLocations().size());
  }

  /**
   * Tests the {@link FileSystemMaster#free} method with a directory but recursive to false.
   */
  @Test
  public void freeDirNonRecursive() throws Exception {
    mNestedFileContext.setWriteType(WriteType.CACHE_THROUGH);
    createFileWithSingleBlock(NESTED_FILE_URI);
    mThrown.expect(UnexpectedAlluxioException.class);
    mThrown.expectMessage(ExceptionMessage.CANNOT_FREE_NON_EMPTY_DIR.getMessage(NESTED_URI));
    // cannot free directory with recursive argument to false
    mFileSystemMaster.free(NESTED_FILE_URI.getParent(),
        FreeContext.mergeFrom(FreePOptions.newBuilder().setRecursive(false)));
  }

  /**
   * Tests the {@link FileSystemMaster#free} method with a directory.
   */
  @Test
  public void freeDir() throws Exception {
    mNestedFileContext.setWriteType(WriteType.CACHE_THROUGH);
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    assertEquals(1, mBlockMaster.getBlockInfo(blockId).getLocations().size());

    // free the dir
    mFileSystemMaster.free(NESTED_FILE_URI.getParent(),
        FreeContext.mergeFrom(FreePOptions.newBuilder().setForced(true).setRecursive(true)));
    // Update the heartbeat of removedBlockId received from worker 1.
    Command heartbeat3 = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat3);
    assertEquals(0, mBlockMaster.getBlockInfo(blockId).getLocations().size());
  }

  /**
   * Tests the {@link FileSystemMaster#free} method with a directory with a file non-persisted.
   */
  @Test
  @Ignore
  public void freeDirWithNonPersistedFile() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);
    mThrown.expect(UnexpectedAlluxioException.class);
    mThrown.expectMessage(ExceptionMessage.CANNOT_FREE_NON_PERSISTED_FILE
        .getMessage(NESTED_FILE_URI.getPath()));
    // cannot free the parent dir of a non-persisted file
    mFileSystemMaster.free(NESTED_FILE_URI.getParent(),
        FreeContext.mergeFrom(FreePOptions.newBuilder().setForced(false).setRecursive(true)));
  }

  /**
   * Tests the {@link FileSystemMaster#free} method with a directory with a file pinned when
   * forced flag is false.
   */
  @Test
  public void freeDirWithPinnedFileAndNotForced() throws Exception {
    mNestedFileContext.setWriteType(WriteType.CACHE_THROUGH);
    createFileWithSingleBlock(NESTED_FILE_URI);
    mFileSystemMaster.setAttribute(NESTED_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder().setPinned(true)));
    mThrown.expect(UnexpectedAlluxioException.class);
    mThrown.expectMessage(ExceptionMessage.CANNOT_FREE_PINNED_FILE
        .getMessage(NESTED_FILE_URI.getPath()));
    // cannot free the parent dir of a pinned file without "forced"
    mFileSystemMaster.free(NESTED_FILE_URI.getParent(),
        FreeContext.mergeFrom(FreePOptions.newBuilder().setForced(false).setRecursive(true)));
  }

  /**
   * Tests the {@link FileSystemMaster#free} method with a directory with a file pinned when
   * forced flag is true.
   */
  @Test
  public void freeDirWithPinnedFileAndForced() throws Exception {
    mNestedFileContext.setWriteType(WriteType.CACHE_THROUGH);
    long blockId = createFileWithSingleBlock(NESTED_FILE_URI);
    mFileSystemMaster.setAttribute(NESTED_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder().setPinned(true)));
    // free the parent dir of a pinned file with "forced"
    mFileSystemMaster.free(NESTED_FILE_URI.getParent(),
        FreeContext.mergeFrom(FreePOptions.newBuilder().setForced(true).setRecursive(true)));
    // Update the heartbeat of removedBlockId received from worker 1.
    Command heartbeat = mBlockMaster.workerHeartbeat(mWorkerId1, null,
        ImmutableMap.of(Constants.MEDIUM_MEM, (long) Constants.KB), ImmutableList.of(blockId),
        ImmutableMap.of(), ImmutableMap.of(), mMetrics);
    // Verify the muted Free command on worker1.
    assertEquals(Command.newBuilder().setCommandType(CommandType.Nothing).build(), heartbeat);
    assertEquals(0, mBlockMaster.getBlockInfo(blockId).getLocations().size());
  }

  /**
   * Tests the {@link FileSystemMaster#mount(AlluxioURI, AlluxioURI, MountContext)} method.
   */
  @Test
  public void mount() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    mFileSystemMaster.mount(alluxioURI, ufsURI, MountContext.defaults());
  }

  /**
   * Tests mounting an existing dir.
   */
  @Test
  public void mountExistingDir() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    mFileSystemMaster.createDirectory(alluxioURI, CreateDirectoryContext.defaults());
    mThrown.expect(InvalidPathException.class);
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    mFileSystemMaster.mount(alluxioURI, ufsURI, MountContext.defaults());
  }

  /**
   * Tests mounting to an Alluxio path whose parent dir does not exist.
   */
  @Test
  public void mountNonExistingParentDir() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/non-existing/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    mThrown.expect(FileDoesNotExistException.class);
    mFileSystemMaster.mount(alluxioURI, ufsURI, MountContext.defaults());
  }

  /**
   * Tests a readOnly mount for the create directory op.
   */
  @Test
  public void mountReadOnlyCreateDirectory() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    mFileSystemMaster.mount(alluxioURI, ufsURI,
        MountContext.mergeFrom(MountPOptions.newBuilder().setReadOnly(true)));

    mThrown.expect(AccessControlException.class);
    AlluxioURI path = new AlluxioURI("/hello/dir1");
    mFileSystemMaster.createDirectory(path, CreateDirectoryContext.defaults());
  }

  /**
   * Tests a readOnly mount for the create file op.
   */
  @Test
  public void mountReadOnlyCreateFile() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    mFileSystemMaster.mount(alluxioURI, ufsURI,
        MountContext.mergeFrom(MountPOptions.newBuilder().setReadOnly(true)));

    mThrown.expect(AccessControlException.class);
    AlluxioURI path = new AlluxioURI("/hello/file1");
    mFileSystemMaster.createFile(path, CreateFileContext.defaults());
  }

  /**
   * Tests a readOnly mount for the delete op.
   */
  @Test
  public void mountReadOnlyDeleteFile() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    createTempUfsFile("ufs/hello/file1");
    mFileSystemMaster.mount(alluxioURI, ufsURI,
        MountContext.mergeFrom(MountPOptions.newBuilder().setReadOnly(true)));

    mThrown.expect(AccessControlException.class);
    AlluxioURI path = new AlluxioURI("/hello/file1");
    mFileSystemMaster.delete(path, DeleteContext.defaults());
  }

  /**
   * Tests a readOnly mount for the rename op.
   */
  @Test
  public void mountReadOnlyRenameFile() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    createTempUfsFile("ufs/hello/file1");
    mFileSystemMaster.mount(alluxioURI, ufsURI,
        MountContext.mergeFrom(MountPOptions.newBuilder().setReadOnly(true)));

    mThrown.expect(AccessControlException.class);
    AlluxioURI src = new AlluxioURI("/hello/file1");
    AlluxioURI dst = new AlluxioURI("/hello/file2");
    mFileSystemMaster.rename(src, dst, RenameContext.defaults());
  }

  /**
   * Tests a readOnly mount for the set attribute op.
   */
  @Test
  public void mountReadOnlySetAttribute() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    createTempUfsFile("ufs/hello/file1");
    mFileSystemMaster.mount(alluxioURI, ufsURI,
        MountContext.mergeFrom(MountPOptions.newBuilder().setReadOnly(true)));

    mThrown.expect(AccessControlException.class);
    AlluxioURI path = new AlluxioURI("/hello/file1");
    mFileSystemMaster.setAttribute(path,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder().setOwner("owner")));
  }

  /**
   * Tests mounting a shadow Alluxio dir.
   */
  @Test
  public void mountShadowDir() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello/shadow");

    mFileSystemMaster.mount(alluxioURI, ufsURI.getParent(),
        MountContext.defaults());
    AlluxioURI shadowAlluxioURI = new AlluxioURI("/hello/shadow");
    AlluxioURI notShadowAlluxioURI = new AlluxioURI("/hello/notshadow");
    AlluxioURI shadowUfsURI = createTempUfsDir("ufs/hi");
    AlluxioURI notShadowUfsURI = createTempUfsDir("ufs/notshadowhi");
    mFileSystemMaster.mount(notShadowAlluxioURI, notShadowUfsURI,
        MountContext.defaults());
    mThrown.expect(InvalidPathException.class);
    mFileSystemMaster.mount(shadowAlluxioURI, shadowUfsURI,
        MountContext.defaults());
  }

  /**
   * Tests mounting a prefix UFS dir.
   */
  @Test
  public void mountPrefixUfsDir() throws Exception {
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello/shadow");
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    mFileSystemMaster.mount(alluxioURI, ufsURI, MountContext.defaults());
    AlluxioURI preUfsURI = ufsURI.getParent();
    AlluxioURI anotherAlluxioURI = new AlluxioURI("/hi");
    mThrown.expect(InvalidPathException.class);
    mFileSystemMaster.mount(anotherAlluxioURI, preUfsURI, MountContext.defaults());
  }

  /**
   * Tests mounting a suffix UFS dir.
   */
  @Test
  public void mountSuffixUfsDir() throws Exception {
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello/shadow");
    AlluxioURI preUfsURI = ufsURI.getParent();
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    mFileSystemMaster.mount(alluxioURI, preUfsURI, MountContext.defaults());
    AlluxioURI anotherAlluxioURI = new AlluxioURI("/hi");
    mThrown.expect(InvalidPathException.class);
    mFileSystemMaster.mount(anotherAlluxioURI, ufsURI, MountContext.defaults());
  }

  /**
   * Tests unmount operation.
   */
  @Test
  public void unmount() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    mFileSystemMaster.mount(alluxioURI, ufsURI, MountContext.defaults());
    mFileSystemMaster.createDirectory(alluxioURI.join("dir"), CreateDirectoryContext
        .defaults().setWriteType(WriteType.CACHE_THROUGH));
    mFileSystemMaster.unmount(alluxioURI);
    // after unmount, ufs path under previous mount point should still exist
    File file = new File(ufsURI.join("dir").getPath());
    assertTrue(file.exists());
    // after unmount, alluxio path under previous mount point should not exist
    mThrown.expect(FileDoesNotExistException.class);
    mFileSystemMaster.getFileInfo(alluxioURI.join("dir"), GET_STATUS_CONTEXT);
  }

  /**
   * Tests unmount operation failed when unmounting root.
   */
  @Test
  public void unmountRootWithException() throws Exception {
    mThrown.expect(InvalidPathException.class);
    mFileSystemMaster.unmount(new AlluxioURI("/"));
  }

  /**
   * Tests unmount operation failed when unmounting non-mount point.
   */
  @Test
  public void unmountNonMountPointWithException() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    mFileSystemMaster.mount(alluxioURI, ufsURI, MountContext.defaults());
    AlluxioURI dirURI = alluxioURI.join("dir");
    mFileSystemMaster.createDirectory(dirURI, CreateDirectoryContext
        .defaults().setWriteType(WriteType.CACHE_THROUGH));
    mThrown.expect(InvalidPathException.class);
    mFileSystemMaster.unmount(dirURI);
  }

  /**
   * Tests unmount operation failed when unmounting non-existing dir.
   */
  @Test
  public void unmountNonExistingPathWithException() throws Exception {
    mThrown.expect(FileDoesNotExistException.class);
    mFileSystemMaster.unmount(new AlluxioURI("/FileNotExists"));
  }

  /**
   * Tests the {@link DefaultFileSystemMaster#stop()} method.
   */
  @Test
  public void stop() throws Exception {
    mRegistry.stop();
    assertTrue(mExecutorService.isShutdown());
    assertTrue(mExecutorService.isTerminated());
  }

  /**
   * Tests the {@link FileSystemMaster#workerHeartbeat} method.
   */
  @Test
  public void workerHeartbeat() throws Exception {
    long blockId = createFileWithSingleBlock(ROOT_FILE_URI);

    long fileId = mFileSystemMaster.getFileId(ROOT_FILE_URI);
    mFileSystemMaster.scheduleAsyncPersistence(ROOT_FILE_URI,
        ScheduleAsyncPersistenceContext.defaults());

    FileSystemCommand command = mFileSystemMaster
        .workerHeartbeat(mWorkerId1, Lists.newArrayList(fileId), WorkerHeartbeatContext.defaults());
    assertEquals(alluxio.wire.CommandType.PERSIST, command.getCommandType());
    assertEquals(0, command.getCommandOptions().getPersistOptions().getFilesToPersist().size());
  }

  /**
   * Tests that lost files can successfully be detected.
   */
  @Test
  @Ignore
  public void lostFilesDetection() throws Exception {
    createFileWithSingleBlock(NESTED_FILE_URI);
    long fileId = mFileSystemMaster.getFileId(NESTED_FILE_URI);

    FileInfo fileInfo = mFileSystemMaster.getFileInfo(fileId);
    mBlockMaster.reportLostBlocks(fileInfo.getBlockIds());

    assertEquals(PersistenceState.NOT_PERSISTED.name(), fileInfo.getPersistenceState());
    // Check with getPersistenceState.
    assertEquals(PersistenceState.NOT_PERSISTED,
        mFileSystemMaster.getPersistenceState(fileId));

    // run the detector
    HeartbeatScheduler.execute(HeartbeatContext.MASTER_LOST_FILES_DETECTION);

    fileInfo = mFileSystemMaster.getFileInfo(fileId);
    assertEquals(PersistenceState.LOST.name(), fileInfo.getPersistenceState());
    // Check with getPersistenceState.
    assertEquals(PersistenceState.LOST, mFileSystemMaster.getPersistenceState(fileId));
  }

  @Test
  public void getUfsInfo() throws Exception {
    FileInfo alluxioRootInfo =
        mFileSystemMaster.getFileInfo(new AlluxioURI("alluxio://"), GET_STATUS_CONTEXT);
    UfsInfo ufsRootInfo = mFileSystemMaster.getUfsInfo(alluxioRootInfo.getMountId());
    assertEquals(mUnderFS, ufsRootInfo.getUri().getPath());
    assertTrue(ufsRootInfo.getMountOptions().getPropertiesMap().isEmpty());
  }

  @Test
  public void getUfsInfoNotExist() throws Exception {
    UfsInfo noSuchUfsInfo = mFileSystemMaster.getUfsInfo(100L);
    assertNull(noSuchUfsInfo.getUri());
    assertNull(noSuchUfsInfo.getMountOptions());
  }

  /**
   * Tests that setting the ufs fingerprint persists across restarts.
   */
  @Test
  public void setUfsFingerprintReplay() throws Exception {
    String fingerprint = "FINGERPRINT";
    createFileWithSingleBlock(NESTED_FILE_URI);

    ((DefaultFileSystemMaster) mFileSystemMaster).setAttribute(NESTED_FILE_URI,
        SetAttributeContext.defaults().setUfsFingerprint(fingerprint));

    // Simulate restart.
    stopServices();
    startServices();

    assertEquals(fingerprint,
        mFileSystemMaster.getFileInfo(NESTED_FILE_URI,
            GetStatusContext.defaults()).getUfsFingerprint());
  }

  @Test
  public void ignoreInvalidFiles() throws Exception {
    FileUtils.createDir(Paths.get(mUnderFS, "test").toString());
    FileUtils.createFile(Paths.get(mUnderFS, "test", "a?b=C").toString());
    FileUtils.createFile(Paths.get(mUnderFS, "test", "valid").toString());
    List<FileInfo> listing = mFileSystemMaster.listStatus(new AlluxioURI("/test"),
        ListStatusContext.mergeFrom(ListStatusPOptions.newBuilder()
            .setLoadMetadataType(LoadMetadataPType.ALWAYS).setRecursive(true)));
    assertEquals(1, listing.size());
    assertEquals("valid", listing.get(0).getName());
  }

  @Test
  @Ignore
  public void propagatePersisted() throws Exception {
    AlluxioURI nestedFile = new AlluxioURI("/nested1/nested2/file");
    AlluxioURI parent1 = new AlluxioURI("/nested1/");
    AlluxioURI parent2 = new AlluxioURI("/nested1/nested2/");

    createFileWithSingleBlock(nestedFile);

    // Nothing is persisted yet.
    assertEquals(PersistenceState.NOT_PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(nestedFile, GetStatusContext.defaults())
            .getPersistenceState());
    assertEquals(PersistenceState.NOT_PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(parent1,
            GetStatusContext.defaults()).getPersistenceState());
    assertEquals(PersistenceState.NOT_PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(parent2,
            GetStatusContext.defaults()).getPersistenceState());

    // Persist the file.
    mFileSystemMaster.setAttribute(nestedFile,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder().setPersisted(true)));

    // Everything component should be persisted.
    assertEquals(PersistenceState.PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(nestedFile, GetStatusContext.defaults())
            .getPersistenceState());
    assertEquals(PersistenceState.PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(parent1,
            GetStatusContext.defaults()).getPersistenceState());
    assertEquals(PersistenceState.PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(parent2,
            GetStatusContext.defaults()).getPersistenceState());

    // Simulate restart.
    stopServices();
    startServices();

    // Everything component should be persisted.
    assertEquals(PersistenceState.PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(nestedFile, GetStatusContext.defaults())
            .getPersistenceState());
    assertEquals(PersistenceState.PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(parent1,
            GetStatusContext.defaults()).getPersistenceState());
    assertEquals(PersistenceState.PERSISTED.toString(),
        mFileSystemMaster.getFileInfo(parent2,
            GetStatusContext.defaults()).getPersistenceState());
  }

  /**
   * Tests using SetAttribute to apply xAttr to a file, using the TRUNCATE update strategy.
   */
  @Test
  public void setAttributeXAttrTruncate() throws Exception {
    // Create file with xAttr to overwrite
    CreateFileContext context = CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder()
        .setBlockSizeBytes(Constants.KB)
        .putXattr("foo", ByteString.copyFrom("foo", StandardCharsets.UTF_8))
        .setXattrPropStrat(XAttrPropagationStrategy.LEAF_NODE)
        .setRecursive(true));
    FileInfo fileInfo = mFileSystemMaster.createFile(ROOT_FILE_URI, context);
    assertEquals(new String(fileInfo.getXAttr().get("foo"), StandardCharsets.UTF_8), "foo");

    // Replace the xAttr map
    mFileSystemMaster.setAttribute(ROOT_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .putXattr("bar", ByteString.copyFrom("bar", StandardCharsets.UTF_8))
            .setXattrUpdateStrategy(alluxio.proto.journal.File.XAttrUpdateStrategy.TRUNCATE)));
    FileInfo updatedFileInfo = mFileSystemMaster.getFileInfo(ROOT_FILE_URI, GET_STATUS_CONTEXT);
    assertEquals(updatedFileInfo.getXAttr().size(), 1);
    assertEquals(new String(updatedFileInfo.getXAttr().get("bar"), StandardCharsets.UTF_8),
        "bar");
  }

  /**
   * Tests using SetAttribute to apply xAttr to a file, using the UNION_REPLACE update strategy.
   */
  @Test
  public void setAttributeXAttrUnionReplace() throws Exception {
    // Create file with xAttr to overwrite
    CreateFileContext context = CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder()
        .setBlockSizeBytes(Constants.KB)
        .putXattr("foo", ByteString.copyFrom("foo", StandardCharsets.UTF_8))
        .setXattrPropStrat(XAttrPropagationStrategy.LEAF_NODE)
        .setRecursive(true));
    FileInfo fileInfo = mFileSystemMaster.createFile(ROOT_FILE_URI, context);
    assertEquals(new String(fileInfo.getXAttr().get("foo"), StandardCharsets.UTF_8), "foo");

    // Combine the xAttr maps, replacing existing keys
    mFileSystemMaster.setAttribute(ROOT_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .putXattr("foo", ByteString.copyFrom("baz", StandardCharsets.UTF_8))
            .putXattr("bar", ByteString.copyFrom("bar", StandardCharsets.UTF_8))
            .setXattrUpdateStrategy(
                alluxio.proto.journal.File.XAttrUpdateStrategy.UNION_REPLACE)));
    FileInfo updatedFileInfo = mFileSystemMaster.getFileInfo(ROOT_FILE_URI, GET_STATUS_CONTEXT);
    assertEquals(updatedFileInfo.getXAttr().size(), 2);
    assertEquals(new String(updatedFileInfo.getXAttr().get("foo"), StandardCharsets.UTF_8),
        "baz");
    assertEquals(new String(updatedFileInfo.getXAttr().get("bar"), StandardCharsets.UTF_8),
        "bar");
  }

  /**
   * Tests using SetAttribute to apply xAttr to a file, using the UNION_PRESERVE update strategy.
   */
  @Test
  public void setAttributeXAttrUnionPreserve() throws Exception {
    // Create file with xAttr to overwrite
    CreateFileContext context = CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder()
        .setBlockSizeBytes(Constants.KB)
        .putXattr("foo", ByteString.copyFrom("foo", StandardCharsets.UTF_8))
        .setXattrPropStrat(XAttrPropagationStrategy.LEAF_NODE)
        .setRecursive(true));
    FileInfo fileInfo = mFileSystemMaster.createFile(ROOT_FILE_URI, context);
    assertEquals(new String(fileInfo.getXAttr().get("foo"), StandardCharsets.UTF_8), "foo");

    // Combine the xAttr maps, preserving existing keys
    mFileSystemMaster.setAttribute(ROOT_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .putXattr("foo", ByteString.copyFrom("baz", StandardCharsets.UTF_8))
            .putXattr("bar", ByteString.copyFrom("bar", StandardCharsets.UTF_8))
            .setXattrUpdateStrategy(
                alluxio.proto.journal.File.XAttrUpdateStrategy.UNION_PRESERVE)));
    FileInfo updatedFileInfo = mFileSystemMaster.getFileInfo(ROOT_FILE_URI, GET_STATUS_CONTEXT);
    assertEquals(updatedFileInfo.getXAttr().size(), 2);
    assertEquals(new String(updatedFileInfo.getXAttr().get("foo"), StandardCharsets.UTF_8),
        "foo");
    assertEquals(new String(updatedFileInfo.getXAttr().get("bar"), StandardCharsets.UTF_8),
        "bar");
  }

  /**
   * Tests using SetAttribute to apply xAttr to a file, using the DELETE_KEYS update strategy.
   */
  @Test
  public void setAttributeXAttrDeleteKeys() throws Exception {
    // Create file with xAttr to delete
    CreateFileContext context = CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder()
        .setBlockSizeBytes(Constants.KB)
        .putXattr("foo", ByteString.copyFrom("foo", StandardCharsets.UTF_8))
        .setXattrPropStrat(XAttrPropagationStrategy.LEAF_NODE)
        .setRecursive(true));
    FileInfo fileInfo = mFileSystemMaster.createFile(ROOT_FILE_URI, context);
    assertEquals(new String(fileInfo.getXAttr().get("foo"), StandardCharsets.UTF_8), "foo");

    // Delete a non-existing key
    mFileSystemMaster.setAttribute(ROOT_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .putXattr("bar", ByteString.copyFrom("", StandardCharsets.UTF_8))
            .setXattrUpdateStrategy(alluxio.proto.journal.File.XAttrUpdateStrategy.DELETE_KEYS)));
    FileInfo updatedFileInfo = mFileSystemMaster.getFileInfo(ROOT_FILE_URI, GET_STATUS_CONTEXT);
    assertEquals(updatedFileInfo.getXAttr().size(), 1);

    // Delete an existing key
    mFileSystemMaster.setAttribute(ROOT_FILE_URI,
        SetAttributeContext.mergeFrom(SetAttributePOptions.newBuilder()
            .putXattr("foo", ByteString.copyFrom("", StandardCharsets.UTF_8))
            .setXattrUpdateStrategy(alluxio.proto.journal.File.XAttrUpdateStrategy.DELETE_KEYS)));
    updatedFileInfo = mFileSystemMaster.getFileInfo(ROOT_FILE_URI, GET_STATUS_CONTEXT);
    assertNullOrEmpty(updatedFileInfo.getXAttr());
  }

  /**
   * Tests setting xAttr when creating a file in a nested directory, and
   * propagates the xAttr back to the parent directories.
   */
  @Test
  public void createFileXAttrPropagateNewPaths() throws Exception {
    createFileXAttr(XAttrPropagationStrategy.NEW_PATHS);
  }

  /**
   * Tests setting xAttr when creating a file in a nested directory, and
   * only assigns xAttr to the leaf node.
   */
  @Test
  public void createFileXAttrPropagateLeaf() throws Exception {
    createFileXAttr(XAttrPropagationStrategy.LEAF_NODE);
  }

  private void createFileXAttr(XAttrPropagationStrategy propStrat) throws Exception {
    // Set xAttr upon creating the file
    CreateFileContext context = CreateFileContext.mergeFrom(CreateFilePOptions.newBuilder()
        .setBlockSizeBytes(Constants.KB)
        .putXattr("foo", ByteString.copyFrom("bar", StandardCharsets.UTF_8))
        .setRecursive(true)
        .setXattrPropStrat(propStrat));
    FileInfo fileInfo = mFileSystemMaster.createFile(NESTED_FILE_URI, context);
    assertEquals(new String(fileInfo.getXAttr().get("foo"), StandardCharsets.UTF_8), "bar");

    checkParentXAttrTags(NESTED_FILE_URI, propStrat);
  }

  /**
   * Tests setting xAttr when creating a nested subdirectory, and
   * propagates the xAttr back to the parent directories.
   */
  @Test
  public void createDirXAttrPropagateNewPaths() throws Exception {
    createDirXAttr(XAttrPropagationStrategy.NEW_PATHS);
  }

  /**
   * Tests setting xAttr when creating a nested subdirectory, and
   * only assigns xAttr to the leaf node.
   */
  @Test
  public void createDirXAttrPropagateLeaf() throws Exception {
    createDirXAttr(XAttrPropagationStrategy.LEAF_NODE);
  }

  private void createDirXAttr(XAttrPropagationStrategy propStrat) throws Exception {
    // Set xAttr upon creating the directory
    CreateDirectoryContext directoryContext = CreateDirectoryContext.mergeFrom(
        CreateDirectoryPOptions.newBuilder()
            .putXattr("foo", ByteString.copyFrom("bar", StandardCharsets.UTF_8))
            .setXattrPropStrat(propStrat)
            .setRecursive(true));
    long dirId = mFileSystemMaster.createDirectory(NESTED_DIR_URI, directoryContext);
    assertEquals(new String(mFileSystemMaster.getFileInfo(dirId).getXAttr().get("foo"),
        StandardCharsets.UTF_8), "bar");

    checkParentXAttrTags(NESTED_DIR_URI, propStrat);
  }

  private void checkParentXAttrTags(AlluxioURI uri, XAttrPropagationStrategy propStrat)
      throws Exception {
    FileInfo fileInfo = mFileSystemMaster.getFileInfo(uri, GET_STATUS_CONTEXT);
    Map<String, byte[]> xAttrs = fileInfo.getXAttr();
    assertNotEquals(xAttrs, null);

    switch (propStrat) {
      case NEW_PATHS:
        // Verify that the parent directories have matching xAttr
        for (int i = 1; uri.getLeadingPath(i + 1) != null; i++) {
          for (Map.Entry<String, byte[]> entry : xAttrs.entrySet()) {
            assertArrayEquals(mFileSystemMaster.getFileInfo(new AlluxioURI(uri.getLeadingPath(i)),
                GET_STATUS_CONTEXT).getXAttr().get(entry.getKey()), xAttrs.get(entry.getKey()));
          }
        }
        break;
      case LEAF_NODE:
        // Verify that the parent directories have no xAttr
        for (int i = 1; uri.getLeadingPath(i + 1) != null; i++) {
          assertNullOrEmpty(mFileSystemMaster.getFileInfo(new AlluxioURI(uri.getLeadingPath(i)),
              GET_STATUS_CONTEXT).getXAttr());
        }
        break;
      default:
        throw new IllegalArgumentException(
            String.format("Invalid XAttrPropagationStrategy: %s", propStrat));
    }
  }

  @Test
  public void writeToReadOnlyFileWhileCreating() throws Exception {
    mFileSystemMaster.createDirectory(NESTED_URI, CreateDirectoryContext
        .mergeFrom(CreateDirectoryPOptions.newBuilder().setRecursive(true)));
    Set<String> newEntries = Sets.newHashSet("user::r--", "group::r--", "other::r--");
    // The owner of the root path will be treated as a privileged user,
    // so we need another user to do validation.
    String user = "test_user1";
    CreateFileContext context = CreateFileContext
        .mergeFrom(mNestedFileContext.getOptions())
        .setOwner(user)
        .setAcl(newEntries.stream().map(AclEntry::fromCliString).collect(Collectors.toList()));
    try (Closeable r = new AuthenticatedUserRule(user, Configuration.global())
        .toResource()) {
      createFileWithSingleBlock(NESTED_FILE_URI, context);
      try {
        mFileSystemMaster.getNewBlockIdForFile(NESTED_FILE_URI);
        Assert.fail("getNewBlockIdForFile after completed should fail!");
      } catch (AccessControlException e) {
        // ignored
      }
    }
  }

  @Test
  public void RecursiveDeleteForceFlushJournals() throws Exception {
    DefaultFileSystemMaster fileSystemMasterWithSpy = spy(mFileSystemMaster);
    AtomicInteger flushCount = new AtomicInteger();
    AtomicInteger closeCount = new AtomicInteger();
    when(fileSystemMasterWithSpy.createJournalContext(true)).thenReturn(
        new JournalContext() {
          private int mNumLogs = 0;

          @Override
          public void append(Journal.JournalEntry entry) {
            mNumLogs++;
          }

          @Override
          public void flush() {
            if (mNumLogs != 0) {
              flushCount.incrementAndGet();
              mNumLogs = 0;
            }
          }

          @Override
          public void close() {
            closeCount.incrementAndGet();
          }
        }
    );

    int level = 2;
    int numInodes = 4 * IntMath.pow(DIR_WIDTH, level) - 3;
    AlluxioURI ufsMount = createPersistedDirectories(level);
    mountPersistedDirectories(ufsMount);
    loadPersistedDirectories(level);
    // delete top-level directory
    fileSystemMasterWithSpy.delete(new AlluxioURI(MOUNT_URI).join(DIR_TOP_LEVEL),
        DeleteContext.mergeFrom(DeletePOptions.newBuilder().setRecursive(true)
            .setAlluxioOnly(false).setUnchecked(false)));
    checkPersistedDirectoriesDeleted(level, ufsMount, Collections.EMPTY_LIST);
    assertEquals(1, closeCount.get());
    if (Configuration.getBoolean(PropertyKey.MASTER_FILE_SYSTEM_MERGE_INODE_JOURNALS)) {
      assertEquals(numInodes, flushCount.get());
    } else {
      assertEquals(0, flushCount.get());
    }
  }

  /**
   * Tests a readOnly mount for the set attribute op.
   */
  @Test
  public void exists() throws Exception {
    AlluxioURI alluxioURI = new AlluxioURI("/hello");
    AlluxioURI ufsURI = createTempUfsDir("ufs/hello");
    mFileSystemMaster.mount(alluxioURI, ufsURI,
        MountContext.mergeFrom(MountPOptions.newBuilder().setReadOnly(true)));

    AlluxioURI alluxioFileURI = new AlluxioURI("/hello/file");

    ExistsContext neverSyncContext = ExistsContext.create(
        ExistsPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.NEVER).setCommonOptions(
            FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(-1).build()));

    ExistsContext alwaysSyncContext = ExistsContext.create(
        ExistsPOptions.newBuilder().setLoadMetadataType(LoadMetadataPType.ALWAYS).setCommonOptions(
        FileSystemMasterCommonPOptions.newBuilder().setSyncIntervalMs(0).build()));

    Assert.assertFalse(mFileSystemMaster.exists(alluxioFileURI, alwaysSyncContext));

    createTempUfsFile("ufs/hello/file");

    Assert.assertFalse(mFileSystemMaster.exists(alluxioFileURI, neverSyncContext));
    Assert.assertTrue(mFileSystemMaster.exists(alluxioFileURI, alwaysSyncContext));
    Assert.assertTrue(mFileSystemMaster.exists(alluxioFileURI, alwaysSyncContext));
    Assert.assertTrue(mFileSystemMaster.exists(alluxioFileURI, neverSyncContext));

    mTestFolder.delete();

    Assert.assertTrue(mFileSystemMaster.exists(alluxioFileURI, neverSyncContext));
    Assert.assertFalse(mFileSystemMaster.exists(alluxioFileURI, alwaysSyncContext));
  }
}
