<commit_before>/*
 * 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 alluxio.exception.ExceptionMessage;
import alluxio.exception.InvalidJournalEntryException;
import alluxio.master.journal.JournalEntryStreamReader;
import alluxio.master.journal.JournalReader;
import alluxio.proto.journal.Journal;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.OpenOptions;

import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Queue;

import javax.annotation.concurrent.NotThreadSafe;

/**
 * Implementation of {@link JournalReader} that reads journal entries from a UFS. It can optionally
 * read after a given sequence number. By default, it starts from 0 sequence number.
 * If this reader runs in a primary master, it reads the incomplete log.
 * If this reader runs in a secondary master, it does not read the incomplete log.
 */
@NotThreadSafe
public final class UfsJournalReader implements JournalReader {
  private static final Logger LOG = LoggerFactory.getLogger(UfsJournalReader.class);

  private final UfsJournal mJournal;
  private final UnderFileSystem mUfs;

  /**
   * The next edit log sequence number to read. This is not incremented when reading from
   * the checkpoint.
   */
  private long mNextSequenceNumber;
  /** The input stream to read the journal entries. */
  private JournalInputStream mInputStream;
  /** A queue of files to be processed including checkpoint and logs. */
  private final Queue<UfsJournalFile> mFilesToProcess;

  private final boolean mReadIncompleteLog;

  /** Whether the reader is closed. */
  private boolean mClosed;

  /**
   * A simple wrapper that wraps the journal file and the input stream.
   */
  private class JournalInputStream implements Closeable {
    final UfsJournalFile mFile;
    /** The reader reading journal entries from the UfsJournalFile. */
    final JournalEntryStreamReader mReader;

    JournalInputStream(UfsJournalFile file) throws IOException {
      mFile = file;
      LOG.info("Reading journal file {}.", file.getLocation());
      mReader = new JournalEntryStreamReader(mUfs.open(file.getLocation().toString(),
          OpenOptions.defaults().setRecoverFailedOpen(true)));
    }

    /**
     * @return whether we have finished reading the current file
     */
    boolean isDone() {
      return mFile.getEnd() == mNextSequenceNumber;
    }

    @Override
    public void close() throws IOException {
      mReader.close();
    }
  }

  /**
   * Creates a new instance of {@link UfsJournalReader}.
   *
   * @param journal the handle to the journal
   * @param readIncompleteLog whether to read incomplete logs
   */
  UfsJournalReader(UfsJournal journal, boolean readIncompleteLog) {
    this(journal, 0, readIncompleteLog);
  }

  /**
   * Creates a new instance of {@link UfsJournalReader}.
   *
   * @param journal the handle to the journal
   * @param startSequenceId the sequence ID to start reading from
   * @param readIncompleteLog whether to read incomplete logs
   */
  public UfsJournalReader(UfsJournal journal, long startSequenceId, boolean readIncompleteLog) {
    mFilesToProcess = new ArrayDeque<>();
    mJournal = Preconditions.checkNotNull(journal, "journal");
    mUfs = mJournal.getUfs();
    mNextSequenceNumber = startSequenceId;
    mReadIncompleteLog = readIncompleteLog;
  }

  @Override
  public void close() throws IOException {
    if (mClosed) {
      return;
    }
    mClosed = true;
    if (mInputStream != null) {
      mInputStream.close();
    }
  }

  @Override
  public long getNextSequenceNumber() {
    return mNextSequenceNumber;
  }

  @Override
  public Journal.JournalEntry read() throws IOException, InvalidJournalEntryException {
    while (true) {
      Journal.JournalEntry entry = readInternal();
      if (entry == null) {
        return null;
      }
      if (mInputStream.mFile.isCheckpoint()) {
        return entry;
      } else if (entry.getSequenceNumber() == mNextSequenceNumber) {
        mNextSequenceNumber++;
        return entry;
      } else if (entry.getSequenceNumber() < mNextSequenceNumber) {
        // This can happen in the following two scenarios:
        // 1. The primary master failed when renaming the current log to completed log which might
        //    result in duplicate logs.
        // 2. The first completed log after the checkpoint's last sequence number might contains
        //    some duplicate entries with the checkpoint.
        LOG.debug("Skipping duplicate log entry {} (next sequence number: {}).", entry,
            mNextSequenceNumber);
      } else {
        throw new InvalidJournalEntryException(ExceptionMessage.JOURNAL_ENTRY_MISSING,
            mNextSequenceNumber, entry.getSequenceNumber());
      }
    }
  }

