alluxio-master-2.x/tests/src/test/java/alluxio/job/plan/persist/PersistIntegrationTest.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.job.plan.persist;

import static alluxio.job.wire.Status.COMPLETED;
import static alluxio.job.wire.Status.FAILED;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.FileSystemMasterClient;
import alluxio.client.file.URIStatus;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.WritePType;
import alluxio.job.JobIntegrationTest;
import alluxio.job.wire.JobInfo;
import alluxio.master.file.meta.PersistenceState;
import alluxio.master.job.JobMaster;
import alluxio.resource.CloseableResource;
import alluxio.security.authorization.Mode;
import alluxio.testutils.LocalAlluxioClusterResource;
import alluxio.underfs.UnderFileSystem;
import alluxio.util.CommonUtils;
import alluxio.util.FileSystemOptionsUtils;
import alluxio.util.WaitForOptions;
import alluxio.util.io.PathUtils;

import org.junit.Assert;
import org.junit.Test;

/**
 * Integration tests for {@link PersistDefinition}.
 */
public final class PersistIntegrationTest extends JobIntegrationTest {
  private static final String TEST_URI = "/test";
  private static final Mode TEST_MODE = new Mode((short) 0777);

  /**
   * Tests persisting a file.
   */
  @Test
  public void persistTest() throws Exception {
    // write a file in alluxio only
    AlluxioURI filePath = new AlluxioURI(TEST_URI);
    FileOutStream os = mFileSystem.createFile(filePath,
        CreateFilePOptions.newBuilder().setWriteType(WritePType.MUST_CACHE).build());
    os.write((byte) 0);
    os.write((byte) 1);
    os.close();

    // check the file is completed but not persisted
    URIStatus status = mFileSystem.getStatus(filePath);
    Assert.assertEquals(PersistenceState.NOT_PERSISTED.toString(), status.getPersistenceState());
    Assert.assertTrue(status.isCompleted());

    // run the persist job and check that it succeeds
    waitForJobToFinish(mJobMaster.run(new PersistConfig("/test", 1, true, status.getUfsPath())));
    String ufsPath = status.getUfsPath();
    UnderFileSystem ufs = UnderFileSystem.Factory.create(ufsPath, Configuration.global());
    Assert.assertTrue(ufs.exists(ufsPath));

    // run the persist job again with the overwrite flag and check that it succeeds
    waitForJobToFinish(mJobMaster.run(new PersistConfig("/test", 1, true, status.getUfsPath())));
    Assert.assertTrue(ufs.exists(ufsPath));

    // run the persist job again without the overwrite flag and check it fails
    final long jobId = mJobMaster.run(new PersistConfig("/test", 1, false, status.getUfsPath()));
    waitForJobFailure(jobId);
  }

  /**
   * Tests persisting a file.
   */
  @Test
  public void persistWithAccessTimeUnchangedTest() throws Exception {
    // write a file in alluxio only
    AlluxioURI filePath = new AlluxioURI(TEST_URI);
    FileOutStream os = mFileSystem.createFile(filePath,
        CreateFilePOptions.newBuilder().setWriteType(WritePType.MUST_CACHE).build());
    os.write((byte) 0);
    os.write((byte) 1);
    os.close();

    // check the file is completed but not persisted
    URIStatus status = mFileSystem.getStatus(filePath);
    Assert.assertEquals(PersistenceState.NOT_PERSISTED.toString(), status.getPersistenceState());
    Assert.assertTrue(status.isCompleted());

    // run the persist job and check that it succeeds
    waitForJobToFinish(mJobMaster.run(new PersistConfig("/test", 1, true, status.getUfsPath())));
    String ufsPath = status.getUfsPath();
    UnderFileSystem ufs = UnderFileSystem.Factory.create(ufsPath, Configuration.global());
    Assert.assertTrue(ufs.exists(ufsPath));

    // check file access time is not changed
    URIStatus newStatus = mFileSystem.getStatus(filePath);
    Assert.assertEquals(newStatus.getLastAccessTimeMs(), status.getLastAccessTimeMs());
  }

