content: /*
   * 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.meta;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertNull;
  import static org.junit.Assert.assertTrue;
  import static org.mockito.Mockito.mock;
  
  import alluxio.AlluxioURI;
  import alluxio.TestLoggerRule;
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.InvalidPathException;
  import alluxio.exception.status.UnavailableException;
  import alluxio.master.file.meta.InodeTree.LockPattern;
  import alluxio.master.journal.FileSystemMergeJournalContext;
  import alluxio.master.journal.JournalContext;
  import alluxio.master.journal.NoopJournalContext;
  
  import org.junit.After;
  import org.junit.Assert;
  import org.junit.Rule;
  import org.junit.Test;
  import org.mockito.Mockito;
  
  import java.util.Arrays;
  import java.util.concurrent.atomic.AtomicInteger;
  
  /**
   * Unit tests for {@link LockedInodePath}.
   */
  public class LockedInodePathTest extends BaseInodeLockingTest {
    private LockedInodePath mPath;
  
    @Rule
    public TestLoggerRule mLogger = new TestLoggerRule();
  
    @After
    public void after() {
      if (mPath != null) {
        mPath.close();
      }
      super.after();
    }
  
    @Test
    public void pathExistsReadLock() throws Exception {
      AlluxioURI uri = new AlluxioURI("/a/b/c");
      mPath =
          new LockedInodePath(
              uri, mInodeStore, mInodeLockManager, mRootDir,
              LockPattern.READ, false, NoopJournalContext.INSTANCE
          );
      assertEquals(uri, mPath.getUri());
      assertEquals(4, mPath.size());
  
      mPath.traverse();
      assertTrue(mPath.fullPathExists());
      assertEquals(mFileC, mPath.getInode());
      assertEquals(mFileC, mPath.getInodeOrNull());
      assertEquals(mFileC, mPath.getInodeFile());
      assertEquals(mFileC, mPath.getLastExistingInode());
      assertEquals(mDirB, mPath.getParentInodeDirectory());
      assertEquals(mDirB, mPath.getParentInodeOrNull());
      assertEquals(mDirB, mPath.getAncestorInode());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB, mFileC), mPath.getInodeList());
      assertEquals(4, mPath.getExistingInodeCount());
      assertEquals(LockPattern.READ, mPath.getLockPattern());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void pathExistsWriteLock() throws Exception {
      mPath = create("/a/b/c", LockPattern.WRITE_INODE);
  
      assertTrue(mPath.fullPathExists());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB, mFileC), mPath.getInodeList());
      assertEquals(LockPattern.WRITE_INODE, mPath.getLockPattern());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyNodesWriteLocked(mFileC);
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void pathExistsWriteEdgeLock() throws Exception {
      mPath = create("/a/b/c", LockPattern.WRITE_EDGE);
  
      assertTrue(mPath.fullPathExists());
      assertEquals(mFileC, mPath.getInode());
      assertEquals(mFileC, mPath.getInodeOrNull());
      assertEquals(mFileC, mPath.getInodeFile());
      assertEquals(mFileC, mPath.getLastExistingInode());
      assertEquals(mDirB, mPath.getParentInodeDirectory());
      assertEquals(mDirB, mPath.getParentInodeOrNull());
      assertEquals(mDirB, mPath.getAncestorInode());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB, mFileC), mPath.getInodeList());
      assertEquals(4, mPath.getExistingInodeCount());
      assertEquals(LockPattern.WRITE_EDGE, mPath.getLockPattern());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyNodesWriteLocked(mFileC);
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyIncomingEdgesWriteLocked(mFileC);
    }
  
    @Test
    public void missingLastReadLock() throws Exception {
      mPath = create("/a/b/missing", LockPattern.READ);
  
      assertFalse(mPath.fullPathExists());
      assertNull(mPath.getInodeOrNull());
      assertEquals(mDirB, mPath.getLastExistingInode());
      assertEquals(mDirB, mPath.getParentInodeDirectory());
      assertEquals(mDirB, mPath.getParentInodeOrNull());
      assertEquals(mDirB, mPath.getAncestorInode());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB), mPath.getInodeList());
      assertEquals(3, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void missingLastWriteLock() throws Exception {
      mPath = create("/a/b/missing", LockPattern.WRITE_INODE);
  
      assertFalse(mPath.fullPathExists());
      assertNull(mPath.getInodeOrNull());
      assertEquals(mDirB, mPath.getLastExistingInode());
      assertEquals(mDirB, mPath.getParentInodeDirectory());
      assertEquals(mDirB, mPath.getParentInodeOrNull());
      assertEquals(mDirB, mPath.getAncestorInode());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB), mPath.getInodeList());
      assertEquals(3, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB);
      checkIncomingEdgeReadLocked(mDirB.getId(), "missing");
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void missingLastWriteEdgeLock() throws Exception {
      mPath = create("/a/b/missing", LockPattern.WRITE_EDGE);
  
      assertFalse(mPath.fullPathExists());
      assertNull(mPath.getInodeOrNull());
      assertEquals(mDirB, mPath.getLastExistingInode());
      assertEquals(mDirB, mPath.getParentInodeDirectory());
      assertEquals(mDirB, mPath.getParentInodeOrNull());
      assertEquals(mDirB, mPath.getAncestorInode());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB), mPath.getInodeList());
      assertEquals(3, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyIncomingEdgesWriteLocked();
      checkIncomingEdgeWriteLocked(mDirB.getId(), "missing");
    }
  
    @Test
    public void missingMultipleReadLock() throws Exception {
      mPath = create("/a/miss1/miss2", LockPattern.READ);
  
      assertFalse(mPath.fullPathExists());
      assertNull(mPath.getInodeOrNull());
      assertNull(mPath.getParentInodeOrNull());
      assertEquals(mDirA, mPath.getLastExistingInode());
      assertEquals(mDirA, mPath.getAncestorInode());
      assertEquals(Arrays.asList(mRootDir, mDirA), mPath.getInodeList());
      assertEquals(2, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void missingMultipleWriteEdgeLock() throws Exception {
      mPath = create("/a/miss1/miss2", LockPattern.WRITE_EDGE);
  
      assertFalse(mPath.fullPathExists());
      assertNull(mPath.getInodeOrNull());
      assertNull(mPath.getParentInodeOrNull());
      assertEquals(mDirA, mPath.getLastExistingInode());
      assertEquals(mDirA, mPath.getAncestorInode());
      assertEquals(Arrays.asList(mRootDir, mDirA), mPath.getInodeList());
      assertEquals(2, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked();
      checkIncomingEdgeWriteLocked(mDirA.getId(), "miss1");
    }
  
    @Test
    public void readLockRoot() throws Exception {
      mPath = create("/", LockPattern.READ);
  
      assertTrue(mPath.fullPathExists());
      assertEquals(mRootDir, mPath.getInodeOrNull());
      assertNull(mPath.getParentInodeOrNull());
      assertEquals(mRootDir, mPath.getLastExistingInode());
      assertEquals(Arrays.asList(mRootDir), mPath.getInodeList());
      assertEquals(1, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void writeLockRoot() throws Exception {
      mPath = create("/", LockPattern.WRITE_INODE);
  
      assertTrue(mPath.fullPathExists());
      assertEquals(mRootDir, mPath.getInodeOrNull());
      assertNull(mPath.getParentInodeOrNull());
      assertEquals(mRootDir, mPath.getLastExistingInode());
      assertEquals(Arrays.asList(mRootDir), mPath.getInodeList());
      assertEquals(1, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked();
      checkOnlyNodesWriteLocked(mRootDir);
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void writeEdgeLockRoot() throws Exception {
      mPath = create("/", LockPattern.WRITE_EDGE);
  
      assertTrue(mPath.fullPathExists());
      assertEquals(mRootDir, mPath.getInodeOrNull());
      assertNull(mPath.getParentInodeOrNull());
      assertEquals(mRootDir, mPath.getLastExistingInode());
      assertEquals(Arrays.asList(mRootDir), mPath.getInodeList());
      assertEquals(1, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked();
      checkOnlyNodesWriteLocked(mRootDir);
      checkOnlyIncomingEdgesReadLocked();
      checkOnlyIncomingEdgesWriteLocked(mRootDir);
    }
  
    @Test
    public void removeLastReadLockedInode() throws Exception {
      mPath = create("/a", LockPattern.READ);
  
      mPath.removeLastInode();
      assertFalse(mPath.fullPathExists());
      assertNull(mPath.getInodeOrNull());
      assertEquals(Arrays.asList(mRootDir), mPath.getInodeList());
      assertEquals(1, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void removeLastWriteEdgeLockedInode() throws Exception {
      mPath = create("/a", LockPattern.WRITE_EDGE);
  
      mPath.removeLastInode();
      assertFalse(mPath.fullPathExists());
      assertNull(mPath.getInodeOrNull());
      assertEquals(Arrays.asList(mRootDir), mPath.getInodeList());
      assertEquals(1, mPath.getExistingInodeCount());
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked(mDirA);
    }
  
    @Test
    public void removeLastInodeImplicitlyLocked() throws Exception {
      mPath = create("/a", LockPattern.WRITE_EDGE);
  
      LockedInodePath pathC = mPath.lockDescendant(new AlluxioURI("/a/b/c"), LockPattern.READ);
      assertTrue(pathC.fullPathExists());
      pathC.removeLastInode();
      assertFalse(pathC.fullPathExists());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB), pathC.getInodeList());
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked(mDirA, mDirB);
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked(mDirA, mDirB, mFileC);
  
      pathC.close();
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked(mDirA);
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked(mDirA);
    }
  
    @Test
    public void addNextFinalInode() throws Exception {
      mPath = create("/a/missing", LockPattern.WRITE_EDGE);
  
      assertFalse(mPath.fullPathExists());
      InodeFile missingInode = inodeFile(10, mDirA.getId(), "missing");
      mInodeStore.addChild(mDirA.getId(), missingInode);
      mPath.addNextInode(missingInode);
      assertTrue(mPath.fullPathExists());
      assertEquals(Arrays.asList(mRootDir, mDirA, missingInode), mPath.getInodeList());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked(missingInode);
    }
  
    @Test
    public void addNextSecondToLastInode() throws Exception {
      mPath = create("/a/miss1/miss2", LockPattern.WRITE_EDGE);
  
      assertFalse(mPath.fullPathExists());
      InodeFile firstMissingInode = inodeFile(10, mDirA.getId(), "miss1");
      mInodeStore.addChild(mDirA.getId(), firstMissingInode);
      mPath.addNextInode(firstMissingInode);
      assertFalse(mPath.fullPathExists());
      assertEquals(Arrays.asList(mRootDir, mDirA, firstMissingInode), mPath.getInodeList());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, firstMissingInode);
      checkOnlyIncomingEdgesWriteLocked();
      // Write lock should be pushed forward when adding a non-final inode.
      checkIncomingEdgeWriteLocked(firstMissingInode.getId(), "miss2");
    }
  
    @Test
    public void downgradeWriteEdgeToRead() throws Exception {
      mPath = create("/a/b/c", LockPattern.WRITE_EDGE);
  
      mPath.downgradeToRead();
      assertTrue(mPath.fullPathExists());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB, mFileC), mPath.getInodeList());
      assertEquals(LockPattern.READ, mPath.getLockPattern());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void downgradeWriteInodeToReadInode() throws Exception {
      mPath = create("/a/b/c", LockPattern.WRITE_INODE);
  
      mPath.downgradeToRead();
      assertTrue(mPath.fullPathExists());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB, mFileC), mPath.getInodeList());
      assertEquals(LockPattern.READ, mPath.getLockPattern());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void lockChildReadToWriteEdge() throws Exception {
      mPath = create("/a", LockPattern.READ);
  
      LockedInodePath childPath = mPath.lockChild(mDirB, LockPattern.WRITE_EDGE);
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB), childPath.getInodeList());
      assertEquals(LockPattern.WRITE_EDGE, childPath.getLockPattern());
      assertTrue(childPath.fullPathExists());
      assertEquals(mDirB, childPath.getInode());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked(mDirB);
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked(mDirB);
  
      childPath.close();
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void lockChildReadToWriteInode() throws Exception {
      mPath = create("/a", LockPattern.READ);
  
      LockedInodePath childPath = mPath.lockChild(mDirB, LockPattern.WRITE_INODE);
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB), childPath.getInodeList());
      assertEquals(LockPattern.WRITE_INODE, childPath.getLockPattern());
      assertTrue(childPath.fullPathExists());
      assertEquals(mDirB, childPath.getInode());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked(mDirB);
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyIncomingEdgesWriteLocked();
  
      childPath.close();
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void lockChildWriteInodeToWriteEdge() throws Exception {
      mPath = create("/a", LockPattern.WRITE_INODE);
  
      LockedInodePath childPath = mPath.lockChild(mDirB, LockPattern.WRITE_EDGE);
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB), childPath.getInodeList());
      assertEquals(LockPattern.WRITE_EDGE, childPath.getLockPattern());
      assertTrue(childPath.fullPathExists());
      assertEquals(mDirB, childPath.getInode());
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked(mDirA, mDirB);
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked(mDirB);
  
      childPath.close();
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked(mDirA);
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void lockChildReadToRead() throws Exception {
      mPath = create("/a", LockPattern.READ);
  
      LockedInodePath childPath = mPath.lockChild(mDirB, LockPattern.READ);
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB), childPath.getInodeList());
      assertEquals(LockPattern.READ, childPath.getLockPattern());
      assertTrue(childPath.fullPathExists());
      assertEquals(mDirB, childPath.getInode());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyIncomingEdgesWriteLocked();
  
      childPath.close();
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void lockChildMultipleReadExtensions() throws Exception {
      mPath = create("/a", LockPattern.READ);
  
      LockedInodePath childPath1 = mPath.lockChild(mDirB, LockPattern.READ);
      LockedInodePath childPath2 = childPath1.lockChild(mFileC, LockPattern.READ);
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyIncomingEdgesWriteLocked();
  
      childPath2.close();
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyIncomingEdgesWriteLocked();
  
      childPath1.close();
  
      checkOnlyNodesReadLocked(mRootDir, mDirA);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void lockDescendantReadToWriteEdge() throws Exception {
      mPath = create("/", LockPattern.READ);
  
      LockedInodePath childPath =
          mPath.lockDescendant(new AlluxioURI("/a/b/c"), LockPattern.WRITE_EDGE);
      assertTrue(childPath.fullPathExists());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB, mFileC), childPath.getInodeList());
  
      checkOnlyNodesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyNodesWriteLocked(mFileC);
      checkOnlyIncomingEdgesReadLocked(mRootDir, mDirA, mDirB);
      checkOnlyIncomingEdgesWriteLocked(mFileC);
  
      childPath.close();
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void lockDescendantWriteEdgeToWriteEdge() throws Exception {
      mPath = create("/", LockPattern.WRITE_EDGE);
  
      LockedInodePath childPath =
          mPath.lockDescendant(new AlluxioURI("/a/b/c"), LockPattern.WRITE_EDGE);
      assertTrue(childPath.fullPathExists());
      assertEquals(Arrays.asList(mRootDir, mDirA, mDirB, mFileC), childPath.getInodeList());
  
      checkOnlyNodesReadLocked();
      checkOnlyNodesWriteLocked(mRootDir, mDirA, mDirB, mFileC);
      checkOnlyIncomingEdgesReadLocked();
      checkOnlyIncomingEdgesWriteLocked(mRootDir, mDirA, mDirB, mFileC);
  
      childPath.close();
  
      checkOnlyNodesReadLocked();
      checkOnlyNodesWriteLocked(mRootDir);
      checkOnlyIncomingEdgesReadLocked();
      checkOnlyIncomingEdgesWriteLocked(mRootDir);
    }
  
    @Test
    public void lockFinalEdgeWrite() throws Exception {
      mInodeStore.removeChild(mRootDir.getId(), "a");
      mPath = create("/a", LockPattern.READ);
      mPath.traverse();
  
      LockedInodePath writeLocked = mPath.lockFinalEdgeWrite();
      assertFalse(writeLocked.fullPathExists());
      assertEquals(Arrays.asList(mRootDir), writeLocked.getInodeList());
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked(mDirA);
  
      writeLocked.close();
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked();
    }
  
    @Test
    public void lockFinalEdgeWriteAlreadyLocked() throws Exception {
      mInodeStore.removeChild(mRootDir.getId(), "a");
      mPath = create("/a", LockPattern.WRITE_EDGE);
  
      LockedInodePath writeLocked = mPath.lockFinalEdgeWrite();
      assertFalse(writeLocked.fullPathExists());
      assertEquals(Arrays.asList(mRootDir), writeLocked.getInodeList());
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked(mDirA);
  
      writeLocked.close();
  
      checkOnlyNodesReadLocked(mRootDir);
      checkOnlyNodesWriteLocked();
      checkOnlyIncomingEdgesReadLocked(mRootDir);
      checkOnlyIncomingEdgesWriteLocked(mDirA);
    }
  
    @Test
    public void testFlushJournal() throws InvalidPathException, UnavailableException {
      AtomicInteger journalFlushCount = new AtomicInteger();
      JournalContext journalContext = mock(FileSystemMergeJournalContext.class);
      Mockito.doAnswer(
          (mock) -> {
            journalFlushCount.getAndIncrement();
            return null;
          }
      ).when(journalContext).flush();
      Configuration.set(
          PropertyKey.MASTER_FILE_SYSTEM_MERGE_INODE_JOURNALS,
          true);
  
      try (LockedInodePath path =
               create("/a/missing/missing2", LockPattern.WRITE_EDGE, journalContext)) {
        InodeDirectory inodeB = inodeDir(10, mDirA.getId(), "missing");
        Assert.assertEquals(0, journalFlushCount.get());
  
        path.addNextInode(inodeB);
        Assert.assertEquals(1, journalFlushCount.get());
  
        // Add the last inode in the path doesn't trigger a journal flush
        InodeFile inodeC = inodeFile(11, inodeB.getId(), "missing2");
        path.addNextInode(inodeC);
        Assert.assertEquals(1, journalFlushCount.get());
  
        path.downgradeToRead();
        Assert.assertEquals(2, journalFlushCount.get());
      }
      Assert.assertEquals(3, journalFlushCount.get());
    }
  
    private LockedInodePath create(String path, LockPattern lockPattern) throws InvalidPathException {
      return create(path, lockPattern, NoopJournalContext.INSTANCE);
    }
  
    private LockedInodePath create(
        String path, LockPattern lockPattern, JournalContext journalContext
    ) throws InvalidPathException {
      LockedInodePath lockedPath = new LockedInodePath(new AlluxioURI(path), mInodeStore,
          mInodeLockManager, mRootDir, lockPattern, false, journalContext);
      lockedPath.traverse();
      return lockedPath;
    }
  }
  