alluxio-master-2.x/tests/src/test/java/alluxio/server/ft/journal/JournalIntegrationTest.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.server.ft.journal;

import static org.junit.Assert.assertTrue;

import alluxio.AlluxioURI;
import alluxio.Constants;
import alluxio.client.WriteType;
import alluxio.client.file.FileSystem;
import alluxio.conf.PropertyKey;
import alluxio.exception.AlluxioException;
import alluxio.exception.FileAlreadyExistsException;
import alluxio.exception.status.CancelledException;
import alluxio.exception.status.UnavailableException;
import alluxio.master.LocalAlluxioCluster;
import alluxio.master.journal.JournalSystem;
import alluxio.master.journal.JournalType;
import alluxio.testutils.LocalAlluxioClusterResource;
import alluxio.util.CommonUtils;
import alluxio.util.WaitForOptions;

import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * Integration tests for implementations of {@link JournalSystem}.
 */
public class JournalIntegrationTest {
  private static final Logger LOG = LoggerFactory.getLogger(JournalIntegrationTest.class);

  @Rule
  public LocalAlluxioClusterResource mClusterResource =
      new LocalAlluxioClusterResource.Builder()
          .setProperty(PropertyKey.MASTER_JOURNAL_TYPE, JournalType.UFS)
          .setProperty(PropertyKey.MASTER_JOURNAL_TAILER_SHUTDOWN_QUIET_WAIT_TIME_MS, 0)
          .setProperty(PropertyKey.USER_FILE_WRITE_TYPE_DEFAULT, WriteType.MUST_CACHE)
          .setNumWorkers(0)
          .build();

  private LocalAlluxioCluster mCluster;
  private AtomicReference<Exception> mException = new AtomicReference<>();

  @Before
  public void before() {
    mCluster = mClusterResource.get();
  }

  @Test
  public void recoverAfterSingleRestart() throws Exception {
    AlluxioURI testFile = new AlluxioURI("/testFile");
    mCluster.getClient().createFile(testFile).close();
    mCluster.restartMasters();
    assertTrue(mCluster.getClient().exists(testFile));
  }

  @Test
  public void recoverAfterDoubleRestart() throws Exception {
    AlluxioURI testFile = new AlluxioURI("/testFile");
    mCluster.getClient().createFile(testFile).close();
    mCluster.restartMasters();
    mCluster.restartMasters();
    assertTrue(mCluster.getClient().exists(testFile));
  }

  @Test
  public void concurrentWriteAndRestart() throws Exception {
    final FileToucher toucher = new FileToucher();
    Thread t = new Thread(toucher);
    t.start();
    try {
      CommonUtils.waitFor("20 files to be created", () -> toucher.getFilesTouched() >= 20,
          WaitForOptions.defaults().setTimeoutMs(5 * Constants.SECOND_MS));
      mCluster.restartMasters();
    } catch (Exception e) {
      // If an exception occurred, throw it instead of the timeout exception
      checkException();
      throw e;
    } finally {
      t.interrupt();
    }
    t.join();
    checkException();
    int numFiles = mCluster.getClient().listStatus(new AlluxioURI("/")).size();
    // There may be one extra file if the client was interrupted during the last RPC.
    assertTrue(numFiles == toucher.getFilesTouched() || numFiles == toucher.getFilesTouched() + 1);
  }

  /**
   * Throws any exception raised in another thread in this class.
   */
  private void checkException() throws Exception {
    if (mException.get() != null) {
      throw mException.get();
    }
  }

  private class FileToucher implements Runnable {
    private AtomicInteger mFilesTouched = new AtomicInteger(0);

    public void run() {
      FileSystem fs;
      try {
        fs = mCluster.getClient();
      } catch (IOException e) {
        mException.set(e);
        return;
      }
      while (!Thread.interrupted()) {
        try {
          fs.createFile(new AlluxioURI("/file-" + UUID.randomUUID()));
          mFilesTouched.incrementAndGet();
        } catch (FileAlreadyExistsException e) {
          // Ignore - we could have restarted after journaling the file write.
          mFilesTouched.incrementAndGet();
        } catch (UnavailableException e) {
          // Ignore - this happen if the client is interrupted while trying to connect.
        } catch (IOException | AlluxioException e) {
          if (e.getCause() instanceof CancelledException) {
            // Ignore - this happen if the client is interrupted while trying to connect.
          } else {
            mException.set(e);
          }
        }
      }
    }

    public int getFilesTouched() {
      return mFilesTouched.get();
    }
  }
}
