alluxio-master-2.x/core/server/master/src/test/java/alluxio/master/journal/ufs/UfsJournalReaderTest.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.master.journal.ufs;

import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import alluxio.conf.Configuration;
import alluxio.master.NoopMaster;
import alluxio.master.journal.JournalReader;
import alluxio.master.journal.JournalReader.State;
import alluxio.master.journal.checkpoint.CheckpointOutputStream;
import alluxio.master.journal.checkpoint.CheckpointType;
import alluxio.proto.journal.Journal;
import alluxio.underfs.UnderFileSystem;
import alluxio.util.CommonUtils;
import alluxio.util.URIUtils;

import org.apache.commons.io.IOUtils;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;

import java.net.URI;
import java.util.Collections;

/**
 * Unit tests for {@link UfsJournalReader}.
 */
public final class UfsJournalReaderTest {
  private static final long CHECKPOINT_SIZE = 10;
  @Rule
  public TemporaryFolder mFolder = new TemporaryFolder();

  private UfsJournal mJournal;
  private UnderFileSystem mUfs;

  @Before
  public void before() throws Exception {
    URI location = URIUtils
        .appendPathOrDie(new URI(mFolder.newFolder().getAbsolutePath()), "FileSystemMaster");
    mUfs = Mockito
        .spy(UnderFileSystem.Factory.create(location.toString(), Configuration.global()));
    mJournal = new UfsJournal(location, new NoopMaster(), mUfs, 0, Collections::emptySet);
    mJournal.start();
    mJournal.gainPrimacy();
  }

  @After
  public void after() throws Exception {
    mJournal.close();
    Configuration.reloadProperties();
  }

  /**
   * Reads checkpoint.
   */
  @Test
  public void readCheckpoint() throws Exception {
    long endSN = CHECKPOINT_SIZE;
    byte[] checkpointBytes = buildCheckpoint(endSN);
    try (JournalReader reader = mJournal.getReader(true)) {
      State state;
      boolean foundCheckpoint = false;
      while ((state = reader.advance()) != State.DONE) {
        switch (state) {
          case CHECKPOINT:
            foundCheckpoint = true;
            assertArrayEquals(checkpointBytes, IOUtils.toByteArray(reader.getCheckpoint()));
            break;
          case LOG:
          case DONE:
          default:
            throw new IllegalStateException("Unexpected state: " + state);
        }
      }
      assertTrue(foundCheckpoint);
      assertEquals(endSN, reader.getNextSequenceNumber());
    }
  }

  /**
   * Reads completed logs.
   */
  @Test
  public void readCompletedLog() throws Exception {
    long fileSize = 10;
    long endSN = 10 * fileSize;
    for (long i = 0; i < endSN / fileSize; i++) {
      buildCompletedLog(i * fileSize, i * fileSize + fileSize);
    }
    try (JournalReader reader = mJournal.getReader(true)) {
      int sn = 0;
      while (reader.advance() != State.DONE) {
        assertEquals(sn, reader.getEntry().getSequenceNumber());
        sn++;
      }

      assertEquals(endSN, sn);
      assertEquals(sn, reader.getNextSequenceNumber());

      // Further advances should return State.DONE
      assertEquals(State.DONE, reader.advance());
    }
  }

  /**
   * Reads incomplete logs in a primary master.
   */
  @Test
  public void readIncompleteLogPrimary() throws Exception {
    long endSN = 10;
    buildCompletedLog(0, endSN);
    buildIncompleteLog(endSN, endSN + 1);
    try (JournalReader reader = mJournal.getReader(true)) {
      int sn = 0;
      while (reader.advance() != State.DONE) {
        assertEquals(sn, reader.getEntry().getSequenceNumber());
        sn++;
      }

      assertEquals(endSN + 1, sn);
      assertEquals(sn, reader.getNextSequenceNumber());
    }
  }

  /**
   * Secondary master cannot read incomplete logs.
   */
  @Test
  public void readIncompleteLogSecondary() throws Exception {
    long endSN = 10;
    buildCompletedLog(0, endSN);
    buildIncompleteLog(endSN, endSN + 1);
    try (JournalReader reader = mJournal.getReader(false)) {
      int sn = 0;
      while (reader.advance() != State.DONE) {
        assertEquals(sn, reader.getEntry().getSequenceNumber());
        sn++;
      }

      assertEquals(endSN, sn);
      assertEquals(sn, reader.getNextSequenceNumber());
    }
  }

  /**
   * Reads logs created while reading the journal.
   */
  @Test
  public void readNewLogs() throws Exception {
    long endSN = 10;
    buildCompletedLog(0, endSN);

    try (JournalReader reader = mJournal.getReader(true)) {
      int sn = 0;
      while (reader.advance() != State.DONE) {
        assertEquals(sn, reader.getEntry().getSequenceNumber());
        sn++;
      }

      assertEquals(endSN, sn);
      assertEquals(sn, reader.getNextSequenceNumber());

      // Write another two logs.
      buildCompletedLog(endSN, endSN * 2);
      buildIncompleteLog(endSN * 2, endSN * 2 + 1);

      while (reader.advance() != State.DONE) {
        assertEquals(sn, reader.getEntry().getSequenceNumber());
        sn++;
      }

      assertEquals(endSN * 2 + 1, sn);
      assertEquals(sn, reader.getNextSequenceNumber());
    }
  }