  @Test
  @LocalAlluxioClusterResource.Config(
      confParams = {
          PropertyKey.Name.MASTER_PERSISTENCE_MAX_TOTAL_WAIT_TIME_MS, "1ms",
          PropertyKey.Name.MASTER_PERSISTENCE_CHECKER_INTERVAL_MS, "50ms",
          PropertyKey.Name.MASTER_PERSISTENCE_SCHEDULER_INTERVAL_MS, "50ms"})
  public void persistTimeoutTest() throws Exception {
    // write a file in alluxio only
    AlluxioURI filePath = new AlluxioURI(TEST_URI);
    FileOutStream os = mFileSystem.createFile(filePath,
        CreateFilePOptions.newBuilder().setWriteType(WritePType.MUST_CACHE).build());
    os.write((byte) 0);
    os.write((byte) 1);
    os.close();
    // check the file is completed but not persisted
    URIStatus status = mFileSystem.getStatus(filePath);
    Assert.assertEquals(PersistenceState.NOT_PERSISTED.toString(), status.getPersistenceState());
    Assert.assertTrue(status.isCompleted());
    // kill job worker
    mLocalAlluxioJobCluster.getWorker().stop();
    // persist the file
    try (CloseableResource<FileSystemMasterClient> client =
        mFsContext.acquireMasterClientResource()) {
      client.get().scheduleAsyncPersist(new AlluxioURI(TEST_URI),
          FileSystemOptionsUtils.scheduleAsyncPersistDefaults(Configuration.global()));
    }
    CommonUtils.waitFor("persist timeout", () -> {
      try {
        return PersistenceState.NOT_PERSISTED.toString().equals(
            mFileSystem.getStatus(filePath).getPersistenceState());
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, WaitForOptions.defaults().setTimeoutMs(10000));
    // verify timeout and reverted to not persisted
    status = mFileSystem.getStatus(filePath);
    Assert.assertEquals(PersistenceState.NOT_PERSISTED.toString(), status.getPersistenceState());
    // restart master
    mLocalAlluxioClusterResource.get().restartMasters();
    mFileSystem = mLocalAlluxioClusterResource.get().getClient(); // need new client after restart
    // verify not persisted
    status = mFileSystem.getStatus(filePath);
    Assert.assertEquals(PersistenceState.NOT_PERSISTED.toString(), status.getPersistenceState());
    String ufsPath = status.getUfsPath();
    UnderFileSystem ufs = UnderFileSystem.Factory.create(ufsPath, Configuration.global());
    Assert.assertFalse(ufs.exists(ufsPath));
  }

  @Test
  @LocalAlluxioClusterResource.Config(
      confParams = {PropertyKey.Name.MASTER_PERSISTENCE_INITIAL_INTERVAL_MS, "10s"})
  public void disallowIncompletePersist() throws Exception {
    AlluxioURI path = new AlluxioURI("/" + CommonUtils.randomAlphaNumString(10));

    // Create file, but do not complete
    FileOutStream os = mFileSystem.createFile(path,
        CreateFilePOptions.newBuilder().setWriteType(WritePType.MUST_CACHE)
            .setMode(TEST_MODE.toProto()).build());

    // schedule an async persist
    try (CloseableResource<FileSystemMasterClient> client =
        mFsContext.acquireMasterClientResource()) {
      client.get().scheduleAsyncPersist(path,
          FileSystemOptionsUtils.scheduleAsyncPersistDefaults(Configuration.global()));
      Assert.fail("Should not be able to schedule persistence for incomplete file");
    } catch (Exception e) {
      // expected
      Assert.assertTrue("Failure expected to be about incomplete files",
          e.getMessage().toLowerCase().contains("incomplete"));
    }
  }

  @Test(timeout = 30000)
  public void persistOnlyCompleteFiles() throws Exception {
    AlluxioURI path = new AlluxioURI("/" + CommonUtils.randomAlphaNumString(10));

    // Create file, but do not complete
    FileOutStream os = mFileSystem.createFile(path,
        CreateFilePOptions.newBuilder().setWriteType(WritePType.MUST_CACHE)
            .setMode(TEST_MODE.toProto()).build());
    URIStatus status = mFileSystem.getStatus(path);

    // Generate a temporary path to be used by the persist job.
    String tempUfsPath =
        PathUtils.temporaryFileName(System.currentTimeMillis(), status.getUfsPath());
    JobMaster jobMaster = mLocalAlluxioJobCluster.getMaster().getJobMaster();
    // Run persist job on incomplete file (expected to fail)
    long failId =
        jobMaster.run(new PersistConfig(path.toString(), status.getMountId(), false, tempUfsPath));

    CommonUtils.waitFor("Wait for persist job to complete", () -> {
      try {
        JobInfo jobInfo = jobMaster.getStatus(failId);
        Assert.assertNotEquals("Persist should not succeed for incomplete file", COMPLETED,
            jobInfo.getStatus());
        if (jobInfo.getStatus() == FAILED) {
          // failed job is expected
          Assert.assertTrue("Failure expected to be about incomplete files",
              jobInfo.getErrorMessage().toLowerCase().contains("incomplete"));
          return true;
        }
        return false;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, WaitForOptions.defaults().setTimeoutMs(10 * Constants.SECOND_MS)
        .setInterval(100));

    // close the file to allow persist to happen
    os.close();

    // Run persist job on complete file (expected to succeed)
    long successId =
        jobMaster.run(new PersistConfig(path.toString(), status.getMountId(), false, tempUfsPath));

    CommonUtils.waitFor("Wait for persist job to complete", () -> {
      try {
        JobInfo jobInfo = jobMaster.getStatus(successId);
        Assert.assertNotEquals("Persist should not fail", FAILED, jobInfo.getStatus());
        return jobInfo.getStatus() == COMPLETED;
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }, WaitForOptions.defaults().setTimeoutMs(10 * Constants.SECOND_MS)
        .setInterval(100));
  }
}
