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

import alluxio.RuntimeConstants;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.ExceptionMessage;
import alluxio.master.NoopMaster;
import alluxio.master.journal.JournalReader;
import alluxio.master.journal.JournalReader.State;
import alluxio.proto.journal.Journal;
import alluxio.underfs.UnderFileSystem;
import alluxio.util.URIUtils;

import org.hamcrest.core.StringStartsWith;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.rules.TemporaryFolder;
import org.mockito.Mockito;
import org.powermock.reflect.Whitebox;

import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URI;
import java.nio.channels.FileChannel;
import java.util.Collections;

/**
 * Unit tests for {@link UfsJournalLogWriter}.
 */
public final class UfsJournalLogWriterTest {
  private static final String INJECTED_IO_ERROR_MESSAGE = "injected I/O error";

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

  @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();
  }

  /**
   * A new journal writer completes the current log.
   */
  @Test
  public void completeCurrentLog() throws Exception {
    long startSN = 0x10;
    long endSN = 0x20;
    mJournal.getUfs().create(
        UfsJournalFile.encodeLogFileLocation(mJournal, startSN, UfsJournal.UNKNOWN_SEQUENCE_NUMBER)
            .toString()).close();
    new UfsJournalLogWriter(mJournal, 0x20).close();
    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);

    String expectedLog =
        URIUtils.appendPathOrDie(mJournal.getLogDir(), String.format("0x%x-0x%x", startSN, endSN))
            .toString();
    Assert.assertEquals(1, snapshot.getLogs().size());
    Assert.assertEquals(expectedLog, snapshot.getLogs().get(0).getLocation().toString());
    Assert.assertTrue(UfsJournalSnapshot.getCurrentLog(mJournal) == null);
  }

  /**
   * The completed log corresponds to the current log exists.
   */
  @Test
  public void duplicateCompletedLog() throws Exception {
    long startSN = 0x10;
    long endSN = 0x20;
    mJournal.getUfs().create(
        UfsJournalFile.encodeLogFileLocation(mJournal, startSN, UfsJournal.UNKNOWN_SEQUENCE_NUMBER)
            .toString()).close();
    mJournal.getUfs()
        .create(UfsJournalFile.encodeLogFileLocation(mJournal, startSN, endSN).toString()).close();
    new UfsJournalLogWriter(mJournal, startSN).close();
    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);

    String expectedLog =
        URIUtils.appendPathOrDie(mJournal.getLogDir(), String.format("0x%x-0x%x", startSN, endSN))
            .toString();
    Assert.assertEquals(1, snapshot.getLogs().size());
    Assert.assertEquals(expectedLog, snapshot.getLogs().get(0).getLocation().toString());
    Assert.assertTrue(UfsJournalSnapshot.getCurrentLog(mJournal) == null);
  }

  /**
   * Writes journal entries while the UFS has flush supported.
   */
  @Test
  public void writeJournalEntryUfsHasFlush() throws Exception {
    Mockito.when(mUfs.supportsFlush()).thenReturn(true);
    long nextSN = 0x20;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, nextSN);

    for (int i = 0; i < 10; i++) {
      writer.write(newEntry(nextSN));
      nextSN++;
      if (i % 5 == 0) {
        writer.flush();
      }
    }
    writer.close();

    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
    Assert.assertTrue(snapshot.getCheckpoints().isEmpty());
    // 0x20 - 0x2a
    Assert.assertEquals(1, snapshot.getLogs().size());
    Assert.assertEquals(UfsJournalFile.encodeLogFileLocation(mJournal, 0x20, 0x2a),
        snapshot.getLogs().get(0).getLocation());
  }

  /**
   * Writes journal entries while the UFS has no flush supported.
   */
  @Test
  public void writeJournalEntryUfsNoFlush() throws Exception {
    Mockito.when(mUfs.supportsFlush()).thenReturn(false);
    long nextSN = 0x20;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, nextSN);
    for (int i = 0; i < 10; i++) {
      writer.write(newEntry(nextSN));
      nextSN++;
      if (i % 5 == 0) {
        writer.flush();
      }
    }
    writer.close();

    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
    Assert.assertTrue(snapshot.getCheckpoints().isEmpty());
    // 0x20 - 0x21, 0x21 - 0x26, 0x26 - 0x2a
    Assert.assertEquals(3, snapshot.getLogs().size());
    Assert.assertEquals(UfsJournalFile.encodeLogFileLocation(mJournal, 0x20, 0x21),
        snapshot.getLogs().get(0).getLocation());
    Assert.assertEquals(UfsJournalFile.encodeLogFileLocation(mJournal, 0x21, 0x26),
        snapshot.getLogs().get(1).getLocation());
    Assert.assertEquals(UfsJournalFile.encodeLogFileLocation(mJournal, 0x26, 0x2a),
        snapshot.getLogs().get(2).getLocation());
  }

  /**
   * Tests journal rotation.
   */
  @Test
  public void writeJournalEntryRotate() throws Exception {
    Mockito.when(mUfs.supportsFlush()).thenReturn(true);
    Configuration.set(PropertyKey.MASTER_JOURNAL_LOG_SIZE_BYTES_MAX, "1");

    long nextSN = 0x20;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, nextSN);
    for (int i = 0; i < 10; i++) {
      writer.write(newEntry(nextSN));
      nextSN++;
      writer.flush();
    }
    writer.close();

    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
    Assert.assertTrue(snapshot.getCheckpoints().isEmpty());
    Assert.assertEquals(10, snapshot.getLogs().size());
    for (int i = 0; i < 10; i++) {
      Assert.assertEquals(UfsJournalFile.encodeLogFileLocation(mJournal, 0x20 + i, 0x21 + i),
          snapshot.getLogs().get(i).getLocation());
    }
  }

  /**
   * Tests that (1) {@link UfsJournalLogWriter#mJournalOutputStream} is reset when an exception
   * is thrown, and (2) the {@link UfsJournalLogWriter} recovers during the next write. It should
   * write out all entries, including the one that initially failed.
   */
  @Test
  public void recoverFromUfsFailure() throws Exception {
    long startSN = 0x10;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, startSN);
    final int numberOfEntriesWrittenBeforeFailure = 10;
    long nextSN = writeJournalEntries(writer, startSN, numberOfEntriesWrittenBeforeFailure);
    DataOutputStream badOut = createMockDataOutputStream(writer);
    Mockito.doThrow(new IOException(INJECTED_IO_ERROR_MESSAGE)).when(badOut)
        .write(Mockito.any(byte[].class), Mockito.anyInt(), Mockito.anyInt());
    tryWriteAndExpectToFail(writer, nextSN);
    // UfsJournalLogWriter will perform recovery before the write operation.
    writer.write(newEntry(nextSN));
    nextSN++;
    writer.close();

    checkJournalEntries(startSN, nextSN);
  }

  /**
   * Tests that {@link UfsJournalLogWriter#flush()} can recover after Ufs failure.
   *
   * This test has the following steps.
   * 1. Write several journal entries, flush and close the journal. This will create a complete
   *    journal file, i.e. &lt;startSN&gt;-&lt;endSN&gt;.
   * 2. Write another journal entry, do NOT flush it, and inject an I/O error.
   * 3. Attempt to write another journal entry, which is expected to fail. After the failure,
   *    the newly created incomplete journal file may or may not have valid content.
   * 4. The UFS does not guarantee the consistency of the incomplete journal. To
   *    test how {@link UfsJournalLogWriter} recovers from Ufs failure before
   *    {@link UfsJournalLogWriter#flush()}, we simply delete the possibly created incomplete
   *    journal.
   */
  @Test
  public void flushAfterUfsFailure() throws Exception {
    Mockito.when(mUfs.supportsFlush()).thenReturn(true);

    // Write several journal entries, creating and closing journal file.
    // This file is complete.
    long startSN = 0x10;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, startSN);
    final int numberOfEntriesWrittenBeforeFailure = 10;
    long nextSN = writeJournalEntries(writer, startSN, numberOfEntriesWrittenBeforeFailure);
    writer.flush();
    writer.close();

    // Write another entry without flushing.
    writer = new UfsJournalLogWriter(mJournal, nextSN);
    nextSN = writeJournalEntries(writer, nextSN, 1);
    DataOutputStream badOut = createMockDataOutputStream(writer);
    Mockito.doThrow(new IOException(INJECTED_IO_ERROR_MESSAGE)).when(badOut)
        .write(Mockito.any(byte[].class), Mockito.anyInt(), Mockito.anyInt());
    tryWriteAndExpectToFail(writer, nextSN);

    // Delete the incomplete journal file to simulate the case in which
    // journal entries that have not been explicitly flushed will be lost.
    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
    UfsJournalFile journalFile = snapshot.getCurrentLog(mJournal);
    File file = new File(journalFile.getLocation().toString());
    file.delete();

    // Flush the journal, recovering from the Ufs failure.
    writer.flush();
    checkJournalEntries(startSN, nextSN);
    writer.close();
  }

  /**
   * Test that failures during {@link UfsJournalLogWriter#flush()} will complete the current file.
   */
  @Test
  public void flushFailureCompletesFile() throws Exception {
    Mockito.when(mUfs.supportsFlush()).thenReturn(true);

    // Write several journal entries, creating and closing journal file.
    // This file is complete.
    long startSN = 0x10;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, startSN);
    final int numberOfEntriesWrittenBeforeFailure = 10;
    long nextSN = writeJournalEntries(writer, startSN, numberOfEntriesWrittenBeforeFailure);
    writer.flush();
    writer.close();

    // Write another entry and then fail the flush.
    writer = new UfsJournalLogWriter(mJournal, nextSN);
    nextSN = writeJournalEntries(writer, nextSN, 1);
    DataOutputStream badOut = createMockDataOutputStream(writer);
    Mockito.doThrow(new IOException(INJECTED_IO_ERROR_MESSAGE)).when(badOut).flush();
    nextSN = writeJournalEntries(writer, nextSN, 1);
    tryFlushAndExpectToFail(writer);

    // Retry the flush, expect it to rotate the log and start a new file
    writer.flush();
    // Complete the last log
    writer.close();

    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
    Assert.assertEquals(3, snapshot.getLogs().size());
    Assert.assertEquals(nextSN - 1, snapshot.getLogs().get(2).getStart());
    Assert.assertEquals(nextSN, snapshot.getLogs().get(2).getEnd());
  }

  /**
   * Tests that {@link UfsJournalLogWriter} can detect the failure in which some flushed journal
   * entries are missing from the journal during recovery.
   */
  @Test
  public void missingJournalEntries() throws Exception {
    long startSN = 0x10;
    long nextSN = startSN;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, nextSN);
    long truncateSize = 0;
    long firstCorruptedEntrySeq = startSN + 4;
    for (int i = 0; i < 5; i++) {
      writer.write(newEntry(nextSN));
      nextSN++;
      if (i == 3) {
        writer.flush();
        UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
        UfsJournalFile journalFile = snapshot.getCurrentLog(mJournal);
        File file = new File(journalFile.getLocation().toString());
        truncateSize = file.length();
      }
    }
    writer.flush();

    writer.write(newEntry(nextSN));
    long seqOfFirstEntryToFlush = nextSN;
    nextSN++;
    Assert.assertNotNull(writer.getJournalOutputStream());
    DataOutputStream badOut = createMockDataOutputStream(writer);
    Mockito.doThrow(new IOException(INJECTED_IO_ERROR_MESSAGE)).when(badOut)
        .write(Mockito.any(byte[].class), Mockito.anyInt(), Mockito.anyInt());
    tryWriteAndExpectToFail(writer, nextSN);
    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
    UfsJournalFile journalFile = snapshot.getCurrentLog(mJournal);
    File file = new File(journalFile.getLocation().toString());
    try (FileOutputStream fileOutputStream = new FileOutputStream(file, true);
        FileChannel fileChannel = fileOutputStream.getChannel()) {
      fileChannel.truncate(truncateSize);
    }
    mThrown.expect(RuntimeException.class);
    mThrown.expectMessage(
        ExceptionMessage.JOURNAL_ENTRY_MISSING.getMessageWithUrl(
            RuntimeConstants.ALLUXIO_DEBUG_DOCS_URL,
            firstCorruptedEntrySeq, seqOfFirstEntryToFlush));
    writer.write(newEntry(nextSN));
    writer.close();
  }

  @Test
  public void recoverWithNoJournalFiles() throws Exception {
    long startSN = 0x10;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, startSN);
    // Put stream into a bad state
    long nextSN = writeJournalEntries(writer, startSN, 5);
    DataOutputStream badOut = createMockDataOutputStream(writer);
    Mockito.doThrow(new IOException(INJECTED_IO_ERROR_MESSAGE)).when(badOut).flush();
    tryFlushAndExpectToFail(writer);

    // Delete the file
    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
    UfsJournalFile journalFile = snapshot.getLogs().get(0);
    File file = new File(journalFile.getLocation().toString());
    file.delete();

    // Recover should fail since we deleted the file
    mThrown.expect(RuntimeException.class);
    mThrown.expectMessage(
        StringStartsWith.startsWith("Cannot find any journal entry to recover."));
    writer.write(newEntry(nextSN));
    writer.close();
  }

  @Test
  public void recoverMissingJournalFiles() throws Exception {
    long startSN = 0x10;
    UfsJournalLogWriter writer = new UfsJournalLogWriter(mJournal, startSN);
    // Create a file
    long nextSN = writeJournalEntries(writer, startSN, 5);
    writer.close();

    // Flush some entries to the next file
    writer = new UfsJournalLogWriter(mJournal, nextSN);
    nextSN = writeJournalEntries(writer, nextSN, 5);
    writer.flush();

    // Put the stream into a bad state
    nextSN = writeJournalEntries(writer, nextSN, 5);
    DataOutputStream badOut = createMockDataOutputStream(writer);
    Mockito.doThrow(new IOException(INJECTED_IO_ERROR_MESSAGE)).when(badOut).flush();
    tryFlushAndExpectToFail(writer);

    // Delete the current log
    UfsJournalSnapshot snapshot = UfsJournalSnapshot.getSnapshot(mJournal);
    UfsJournalFile journalFile = snapshot.getCurrentLog(mJournal);
    File file = new File(journalFile.getLocation().toString());
    file.delete();

    // Recover should fail since we deleted the current log
    mThrown.expect(RuntimeException.class);
    mThrown.expectMessage(
        ExceptionMessage.JOURNAL_ENTRY_MISSING.getMessageWithUrl(
            RuntimeConstants.ALLUXIO_DEBUG_DOCS_URL, 0x15, 0x1A));
    writer.write(newEntry(nextSN));
    writer.close();
  }

  /**
   * Writes {@code numberOfEntries} journal entries starting from {@code startSN}.
   *
   * @param writer {@link UfsJournalLogWriter} instance to use
   * @param startSN start journal sequence number
   * @param numberOfEntries number of journal entries
   * @return next journal entry sequence number after writing these entries
   */
  private long writeJournalEntries(UfsJournalLogWriter writer, long startSN,
      int numberOfEntries) throws Exception {
    long nextSN = startSN;
    for (int i = 0; i < numberOfEntries; i++) {
      writer.write(newEntry(nextSN));
      nextSN++;
    }
    Assert.assertNotNull(writer.getJournalOutputStream());
    return nextSN;
  }

  /**
   * Creates a mock {@link DataOutputStream} for {@link UfsJournalLogWriter#mJournalOutputStream}.
   * Mock's internal state is altered to report its written byte count as 1.
   *
   * @param writer the {@link UfsJournalLogWriter} instance for which the mock is created
   * @return the created mock {@link DataOutputStream} instance
   * @throws IOException
   */
  private DataOutputStream createMockDataOutputStream(UfsJournalLogWriter writer)
      throws IOException {
    flushOutputStream(writer);
    DataOutputStream badOut = Mockito.mock(DataOutputStream.class);
    Whitebox.setInternalState(badOut, "written", 1);
    Object journalOutputStream = writer.getJournalOutputStream();
    Whitebox.setInternalState(journalOutputStream, "mOutputStream", badOut);
    return badOut;
  }

  /**
   * Flushes the writer's data output stream.
   * @param writer
   * @throws IOException
   */
  private void flushOutputStream(UfsJournalLogWriter writer) throws IOException {
    Object journalOutputStream = writer.getJournalOutputStream();
    DataOutputStream outputStream =
        Whitebox.getInternalState(journalOutputStream, "mOutputStream");
    outputStream.flush();
  }

  /**
   * Tries to write a journal entry and expects the write to fail due to {@link IOException}.
   *
   * @param writer {@link UfsJournalLogWriter} that attempts the write
   * @param nextSN the sequence number that the entry is expected to have
   */
  private void tryWriteAndExpectToFail(UfsJournalLogWriter writer, long nextSN) throws Exception {
    try {
      writer.write(newEntry(nextSN));
      Assert.fail("Should not reach here.");
    } catch (IOException e) {
      Assert.assertThat(e.getMessage(), containsString(INJECTED_IO_ERROR_MESSAGE));
    }
  }

  /**
   * Tries to flush a journalwriter it to fail due to {@link IOException}.
   *
   * @param writer {@link UfsJournalLogWriter} that attempts the write
   */
  private void tryFlushAndExpectToFail(UfsJournalLogWriter writer) throws Exception {
    try {
      writer.flush();
      Assert.fail("Should not reach here.");
    } catch (IOException e) {
      Assert.assertThat(e.getMessage(), containsString(INJECTED_IO_ERROR_MESSAGE));
    }
  }

  /**
   * Checks that journal entries with sequence number between startSN (inclusive) and endSN
   * (exclusive) exist in the current journal files.
   *
   * @param startSN start sequence number (inclusive)
   * @param endSN end sequence number (exclusive)
   */
  private void checkJournalEntries(long startSN, long endSN)
      throws IOException {
    try (JournalReader reader = new UfsJournalReader(mJournal, startSN, true)) {
      long seq = startSN;
      while (reader.advance() == State.LOG) {
        Assert.assertEquals(seq, reader.getEntry().getSequenceNumber());
        seq++;
      }
      Assert.assertEquals(endSN, seq);
    }
  }

  /**
   * 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();
  }
}
