alluxio-master-2.x/tests/src/test/java/alluxio/client/fs/io/FileOutStreamIntegrationTest.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.io;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.client.WriteType;
import alluxio.client.file.FileOutStream;
import alluxio.client.file.URIStatus;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.grpc.CreateDirectoryPOptions;
import alluxio.grpc.CreateFilePOptions;
import alluxio.grpc.WritePType;
import alluxio.master.file.FileSystemMaster;
import alluxio.testutils.LocalAlluxioClusterResource;
import alluxio.underfs.UnderFileSystem;
import alluxio.util.CommonUtils;
import alluxio.util.io.BufferUtils;
import alluxio.util.io.PathUtils;
import alluxio.wire.FileBlockInfo;
import alluxio.wire.WorkerInfo;

import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameter;
import org.junit.runners.Parameterized.Parameters;

import java.util.List;

/**
 * Integration tests for {@link alluxio.client.file.FileOutStream}, parameterized by the write
 * types.
 */
@RunWith(Parameterized.class)
public final class FileOutStreamIntegrationTest extends AbstractFileOutStreamIntegrationTest {
  // TODO(binfan): Run tests with local writes enabled and disabled.

  @Parameters
  public static Object[] dataFileInStreamIntegrationTest() {
    return new Object[] {
        WriteType.ASYNC_THROUGH,
        WriteType.CACHE_THROUGH,
        WriteType.MUST_CACHE,
        WriteType.THROUGH,
    };
  }

  @Parameter
  public WriteType mWriteType;