  /**
   * Reads the next journal entry.
   *
   * @return the journal entry, null if no journal entry is found
   * @throws InvalidJournalEntryException if the journal entry found is invalid
   */
  private Journal.JournalEntry readInternal() throws IOException, InvalidJournalEntryException {
    updateInputStream();
    if (mInputStream == null) {
      return null;
    }
    JournalEntry entry = mInputStream.mReader.readEntry();
    if (entry != null) {
      return entry;
    }
    // If this is the checkpoint file, we need to reset the sequence number to update the stream
    // because the sequence number in the checkpoint entries is not in the same space as the
    // sequence number in the edit logs.
    if (mInputStream.mFile.isCheckpoint()) {<commit_msg>Improve handling for missing journal entries (#6913)<commit_after>/*
 * 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 alluxio.exception.ExceptionMessage;
import alluxio.exception.InvalidJournalEntryException;
import alluxio.master.journal.JournalEntryStreamReader;
import alluxio.master.journal.JournalReader;
import alluxio.proto.journal.Journal;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.OpenOptions;

import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayDeque;
import java.util.Queue;

import javax.annotation.concurrent.NotThreadSafe;

/**
 * Implementation of {@link JournalReader} that reads journal entries from a UFS. It can optionally
 * read after a given sequence number. By default, it starts from 0 sequence number.
 * If this reader runs in a primary master, it reads the incomplete log.
 * If this reader runs in a secondary master, it does not read the incomplete log.
 */
@NotThreadSafe
public final class UfsJournalReader implements JournalReader {
  private static final Logger LOG = LoggerFactory.getLogger(UfsJournalReader.class);

  private final UfsJournal mJournal;
  private final UnderFileSystem mUfs;

  /**
   * The next edit log sequence number to read. This is not incremented when reading from
   * the checkpoint.
   */
  private long mNextSequenceNumber;
  /** The input stream to read the journal entries. */
  private JournalInputStream mInputStream;
  /** A queue of files to be processed including checkpoint and logs. */
  private final Queue<UfsJournalFile> mFilesToProcess;

  private final boolean mReadIncompleteLog;

  /** Whether the reader is closed. */
  private boolean mClosed;

  /**
   * A simple wrapper that wraps the journal file and the input stream.
   */
  private class JournalInputStream implements Closeable {
    final UfsJournalFile mFile;
    /** The reader reading journal entries from the UfsJournalFile. */
    final JournalEntryStreamReader mReader;

    JournalInputStream(UfsJournalFile file) throws IOException {
      mFile = file;
      LOG.info("Reading journal file {}.", file.getLocation());
      mReader = new JournalEntryStreamReader(mUfs.open(file.getLocation().toString(),
          OpenOptions.defaults().setRecoverFailedOpen(true)));
    }

    /**
     * @return whether we have finished reading the current file
     */
    boolean isDone() {
      return mFile.getEnd() == mNextSequenceNumber;
    }

    @Override
    public void close() throws IOException {
      mReader.close();
    }
  }

  /**
   * Creates a new instance of {@link UfsJournalReader}.
   *
   * @param journal the handle to the journal
   * @param readIncompleteLog whether to read incomplete logs
   */
  UfsJournalReader(UfsJournal journal, boolean readIncompleteLog) {
    this(journal, 0, readIncompleteLog);
  }

  /**
   * Creates a new instance of {@link UfsJournalReader}.
   *
   * @param journal the handle to the journal
   * @param startSequenceId the sequence ID to start reading from
   * @param readIncompleteLog whether to read incomplete logs
   */
  public UfsJournalReader(UfsJournal journal, long startSequenceId, boolean readIncompleteLog) {
    mFilesToProcess = new ArrayDeque<>();
    mJournal = Preconditions.checkNotNull(journal, "journal");
    mUfs = mJournal.getUfs();
    mNextSequenceNumber = startSequenceId;
    mReadIncompleteLog = readIncompleteLog;
  }

  @Override
  public void close() throws IOException {
    if (mClosed) {
      return;
    }
    mClosed = true;
    if (mInputStream != null) {
      mInputStream.close();
    }
  }

  @Override
  public long getNextSequenceNumber() {
    return mNextSequenceNumber;
  }

  @Override
  public Journal.JournalEntry read() throws IOException, InvalidJournalEntryException {
    while (true) {
      Journal.JournalEntry entry = readInternal();
      if (entry == null) {
        return null;
      }
      if (mInputStream.mFile.isCheckpoint()) {
        return entry;
      } else if (entry.getSequenceNumber() == mNextSequenceNumber) {
        mNextSequenceNumber++;
        return entry;
      } else if (entry.getSequenceNumber() < mNextSequenceNumber) {
        // This can happen in the following two scenarios:
        // 1. The primary master failed when renaming the current log to completed log which might
        //    result in duplicate logs.
        // 2. The first completed log after the checkpoint's last sequence number might contains
        //    some duplicate entries with the checkpoint.
        LOG.debug("Skipping duplicate log entry {} (next sequence number: {}).", entry,
            mNextSequenceNumber);
      } else {
        throw new IllegalStateException(ExceptionMessage.JOURNAL_ENTRY_MISSING.getMessage(
            mNextSequenceNumber, entry.getSequenceNumber()));
      }
    }
  }

  /**
   * Reads the next journal entry.
   *
   * @return the journal entry, null if no journal entry is found
   * @throws InvalidJournalEntryException if the journal entry found is invalid
   */
  private Journal.JournalEntry readInternal() throws IOException, InvalidJournalEntryException {
    updateInputStream();
    if (mInputStream == null) {
      return null;
    }
    JournalEntry entry = mInputStream.mReader.readEntry();
    if (entry != null) {
      return entry;
    }
    // If this is the checkpoint file, we need to reset the sequence number to update the stream
    // because the sequence number in the checkpoint entries is not in the same space as the
    // sequence number in the edit logs.
    if (mInputStream.mFile.isCheckpoint()) {
{'fraction_non_alphanumeric': 0.0798359864930053, 'fraction_numerical': 0.001929570670525808, 'mean_word_length': 3.4552292263610314, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2773194', 'n_tokens_mistral': 3300, 'n_tokens_neox': 3217, 'n_words': 1178}