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

package alluxio.client.fs;

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

import alluxio.AlluxioURI;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.FileSystem;
import alluxio.client.file.FileSystemTestUtils;
import alluxio.client.file.FileSystemUtils;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.AlluxioException;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.WritePType;
import alluxio.master.LocalAlluxioJobCluster;
import alluxio.testutils.BaseIntegrationTest;
import alluxio.testutils.LocalAlluxioClusterResource;
import alluxio.util.CommonUtils;
import alluxio.util.io.PathUtils;

import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * Tests for {@link alluxio.client.file.FileSystemUtils}.
 */
public class FileSystemUtilsIntegrationTest extends BaseIntegrationTest {
  private static final int USER_QUOTA_UNIT_BYTES = 1000;
  @ClassRule
  public static LocalAlluxioClusterResource sLocalAlluxioClusterResource =
      new LocalAlluxioClusterResource.Builder()
          .setProperty(PropertyKey.USER_FILE_BUFFER_BYTES, USER_QUOTA_UNIT_BYTES)
          .build();

  public static LocalAlluxioJobCluster sJobCluster;
  private static CreateFilePOptions sWriteBoth;
  private static FileSystem sFileSystem = null;

  @Rule
  public ExpectedException mThrown = ExpectedException.none();

  @BeforeClass
  public static void beforeClass() throws Exception {
    sJobCluster = new LocalAlluxioJobCluster();
    sJobCluster.start();
    sFileSystem = sLocalAlluxioClusterResource.get().getClient();
    sWriteBoth = CreateFilePOptions.newBuilder().setWriteType(WritePType.CACHE_THROUGH)
        .setRecursive(true).build();
  }

  @Test
  public void waitCompletedTest1() throws IOException, AlluxioException, InterruptedException {
    final String uniqPath = PathUtils.uniqPath();
    final int numWrites = 4; // random value chosen through a fair dice roll :P
    final AlluxioURI uri = new AlluxioURI(uniqPath);

    final Runnable writer = new Runnable() {
      @Override
      public void run() {
        try {
          FileOutStream os = sFileSystem.createFile(uri, sWriteBoth);
          boolean completed = sFileSystem.getStatus(uri).isCompleted();
          assertFalse(completed);
          for (int i = 0; i < numWrites; i++) {
            os.write(42);
            CommonUtils.sleepMs(200);
          }
          os.close();
          completed = sFileSystem.getStatus(uri).isCompleted();
          assertTrue(completed);
        } catch (Exception e) {
          Assert.fail(e.getMessage());
        }
      }
    };

    final Runnable waiter = new Runnable() {
      @Override
      public void run() {
        try {
          boolean completed = FileSystemUtils.waitCompleted(sFileSystem, uri);
          assertTrue(completed);
          completed = sFileSystem.getStatus(uri).isCompleted();
          assertTrue(completed);
        } catch (Exception e) {
          e.printStackTrace();
          Assert.fail(e.getMessage());
        }
      }
    };

    final Thread waitingThread = new Thread(waiter);
    waitingThread.start();

    final Thread writingThread = new Thread(writer);
    writingThread.start();

    waitingThread.join();
    writingThread.join();
  }

  @Test
  public void waitCompletedTest2() throws IOException, AlluxioException, InterruptedException {
    final String uniqPath = PathUtils.uniqPath();
    final int numWrites = 4; // random value chosen through a fair dice roll :P
    final AlluxioURI uri = new AlluxioURI(uniqPath);

    final Runnable writer = new Runnable() {
      @Override
      public void run() {
        try {
          FileOutStream os = sFileSystem.createFile(uri, sWriteBoth);
          boolean completed = sFileSystem.getStatus(uri).isCompleted();
          assertFalse(completed);
          // four writes that will take > 600ms due to the sleeps
          for (int i = 0; i < numWrites; i++) {
            os.write(42);
            CommonUtils.sleepMs(200);
          }
          os.close();
          completed = sFileSystem.getStatus(uri).isCompleted();
          assertTrue(completed);
        } catch (Exception e) {
          Assert.fail(e.getMessage());
        }
      }
    };

    final Runnable waiter = new Runnable() {
      @Override
      public void run() {
        try {
          // set the slow default polling period to a more sensible value, in order
          // to speed up the tests artificial waiting times
          String original = Configuration.getString(
              PropertyKey.USER_FILE_WAITCOMPLETED_POLL_MS);
          Configuration.set(PropertyKey.USER_FILE_WAITCOMPLETED_POLL_MS, "100");
          try {
            // The write will take at most 600ms I am waiting for at most 400ms - epsilon.
            boolean completed = FileSystemUtils.waitCompleted(sFileSystem, uri, 300,
                TimeUnit.MILLISECONDS);
            assertFalse(completed);
            completed = sFileSystem.getStatus(uri).isCompleted();
            assertFalse(completed);
          } finally {
            Configuration.set(PropertyKey.USER_FILE_WAITCOMPLETED_POLL_MS, original);
          }
        } catch (Exception e) {
          e.printStackTrace();
          Assert.fail(e.getMessage());
        }
      }
    };

    final Thread waitingThread = new Thread(waiter);
    waitingThread.start();

    final Thread writingThread = new Thread(writer);
    writingThread.start();

    waitingThread.join();
    writingThread.join();
  }

  @Test
  public void waitPersistTimeoutTest() throws Exception {
    String path = PathUtils.uniqPath();
    AlluxioURI alluxioPath = new AlluxioURI(path);
    FileSystemTestUtils.createByteFile(sFileSystem, path, WritePType.MUST_CACHE, 4096);
    assertFalse("File cannot yet be persisted", sFileSystem.getStatus(alluxioPath).isPersisted());
    mThrown.expect(TimeoutException.class);
    FileSystemUtils.persistAndWait(sFileSystem, alluxioPath, 0, 1); // 1ms timeout
  }

  @Test
  public void waitPersistIndefiniteTimeoutTest() throws Exception {
    String path = PathUtils.uniqPath();
    AlluxioURI alluxioPath = new AlluxioURI(path);
    FileSystemTestUtils.createByteFile(sFileSystem, path, WritePType.MUST_CACHE, 4096);
    assertFalse("File cannot yet be persisted", sFileSystem.getStatus(alluxioPath).isPersisted());
    FileSystemUtils.persistAndWait(sFileSystem, alluxioPath, 5000, -1);
    assertTrue("File must be persisted", sFileSystem.getStatus(alluxioPath).isPersisted());
  }
}