  /**
   * Tests {@link FileOutStream#write(int)}.
   */
  @Test
  public void writeBytes() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    for (int len = MIN_LEN; len <= MAX_LEN; len += DELTA) {
      CreateFilePOptions op = CreateFilePOptions.newBuilder().setWriteType(mWriteType.toProto())
          .setRecursive(true).build();
      AlluxioURI filePath =
          new AlluxioURI(PathUtils.concatPath(uniqPath, "file_" + len + "_" + mWriteType));
      writeIncreasingBytesToFile(filePath, len, op);
      if (mWriteType.getAlluxioStorageType().isStore()) {
        checkFileInAlluxio(filePath, len);
      }
      if (mWriteType.getUnderStorageType().isSyncPersist()) {
        checkFileInUnderStorage(filePath, len);
      }
    }
  }

  /**
   * Tests {@link FileOutStream#write(int)}.
   */
  @Test
  public void writeInNonExistDirectory() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    CreateFilePOptions op = CreateFilePOptions.newBuilder().setWriteType(WritePType.CACHE_THROUGH)
        .setRecursive(true).build();
    AlluxioURI filePath =
        new AlluxioURI(PathUtils.concatPath(uniqPath, "file_" + MIN_LEN + "_" + mWriteType));
    AlluxioURI parentPath = new AlluxioURI(uniqPath);

    // create a directory without a backing directory in UFS
    mFileSystem.createDirectory(parentPath, CreateDirectoryPOptions.newBuilder()
        .setRecursive(true)
        .setWriteType(WritePType.CACHE_THROUGH)
        .build());
    URIStatus status = mFileSystem.getStatus(parentPath);
    String checkpointPath = status.getUfsPath();
    UnderFileSystem ufs = UnderFileSystem.Factory.create(checkpointPath,
        Configuration.global());
    ufs.deleteDirectory(checkpointPath);

    // write a file to a directory exists in Alluxio but not in UFS
    writeIncreasingBytesToFile(filePath, MIN_LEN, op);
    checkFileInAlluxio(filePath, MIN_LEN);
    checkFileInUnderStorage(filePath, MIN_LEN);
  }

  /**
   * Tests {@link FileOutStream#write(byte[])}.
   */
  @Test
  public void writeByteArray() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    for (int len = MIN_LEN; len <= MAX_LEN; len += DELTA) {
      CreateFilePOptions op = CreateFilePOptions.newBuilder().setWriteType(mWriteType.toProto())
          .setRecursive(true).build();
      AlluxioURI filePath =
          new AlluxioURI(PathUtils.concatPath(uniqPath, "file_" + len + "_" + mWriteType));
      writeIncreasingByteArrayToFile(filePath, len, op);
      if (mWriteType.getAlluxioStorageType().isStore()) {
        checkFileInAlluxio(filePath, len);
      }
      if (mWriteType.getUnderStorageType().isSyncPersist()) {
        checkFileInUnderStorage(filePath, len);
      }
    }
  }

  /**
   * Tests {@link FileOutStream#write(byte[], int, int)}.
   */
  @Test
  public void writeTwoByteArrays() throws Exception {
    String uniqPath = PathUtils.uniqPath();
    for (int len = MIN_LEN; len <= MAX_LEN; len += DELTA) {
      CreateFilePOptions op = CreateFilePOptions.newBuilder().setWriteType(mWriteType.toProto())
          .setRecursive(true).build();
      AlluxioURI filePath =
          new AlluxioURI(PathUtils.concatPath(uniqPath, "file_" + len + "_" + mWriteType));
      writeTwoIncreasingByteArraysToFile(filePath, len, op);
      if (mWriteType.getAlluxioStorageType().isStore()) {
        checkFileInAlluxio(filePath, len);
      }
      if (mWriteType.getUnderStorageType().isSyncPersist()) {
        checkFileInUnderStorage(filePath, len);
      }
    }
  }

  /**
   * Tests writing to a file and specify the location to be localhost.
   */
  @Test
  @LocalAlluxioClusterResource.Config(confParams = {
      PropertyKey.Name.USER_BLOCK_WRITE_LOCATION_POLICY,
      "alluxio.client.block.policy.LocalFirstPolicy"
      })
  public void writeSpecifyLocal() throws Exception {
    AlluxioURI filePath = new AlluxioURI(PathUtils.uniqPath());
    final int length = 2;
    CreateFilePOptions op = CreateFilePOptions.newBuilder().setWriteType(mWriteType.toProto())
        .setRecursive(true).build();
    try (FileOutStream os = mFileSystem.createFile(filePath, op)) {
      os.write((byte) 0);
      os.write((byte) 1);
    }
    if (mWriteType.getAlluxioStorageType().isStore()) {
      checkFileInAlluxio(filePath, length);
    }
    if (mWriteType.getUnderStorageType().isSyncPersist()) {
      checkFileInUnderStorage(filePath, length);
    }
  }

  /**
   * Tests writing to a file for longer than HEARTBEAT_INTERVAL_MS to make sure the sessionId
   * doesn't change. Tracks [ALLUXIO-171].
   */
  @Test
  public void longWrite() throws Exception {
    AlluxioURI filePath = new AlluxioURI(PathUtils.uniqPath());
    final int length = 2;
    try (FileOutStream os = mFileSystem.createFile(filePath, CreateFilePOptions.newBuilder()
        .setWriteType(mWriteType.toProto()).setRecursive(true).build())) {
      os.write((byte) 0);
      Thread.sleep(Constants.SECOND_MS * 2);
      os.write((byte) 1);
    }
    if (mWriteType.getAlluxioStorageType().isStore()) {
      checkFileInAlluxio(filePath, length);
    }
    if (mWriteType.getUnderStorageType().isSyncPersist()) {
      checkFileInUnderStorage(filePath, length);
    }
  }

  /**
   * Tests if out-of-order writes are possible. Writes could be out-of-order when the following are
   * both true: - a "large" write (over half the internal buffer size) follows a smaller write. -
   * the "large" write does not cause the internal buffer to overflow.
   */
  @Test
  public void outOfOrderWrite() throws Exception {
    AlluxioURI filePath = new AlluxioURI(PathUtils.uniqPath());
    // A length greater than 0.5 * BUFFER_BYTES and less than BUFFER_BYTES.
    int length = (BUFFER_BYTES * 3) / 4;
    try (FileOutStream os = mFileSystem.createFile(filePath, CreateFilePOptions.newBuilder()
        .setWriteType(mWriteType.toProto()).setRecursive(true).build())) {
      // Write something small, so it is written into the buffer, and not directly to the file.
      os.write((byte) 0);
      // Write a large amount of data (larger than BUFFER_BYTES/2, but will not overflow the buffer.
      os.write(BufferUtils.getIncreasingByteArray(1, length));
    }
    if (mWriteType.getAlluxioStorageType().isStore()) {
      checkFileInAlluxio(filePath, length + 1);
    }
    if (mWriteType.getUnderStorageType().isSyncPersist()) {
      checkFileInUnderStorage(filePath, length + 1);
    }
  }

  /**
   * Tests canceling after multiple blocks have been written correctly cleans up temporary worker
   * resources.
   */
  @LocalAlluxioClusterResource.Config(
      confParams = {PropertyKey.Name.MASTER_LOST_WORKER_FILE_DETECTION_INTERVAL, "250ms",
          PropertyKey.Name.WORKER_BLOCK_HEARTBEAT_INTERVAL_MS, "250ms"})
  @Test
  public void cancelWrite() throws Exception {
    AlluxioURI path = new AlluxioURI(PathUtils.uniqPath());
    try (FileOutStream os = mFileSystem.createFile(path, CreateFilePOptions.newBuilder()
        .setWriteType(mWriteType.toProto()).setRecursive(true).build())) {
      os.write(BufferUtils.getIncreasingByteArray(0, BLOCK_SIZE_BYTES * 3 + 1));
      os.cancel();
    }
    long gracePeriod = Configuration
        .getMs(PropertyKey.WORKER_BLOCK_HEARTBEAT_INTERVAL_MS) * 2;
    CommonUtils.sleepMs(gracePeriod);
    List<WorkerInfo> workers =
        mLocalAlluxioClusterResource.get().getLocalAlluxioMaster().getMasterProcess()
            .getMaster(FileSystemMaster.class).getFileSystemMasterView().getWorkerInfoList();
    for (WorkerInfo worker : workers) {
      Assert.assertEquals(0, worker.getUsedBytes());
    }
  }

  @Test
  public void getStatusBeforeClose() throws Exception {
    AlluxioURI path = new AlluxioURI(PathUtils.uniqPath());
    try (FileOutStream os = mFileSystem.createFile(path, CreateFilePOptions.newBuilder()
            .setWriteType(mWriteType.toProto()).setRecursive(true).build())) {
      for (int i = 0; i < 3; i++) {
        os.write(BufferUtils.getIncreasingByteArray(i * BLOCK_SIZE_BYTES, BLOCK_SIZE_BYTES));
        // Fetch file status when the stream is still open
        URIStatus status = mFileSystem.getStatus(path);
        if (!mWriteType.isThrough()) {
          // When the writeType is THROUGH, we only see the blocks when the file is committed
          // When the file is not committed, we only see incomplete FileBlockInfo
          Assert.assertEquals(i + 1, status.getBlockIds().size());
        }
      }
      os.write(BufferUtils.getIncreasingByteArray(3 * BLOCK_SIZE_BYTES, 1));
    }
    URIStatus finalStatus = mFileSystem.getStatus(path);
    Assert.assertEquals(4, finalStatus.getBlockIds().size());
    List<FileBlockInfo> fileBlocks = finalStatus.getFileBlockInfos();
    Assert.assertEquals(4, fileBlocks.size());
    for (int i = 0; i < 3; i++) {
      Assert.assertEquals(BLOCK_SIZE_BYTES, fileBlocks.get(i).getBlockInfo().getLength());
    }
    Assert.assertEquals(1, fileBlocks.get(3).getBlockInfo().getLength());
  }
}
