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.underfs;
  
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertFalse;
  import static org.junit.Assert.assertNotNull;
  import static org.junit.Assert.assertNull;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assert.fail;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.Mockito.doAnswer;
  import static org.mockito.Mockito.doThrow;
  import static org.mockito.Mockito.mock;
  import static org.mockito.Mockito.times;
  import static org.mockito.Mockito.when;
  
  import alluxio.AlluxioURI;
  import alluxio.Constants;
  import alluxio.conf.Configuration;
  import alluxio.exception.InvalidPathException;
  import alluxio.grpc.MountPOptions;
  import alluxio.master.file.BlockDeletionContext;
  import alluxio.master.file.RpcContext;
  import alluxio.master.file.contexts.CallTracker;
  import alluxio.master.file.contexts.OperationContext;
  import alluxio.master.file.meta.MountTable;
  import alluxio.master.file.meta.NoopUfsAbsentPathCache;
  import alluxio.master.file.meta.UfsAbsentPathCache;
  import alluxio.master.file.meta.options.MountInfo;
  import alluxio.master.journal.JournalContext;
  import alluxio.underfs.local.LocalUnderFileSystem;
  import alluxio.util.IdUtils;
  import alluxio.util.io.PathUtils;
  
  import com.google.common.collect.Lists;
  import org.hamcrest.MatcherAssert;
  import org.hamcrest.Matchers;
  import org.junit.After;
  import org.junit.Before;
  import org.junit.Rule;
  import org.junit.Test;
  import org.junit.rules.ExpectedException;
  import org.junit.rules.TemporaryFolder;
  import org.mockito.Mockito;
  import org.mockito.stubbing.Answer;
  
  import java.io.IOException;
  import java.time.Clock;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.Executors;
  import java.util.concurrent.Future;
  import java.util.concurrent.SynchronousQueue;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.atomic.AtomicReference;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantLock;
  
  public class UfsStatusCacheTest {
  
    @Rule
    public TemporaryFolder mTempDir = new TemporaryFolder();
  
    @Rule
    public ExpectedException mThrown = ExpectedException.none();
  
    private String mUfsUri;
    private LocalUnderFileSystem mUfs;
    private UfsStatusCache mCache;
    private ExecutorService mService;
    private MountTable mMountTable;
  
    @Before
    public void before() throws Exception {
      mUfsUri = mTempDir.newFolder().getAbsolutePath();
      mUfs = new LocalUnderFileSystem(new AlluxioURI(mUfsUri),
          UnderFileSystemConfiguration.defaults(Configuration.global()));
      mService = Executors.newSingleThreadExecutor();
      mCache = new UfsStatusCache(mService, new NoopUfsAbsentPathCache(),
          UfsAbsentPathCache.ALWAYS);
      MountInfo rootMountInfo = new MountInfo(
          new AlluxioURI(MountTable.ROOT),
          new AlluxioURI(mUfsUri),
          IdUtils.ROOT_MOUNT_ID,
          MountPOptions.newBuilder()
              .setReadOnly(false)
              .setShared(false)
              .build());
      MasterUfsManager manager = new MasterUfsManager();
      manager.getRoot(); // add root mount
      mMountTable = new MountTable(manager, rootMountInfo, Clock.systemUTC());
    }
  
    @After
    public void after() throws Exception {
      mService.shutdown();
      if (!mService.awaitTermination(50, TimeUnit.MILLISECONDS)) {
        mService.shutdownNow();
      }
    }
  
    @Test
    public void testMismatchingChildComponent() {
      UfsStatus s = Mockito.mock(UfsStatus.class);
      when(s.getName()).thenReturn("name");
      mThrown.expect(IllegalArgumentException.class);
      mCache.addStatus(new AlluxioURI("/notName"), s);
    }
  
    @Test
    public void testDoubleRemove() {
      UfsStatus s = Mockito.mock(UfsStatus.class);
      when(s.getName()).thenReturn("name");
      AlluxioURI u = new AlluxioURI("/name");
      mCache.addStatus(u, s);
      assertEquals(s, mCache.remove(u));
      assertNull(mCache.remove(u));
    }
  
    @Test
    public void testAddRemove() throws Exception {
      AlluxioURI path = new AlluxioURI("/abc/123");
      UfsStatus stat = Mockito.mock(UfsStatus.class);
      when(stat.getName()).thenReturn("123");
      mCache.addStatus(path, stat);
      assertEquals(stat, mCache.getStatus(path));
      mCache.remove(path);
      assertNull(mCache.getStatus(path));
    }
  
    @Test
    public void testCancelPrefetch() throws Exception {
      for (int i = 65; i < 90; i++) {
        createUfsDirs(Character.getName(i));
      }
      for (int i = 65; i < 90; i++) {
        mCache.prefetchChildren(new AlluxioURI("/" + Character.getName(i)), mMountTable);
      }
      mCache.cancelAllPrefetch();
      assertNull(mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/" + Character.getName(89)),
          mMountTable, false));
    }
  
    @Test
    public void testUseFallback() throws Exception {
      createUfsDirs("a");
      createUfsFile("a/b");
      Collection<UfsStatus> children = mCache
          .fetchChildrenIfAbsent(null, new AlluxioURI("/a"), mMountTable);
      assertEquals(1, children.size());
      children.forEach(stat -> assertEquals("b", stat.getName()));
    }
  
    @Test
    public void testFetchInterruptedException() throws Exception {
      spyUfs();
      doAnswer((Answer<UfsStatus[]>) invocation -> {
        Thread.sleep(30 * Constants.HOUR_MS);
        return new UfsStatus[] {Mockito.mock(UfsStatus.class)};
      }).when(mUfs).listStatus(any(String.class));
      mCache.prefetchChildren(new AlluxioURI("/"), mMountTable);
      AtomicReference<Error> ref = new AtomicReference<>(null);
      Thread t = new Thread(() -> {
        try {
          try {
            mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/"), mMountTable);
            fail("Should not have been able to fetch children");
          } catch (InterruptedException | InvalidPathException e) {
            // Assert interrupted flag was set properly.
            assertTrue(Thread.currentThread().isInterrupted());
            assertTrue(e instanceof InterruptedException);
          }
        } catch (AssertionError err) {
          ref.set(err);
        }
      });
      t.start();
      t.interrupt();
      t.join();
      assertNull(ref.get());
    }
  
    @Test
    public void testFetchCancel() throws Exception {
      spyUfs();
      doAnswer((Answer<UfsStatus[]>) invocation -> {
        Thread.sleep(30 * Constants.HOUR_MS);
        return new UfsStatus[] {Mockito.mock(UfsStatus.class)};
      }).when(mUfs).listStatus(any(String.class));
      mCache.prefetchChildren(new AlluxioURI("/"), mMountTable);
  
      final BlockDeletionContext bdc = mock(BlockDeletionContext.class);
      final JournalContext jc = mock(JournalContext.class);
      final OperationContext oc = mock(OperationContext.class);
      when(oc.getCancelledTrackers()).thenReturn(Lists.newArrayList());
      final RpcContext rpcContext = new RpcContext(bdc, jc, oc);
  
      AtomicReference<RuntimeException> ref = new AtomicReference<>(null);
      Thread t = new Thread(() -> {
        try {
          mCache.fetchChildrenIfAbsent(rpcContext, new AlluxioURI("/"), mMountTable);
          fail("Should not have been able to fetch children");
        } catch (RuntimeException e) {
          ref.set(e);
        } catch (InterruptedException | InvalidPathException e) {
          // do nothing
        }
      });
      t.start();
      when(oc.getCancelledTrackers()).thenReturn(Lists.newArrayList(new CallTracker() {
        @Override
        public boolean isCancelled() {
          return true;
        }
  
        @Override
        public Type getType() {
          return Type.GRPC_CLIENT_TRACKER;
        }
      }));
      t.join();
      final RuntimeException runtimeException = ref.get();
      assertNotNull(runtimeException);
      MatcherAssert.assertThat(runtimeException.getMessage(),
          Matchers.stringContainsInOrder("Call cancelled"));
    }
  
    @Test
    public void testFetchExecutionException() throws Exception {
      spyUfs();
      // Now test execution exception
      AtomicReference<Error> ref = new AtomicReference<>(null);
  
      doThrow(new RuntimeException("Purposefully thrown exception."))
          .when(mUfs).listStatus(any(String.class));
      mCache.prefetchChildren(new AlluxioURI("/"), mMountTable);
      Thread t = new Thread(() -> {
        try {
          try {
            assertNull("Should return null when fetching children",
                mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/"), mMountTable, false));
          } catch (InterruptedException | InvalidPathException e) {
            // Thread should not be interrupted if interrupt not called
            assertFalse(Thread.currentThread().isInterrupted());
            assertTrue(e instanceof InterruptedException);
          }
        } catch (AssertionError err) {
          ref.set(err);
        }
      });
      t.start();
      t.join();
      assertNull(ref.get());
    }
  
    @Test
    public void testAddRemoveChildren() throws Exception {
      AlluxioURI path = new AlluxioURI("/abc");
      AlluxioURI pathChild = new AlluxioURI("/abc/123");
  
      UfsStatus stat = Mockito.mock(UfsStatus.class);
      when(stat.getName()).thenReturn("abc");
  
      UfsStatus statChild = Mockito.mock(UfsStatus.class);
      when(statChild.getName()).thenReturn("123");
  
      mCache.addStatus(path, stat);
      mCache.addChildren(path, Collections.singleton(statChild));
      assertEquals(stat, mCache.getStatus(path));
      assertEquals(statChild, mCache.getStatus(pathChild));
      assertEquals(Collections.singleton(statChild), mCache.getChildren(path));
      assertEquals(Collections.singleton(statChild),
          mCache.fetchChildrenIfAbsent(null, path, mMountTable));
  
      mCache.remove(path);
      assertNull(mCache.getStatus(path));
      assertNull(mCache.getChildren(path));
      assertNotNull(mCache.getStatus(pathChild));
      assertEquals(statChild, mCache.getStatus(pathChild));
    }
  
    @Test
    public void conflictingParentStatus() throws Exception {
      AlluxioURI path = new AlluxioURI("/mnt/dir1/dir0/dir2");
      AlluxioURI path2 = new AlluxioURI("/mnt/dir2/dir0/dir3");
      createUfsDirs(path.getPath());
      createUfsDirs(path2.getPath());
      // Both parent dirs have the same name - previously caused issues with the way child references
      // were stored
      mCache.addStatus(new AlluxioURI("/mnt/dir1/dir0"),
          mUfs.getStatus(PathUtils.concatPath(mUfsUri, "/mnt/dir1/dir0")).setName("dir0"));
      mCache.addStatus(new AlluxioURI("/mnt/dir2/dir0"),
          mUfs.getStatus(PathUtils.concatPath(mUfsUri, "/mnt/dir2/dir0")).setName("dir0"));
      mCache.prefetchChildren(new AlluxioURI("/mnt/dir1/dir0"), mMountTable);
      mCache.prefetchChildren(new AlluxioURI("/mnt/dir2/dir0"), mMountTable);
      Collection<UfsStatus> children;
      children = mCache.fetchChildrenIfAbsent(
          null, new AlluxioURI("/mnt/dir1/dir0"), mMountTable, false);
      assertNotNull(children);
      assertEquals(1, children.size());
      children.forEach(s -> assertEquals("dir2", s.getName()));
      children = mCache.fetchChildrenIfAbsent(
          null, new AlluxioURI("/mnt/dir2/dir0"), mMountTable, false);
      assertNotNull(children);
      assertEquals(1, children.size());
      children.forEach(s -> assertEquals("dir3", s.getName()));
      children = mCache.fetchChildrenIfAbsent(
          null, new AlluxioURI("/mnt/dir1/dir0"), mMountTable, false);
      assertNotNull(children);
      assertEquals(1, children.size());
      children.forEach(s -> assertEquals("dir2", s.getName()));
    }
  
    @Test
    public void testPrefetch() throws Exception {
      createUfsDirs("dir0/dir0");
      createUfsFile("dir0/dir0/file");
      mCache.prefetchChildren(new AlluxioURI("/dir0/dir0"), mMountTable);
      mCache.prefetchChildren(new AlluxioURI("/dir0"), mMountTable);
      Collection<UfsStatus> statuses =
          mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/dir0/dir0"), mMountTable, false);
      assertEquals(1, statuses.size());
      statuses.forEach(s -> assertEquals("file", s.getName()));
      statuses = mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/dir0"), mMountTable, false);
      assertEquals(1, statuses.size());
      statuses.forEach(s -> assertEquals("dir0", s.getName()));
    }
  
    @Test
    public void testDoublePrefetch() throws Exception {
      createUfsFile("dir0/dir0/file");
      createUfsDirs("dir0/dir0");
      createUfsFile("dir0/dir0/file");
      mCache = Mockito.spy(mCache);
      Lock l = new ReentrantLock();
      doAnswer((invocation) -> {
        try {
          l.lock();
          return invocation.callRealMethod();
        } finally {
          l.unlock();
        }
      }).when(mCache).getChildrenIfAbsent(any(AlluxioURI.class), any(MountTable.class));
      l.lock();
      Future<?> f1 = mCache.prefetchChildren(new AlluxioURI("/dir0"), mMountTable);
      Future<?> f2 = mCache.prefetchChildren(new AlluxioURI("/dir0"), mMountTable);
      assertNotNull(f1);
      assertFalse("first future is not cancelled", f1.isCancelled());
      assertEquals("the same job is running on the same path", f1, f2);
      Collection<UfsStatus> statuses =
          mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/dir0/dir0"), mMountTable, true);
      assertEquals(1, statuses.size());
      statuses.forEach(s -> assertEquals("file", s.getName()));
      l.unlock();
      statuses = mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/dir0"), mMountTable, false);
      assertNotNull(f2);
      assertTrue("second future should be finished", f2.isDone());
      assertEquals(1, statuses.size());
      statuses.forEach(s -> assertEquals("dir0", s.getName()));
    }
  
    @Test
    public void testNullExecutor() throws Exception {
      createUfsDirs("dir0/dir0");
      mCache = new UfsStatusCache(null, new NoopUfsAbsentPathCache(),
          UfsAbsentPathCache.ALWAYS);
      mCache.prefetchChildren(new AlluxioURI("/dir0"), mMountTable);
      assertNull(mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/dir0"), mMountTable, false));
    }
  
    @Test
    public void testRejectedExecution() throws Exception {
      createUfsDirs("dir0/dir1");
      ExecutorService executor =
          new ThreadPoolExecutor(1, 1, 1, TimeUnit.MINUTES, new SynchronousQueue<>());
      mCache = new UfsStatusCache(executor, new NoopUfsAbsentPathCache(),
          UfsAbsentPathCache.ALWAYS);
      mCache = Mockito.spy(mCache);
      Lock l = new ReentrantLock();
      l.lock();
      doAnswer((invocation) -> {
        try {
          l.lock();
          return invocation.callRealMethod();
        } finally {
          l.unlock();
        }
      }).when(mCache).getChildrenIfAbsent(any(AlluxioURI.class), any(MountTable.class));
      Future<Collection<UfsStatus>> job = mCache.prefetchChildren(new AlluxioURI("/dir0"),
          mMountTable);
      assertNotNull(job);
      // the same job should still be running
      assertEquals(job, mCache.prefetchChildren(new AlluxioURI("/dir0"), mMountTable));
      // the following will be rejected as there is no available thread to execute them
      assertNull(mCache.prefetchChildren(new AlluxioURI("/dir1"), mMountTable));
      assertNull(mCache.prefetchChildren(new AlluxioURI("/dir1"), mMountTable));
      assertNull(mCache.prefetchChildren(new AlluxioURI("/dir1"), mMountTable));
      assertNull(mCache.prefetchChildren(new AlluxioURI("/dir1"), mMountTable));
      l.unlock();
      Collection<UfsStatus> statuses =
          mCache.fetchChildrenIfAbsent(null, new AlluxioURI("/dir0"), mMountTable, false);
      assertEquals(1, statuses.size());
      statuses.forEach(s -> assertEquals("dir1", s.getName()));
    }
  
    @Test
    public void testFetchNonExistingSingleStatus() throws Exception {
      spyUfs();
      createUfsFile("testFile");
      UfsStatus status = mCache.fetchStatusIfAbsent(new AlluxioURI("/testFile"), mMountTable);
      assertNotNull(status);
      assertEquals("testFile", status.getName());
      Mockito.verify(mUfs, times(1)).getStatus(any(String.class));
    }
  
    @Test
    public void testFetchExistingSingleStatus() throws Exception {
      spyUfs();
      createUfsFile("testFile");
      UfsStatus status = mUfs.getStatus(PathUtils.concatPath(mUfsUri, "testFile"));
      status.setName("testFile");
      mCache.addStatus(new AlluxioURI("/testFile"), status);
      UfsStatus fetched = mCache.fetchStatusIfAbsent(new AlluxioURI("/testFile"), mMountTable);
      Mockito.verify(mUfs, times(1)).getStatus(any(String.class));
      assertNotNull(fetched);
      assertEquals("testFile", fetched.getName());
    }
  
    @Test
    public void testFetchSingleStatusNonExistingPath() throws Exception {
      spyUfs();
      assertNull(mCache.fetchStatusIfAbsent(new AlluxioURI("/testFile"), mMountTable));
      Mockito.verify(mUfs, times(1)).getStatus(any(String.class));
    }
  
    @Test
    public void testFetchSingleStatusThrowsException() throws Exception {
      spyUfs();
      doThrow(new IOException("test exception")).when(mUfs).getStatus(any(String.class));
      assertNull(mCache.fetchStatusIfAbsent(new AlluxioURI("/testFile"), mMountTable));
      Mockito.verify(mUfs, times(1)).getStatus(any(String.class));
    }
  
    /**
     * Recreates the mount table with the local UFS as a spy'd mockito object.
     */
    private void spyUfs() {
      mUfs = Mockito.spy(mUfs);
      MountInfo rootMountInfo = new MountInfo(
          new AlluxioURI(MountTable.ROOT),
          new AlluxioURI(mUfsUri),
          IdUtils.ROOT_MOUNT_ID,
          MountPOptions.newBuilder()
              .setReadOnly(false)
              .setShared(false)
              .build());
      MasterUfsManager manager = new MasterUfsManager();
      manager.getRoot(); // add root mount
      manager.mUnderFileSystemMap.put(new AbstractUfsManager.Key(new AlluxioURI("/"), null), mUfs);
      mMountTable = new MountTable(manager, rootMountInfo, Clock.systemUTC());
    }
  
    public void createUfsFile(String relPath) throws Exception {
      mUfs.create(PathUtils.concatPath(mUfsUri, relPath)).close();
    }
  
    public void createUfsDirs(String relPath) throws Exception {
      mUfs.mkdirs(PathUtils.concatPath(mUfsUri, relPath));
    }
  }
  