  /**
   * Reads checkpoint and logs. The checkpoint's last sequence number does not match any one of
   * the logs' sequence number.
   */
  @Test
  public void readCheckpointAndLogsSnNotMatch() throws Exception {
    long fileSize = 10;
    buildCheckpoint(fileSize * 3 + 1);

    for (int i = 0; i < 10; i++) {
      buildCompletedLog(i * fileSize, (i + 1) * fileSize);
    }

    try (JournalReader reader = mJournal.getReader(true)) {
      while (reader.advance() != State.DONE) {
      }
      assertEquals(10 * fileSize, reader.getNextSequenceNumber());
    }
  }

  /**
   * Reads checkpoint and logs. The checkpoint's last sequence number matches one of
   * the logs' sequence number.
   */
  @Test
  public void readCheckpointAndLogsSnMatch() throws Exception {
    long fileSize = 10;
    buildCheckpoint(fileSize * 3);

    for (int i = 0; i < 10; i++) {
      buildCompletedLog(i * fileSize, (i + 1) * fileSize);
    }

    try (JournalReader reader = mJournal.getReader(true)) {
      while (reader.advance() != State.DONE) {
      }
      assertEquals(10 * fileSize, reader.getNextSequenceNumber());
    }
  }

  /**
   * Reads checkpoint and logs from a non-zero sequence number. The given sequence number is within
   * the checkpoint.
   */
  @Test
  public void resumeReadingWithinCheckpoint() throws Exception {
    long fileSize = 10;
    buildCheckpoint(fileSize * 3);

    for (int i = 0; i < 10; i++) {
      buildCompletedLog(i * fileSize, (i + 1) * fileSize);
    }

    try (JournalReader reader = new UfsJournalReader(mJournal, fileSize * 2, true)) {
      while (reader.advance() != State.DONE) {
      }
      assertEquals(10 * fileSize, reader.getNextSequenceNumber());
    }
  }

  /**
   * Reads checkpoint and logs from a non-zero sequence number. The given sequence number is after
   * the checkpoint.
   */
  @Test
  public void resumeReadingAfterCheckpoint() throws Exception {
    long fileSize = 10;
    buildCheckpoint(fileSize * 3);

    for (int i = 0; i < 10; i++) {
      buildCompletedLog(i * fileSize, (i + 1) * fileSize);
    }

    try (JournalReader reader = new UfsJournalReader(mJournal, fileSize * 3 + 1, true)) {
      while (reader.advance() != State.DONE) {
      }
      assertEquals(10 * fileSize, reader.getNextSequenceNumber());
    }
  }

  /**
   * Builds checkpoint.
   *
   * @param sequenceNumber the sequence number after the checkpoint
   * @return the checkpoint data
   */
  private byte[] buildCheckpoint(long sequenceNumber) throws Exception {
    byte[] bytes = CommonUtils.randomAlphaNumString(10).getBytes();
    try (UfsJournalCheckpointWriter writer =
        UfsJournalCheckpointWriter.create(mJournal, sequenceNumber)) {
      CheckpointOutputStream stream =
          new CheckpointOutputStream(writer, CheckpointType.JOURNAL_ENTRY);
      stream.write(bytes);
    }
    return bytes;
  }

  /**
   * Builds complete log from the sequence number interval.
   *
   * @param start start of the sequence number (included)
   * @param end end of the sequence number (excluded)
   */
  private void buildCompletedLog(long start, long end) throws Exception {
    Mockito.when(mUfs.supportsFlush()).thenReturn(true);
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, start);
    for (long i = start; i < end; i++) {
      writer.write(newEntry(i));
    }
    writer.close();
  }

  /**
   * Builds incomplete log.
   *
   * @param start start of the sequence number (included)
   * @param end end of the sequence number (excluded)
   */
  private void buildIncompleteLog(long start, long end) throws Exception {
    Mockito.when(mUfs.supportsFlush()).thenReturn(true);
    buildCompletedLog(start, end);
    assertTrue(
        mUfs.renameFile(UfsJournalFile.encodeLogFileLocation(mJournal, start, end).toString(),
            UfsJournalFile
                .encodeLogFileLocation(mJournal, start, UfsJournal.UNKNOWN_SEQUENCE_NUMBER)
                .toString()));
  }

  /**
   * Creates a dummy journal entry with the given sequence number.
   *
   * @param sequenceNumber the sequence number
   * @return the journal entry
   */
  private Journal.JournalEntry newEntry(long sequenceNumber) {
    return Journal.JournalEntry.newBuilder().setSequenceNumber(sequenceNumber).build();
  }
}
