/*
 * 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.raft;

import alluxio.Constants;
import alluxio.ProcessUtils;
import alluxio.annotation.SuppressFBWarnings;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.status.UnavailableException;
import alluxio.grpc.AddQuorumServerRequest;
import alluxio.grpc.JournalQueryRequest;
import alluxio.master.StateLockManager;
import alluxio.master.StateLockOptions;
import alluxio.master.journal.CatchupFuture;
import alluxio.master.journal.JournalUtils;
import alluxio.master.journal.Journaled;
import alluxio.master.journal.SingleEntryJournaled;
import alluxio.master.journal.checkpoint.CheckpointInputStream;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.resource.LockResource;
import alluxio.util.StreamUtils;
import alluxio.util.logging.SamplingLogger;

import com.codahale.metrics.Timer;
import com.google.common.base.Preconditions;
import org.apache.ratis.proto.RaftProtos;
import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftGroupMemberId;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.server.raftlog.RaftLog;
import org.apache.ratis.server.storage.RaftStorage;
import org.apache.ratis.statemachine.SnapshotInfo;
import org.apache.ratis.statemachine.StateMachineStorage;
import org.apache.ratis.statemachine.TransactionContext;
import org.apache.ratis.statemachine.impl.BaseStateMachine;
import org.apache.ratis.statemachine.impl.SimpleStateMachineStorage;
import org.apache.ratis.util.LifeCycle;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.DataInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.time.Duration;
import java.time.Instant;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Stream;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;

/**
 * A state machine for managing all of Alluxio's journaled state. Entries applied to this state
 * machine will be forwarded to the appropriate internal master.
 *
 * The state machine starts by resetting all state, then applying the entries offered by Ratis.
 * When the master becomes primary, it should wait until the state machine is up to date and no
 * other primary master is serving, then call {@link #upgrade}. Once the state machine is upgraded,
 * it will ignore all entries appended by Ratis because those entries are applied to primary
 * master state before being written to Rati.
 *
 */
@ThreadSafe
public class JournalStateMachine extends BaseStateMachine {
  private static final Logger LOG = LoggerFactory.getLogger(JournalStateMachine.class);
  private static final Logger SAMPLING_LOG = new SamplingLogger(LOG, 10L * Constants.SECOND_MS);

  private static final CompletableFuture<Message> EMPTY_FUTURE =
      CompletableFuture.completedFuture(Message.EMPTY);

  /** Journals managed by this applier. */
  private final Map<String, RaftJournal> mJournals;
  private final RaftJournalSystem mJournalSystem;
  private final RaftSnapshotManager mSnapshotManager;
  private final SnapshotDirStateMachineStorage mStorage;
  private final AtomicReference<StateLockManager> mStateLockManagerRef
      = new AtomicReference<>(null);
  @GuardedBy("this")
  private boolean mIgnoreApplys = false;
  @GuardedBy("this")
  private boolean mClosed = false;

  private volatile long mLastAppliedCommitIndex = -1;
  // The last special "primary start" sequence number applied to this state machine. These special
  // sequence numbers are identified by being negative.
  private volatile long mLastPrimaryStartSequenceNumber = 0;
  private volatile long mNextSequenceNumberToRead = 0;
  private volatile boolean mSnapshotting = false;
  private volatile boolean mIsLeader = false;

  private final ExecutorService mJournalPool = Executors.newCachedThreadPool();

  /**
   * This callback is used for interrupting someone who suspends the journal applier to work on
   * the states. It helps prevent dirty read/write of the states when the journal is reloading.
   *
   * Here is an example of interrupting backup tasks when the state machine reloads:
   *
   * - Backup worker suspends state machine before backup, passing in the callback.
   * - Backup worker writes journal entries to UFS.
   * - Raft state machine downloads a new snapshot from leader.
   * - Raft state machine transitions to PAUSE state and invokes the callback.
   * - Backup worker handles the callback and interrupts the backup tasks.
   * - Raft state machine starts reloading the states.
   * - Raft state machine finished the reload and transitions back to RUNNING state.
   */
  private volatile Runnable mInterruptCallback;

  // The last index of the latest journal snapshot
  // created by this master or downloaded from other masters
  private volatile long mSnapshotLastIndex = -1;
  private long mLastSnapshotTime = -1;
  @SuppressFBWarnings(value = "IS2_INCONSISTENT_SYNC",
      justification = "Written in synchronized block, read by metrics")
  private long mLastSnapshotDurationMs = -1;
  @SuppressFBWarnings(value = "IS2_INCONSISTENT_SYNC",
      justification = "Written in synchronized block, read by metrics")
  private long mLastSnapshotEntriesCount = -1;
  private long mLastSnapshotReplayDurationMs = -1;
  private long mLastSnapshotReplayEntriesCount = -1;
  /** Used to control applying to masters. */
  private BufferedJournalApplier mJournalApplier;

  /**
   * @param journals      master journals; these journals are still owned by the caller, not by the
   *                      journal state machine
   * @param journalSystem the raft journal system
   * @param storage the {@link SnapshotDirStateMachineStorage} that this state machine will use
   */
  public JournalStateMachine(Map<String, RaftJournal> journals, RaftJournalSystem journalSystem,
                             SnapshotDirStateMachineStorage storage) {
    mJournals = journals;
    mJournalApplier = new BufferedJournalApplier(journals,
        () -> journalSystem.getJournalSinks(null));
    resetState();
    LOG.info("Initialized new journal state machine");
    mJournalSystem = journalSystem;
    mStorage = storage;
    mSnapshotManager = new RaftSnapshotManager(mStorage, mJournalPool);

    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_EMBEDDED_JOURNAL_SNAPSHOT_LAST_INDEX.getName(),
        () -> mSnapshotLastIndex);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_JOURNAL_ENTRIES_SINCE_CHECKPOINT.getName(),
        () -> getLastAppliedTermIndex().getIndex() - mSnapshotLastIndex);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_JOURNAL_LAST_CHECKPOINT_TIME.getName(),
        () -> mLastSnapshotTime);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_JOURNAL_LAST_APPLIED_COMMIT_INDEX.getName(),
        () -> mLastAppliedCommitIndex);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_JOURNAL_CHECKPOINT_WARN.getName(),
        () -> getLastAppliedTermIndex().getIndex() - mSnapshotLastIndex
                > Configuration.getInt(PropertyKey.MASTER_JOURNAL_CHECKPOINT_PERIOD_ENTRIES)
                && System.currentTimeMillis() - mLastSnapshotTime > Configuration.getMs(
                PropertyKey.MASTER_WEB_JOURNAL_CHECKPOINT_WARNING_THRESHOLD_TIME)
    );
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_EMBEDDED_JOURNAL_LAST_SNAPSHOT_DURATION_MS.getName(),
        () -> mLastSnapshotDurationMs);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_EMBEDDED_JOURNAL_LAST_SNAPSHOT_ENTRIES_COUNT.getName(),
        () -> mLastSnapshotEntriesCount);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_EMBEDDED_JOURNAL_LAST_SNAPSHOT_REPLAY_DURATION_MS.getName(),
        () -> mLastSnapshotReplayDurationMs);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_EMBEDDED_JOURNAL_LAST_SNAPSHOT_REPLAY_ENTRIES_COUNT.getName(),
        () -> mLastSnapshotReplayEntriesCount);
  }

  @Override
  public void initialize(RaftServer server, RaftGroupId groupId,
      RaftStorage raftStorage) throws IOException {
    getLifeCycle().startAndTransition(() -> {
      super.initialize(server, groupId, raftStorage);
      mStorage.init(raftStorage);
      loadSnapshot(getLatestSnapshot());
    });
  }

  @Override
  public void reinitialize() throws IOException {
    LOG.info("Reinitializing state machine.");
    mStorage.loadLatestSnapshot();
    loadSnapshot(getLatestSnapshot());
    unpause();
  }

  private synchronized void loadSnapshot(SnapshotInfo snapshot) throws IOException {
    if (snapshot == null) {
      LOG.info("No snapshot to load");
      return;
    }
    try {
      resetState();
      setLastAppliedTermIndex(snapshot.getTermIndex());
      LOG.debug("Loading snapshot {}", snapshot);
      install(snapshot);
      LOG.debug("Finished loading snapshot {}", snapshot);
      mSnapshotLastIndex = getLatestSnapshot() != null ? getLatestSnapshot().getIndex() : -1;
    } catch (Exception e) {
      throw new IOException(String.format("Failed to load snapshot %s", snapshot), e);
    }
  }

  /**
   * Allows leader to take snapshots. This is used exclusively for the
   * `bin/alluxio journal checkpoint` command.
   * @param manager allows the state machine to take a snapshot as leader by using it
   */
  void allowLeaderSnapshots(StateLockManager manager) {
    mStateLockManagerRef.set(manager);
  }

  void disallowLeaderSnapshots() {
    mStateLockManagerRef.set(null);
  }

  @Override
  public long takeSnapshot() {
    long index;
    StateLockManager stateLockManager = mStateLockManagerRef.get();
    if (!mIsLeader) {
      LOG.info("Taking local snapshot as follower");
      index = takeLocalSnapshot(false);
    } else if (stateLockManager != null) {
      // the leader has been allowed to take a local snapshot by being given a non-null
      // StateLockManager through the #allowLeaderSnapshots method
      try (LockResource stateLock = stateLockManager.lockExclusive(StateLockOptions.defaults())) {
        LOG.info("Taking local snapshot as leader");
        index = takeLocalSnapshot(true);
      } catch (Exception e) {
        return RaftLog.INVALID_LOG_INDEX;
      }
    } else {
      index = mSnapshotManager.downloadSnapshotFromOtherMasters();
    }
    // update metrics if took a snapshot
    if (index != RaftLog.INVALID_LOG_INDEX) {
      mSnapshotLastIndex = index;
      mLastSnapshotTime = System.currentTimeMillis();
      LOG.info("Took snapshot up to index {} at time {}", mSnapshotLastIndex, DateTime.now());
    }
    return index;
  }

  /**
   * @return the latest snapshot information, or null of no snapshot exists
   */
  @Override
  public SnapshotInfo getLatestSnapshot() {
    return mStorage.getLatestSnapshot();
  }

  @Override
  public StateMachineStorage getStateMachineStorage() {
    return mStorage;
  }

  @Override
  public CompletableFuture<Message> query(Message request) {
    CompletableFuture<Message> future = new CompletableFuture<>();
    try {
      JournalQueryRequest queryRequest = JournalQueryRequest.parseFrom(
          request.getContent().asReadOnlyByteBuffer());
      LOG.debug("Received query request: {}", queryRequest);
      if (queryRequest.hasAddQuorumServerRequest()) {
        AddQuorumServerRequest addRequest = queryRequest.getAddQuorumServerRequest();
        return CompletableFuture.supplyAsync(() -> {
          try {
            mJournalSystem.addQuorumServer(addRequest.getServerAddress());
          } catch (IOException e) {
            throw new CompletionException(e);
          }
          return Message.EMPTY;
        });
      } else {
        return super.query(request);
      }
    } catch (Exception e) {
      LOG.error("failed processing request {}", request, e);
      future.completeExceptionally(e);
      return future;
    }
  }

  @Override
  public void close() {
    mClosed = true;
    MetricsSystem.removeMetrics(MetricKey.MASTER_EMBEDDED_JOURNAL_SNAPSHOT_LAST_INDEX.getName());
    MetricsSystem.removeMetrics(MetricKey.MASTER_JOURNAL_ENTRIES_SINCE_CHECKPOINT.getName());
    MetricsSystem.removeMetrics(MetricKey.MASTER_JOURNAL_LAST_CHECKPOINT_TIME.getName());
    MetricsSystem.removeMetrics(MetricKey.MASTER_JOURNAL_LAST_APPLIED_COMMIT_INDEX.getName());
    MetricsSystem.removeMetrics(MetricKey.MASTER_JOURNAL_CHECKPOINT_WARN.getName());
    mSnapshotManager.close();
  }

  @Override
  public CompletableFuture<Message> applyTransaction(TransactionContext trx) {
    try {
      applyJournalEntryCommand(trx);
      RaftProtos.LogEntryProto entry = Objects.requireNonNull(trx.getLogEntry());
      updateLastAppliedTermIndex(entry.getTerm(), entry.getIndex());
      // explicitly return empty future since no response message is expected by the journal writer
      // avoid using super.applyTransaction() since it will echo the message and add overhead
      return EMPTY_FUTURE;
    } catch (Exception e) {
      return RaftJournalUtils.completeExceptionally(e);
    }
  }

  @Override
  public void notifyNotLeader(Collection<TransactionContext> pendingEntries) {
    mIsLeader = false;
    mJournalSystem.notifyLeadershipStateChanged(false);
  }

  @Override
  public void notifyConfigurationChanged(long term, long index,
      RaftProtos.RaftConfigurationProto newRaftConfiguration) {
    CompletableFuture.runAsync(mJournalSystem::updateGroup, mJournalPool);
  }

  @Override
  public CompletableFuture<TermIndex> notifyInstallSnapshotFromLeader(
      RaftProtos.RoleInfoProto roleInfoProto, TermIndex firstTermIndexInLog) {
    // this method is called automatically by Ratis when the leader does not have all the logs to
    // give to this follower. This method instructs the follower to download a snapshot from
    // other masters to become up-to-date.
    LOG.info("Received instruction to install snapshot from other master asynchronously");
    return CompletableFuture.supplyAsync(() -> {
      mSnapshotManager.downloadSnapshotFromOtherMasters();
      long index = mSnapshotManager.waitForAttemptToComplete();
      if (index == RaftLog.INVALID_LOG_INDEX) {
        LOG.info("Failed to install snapshot from other master asynchronously");
        return null;
      }
      return getLatestSnapshot().getTermIndex();
    }, mJournalPool);
  }

  @Override
  public synchronized void pause() {
    LOG.info("Pausing raft state machine.");
    getLifeCycle().transition(LifeCycle.State.PAUSING);
    if (mInterruptCallback != null) {
      LOG.info("Invoking suspension interrupt callback.");
      mInterruptCallback.run();
      mInterruptCallback = null;
    }
    try {
      if (mJournalApplier.isSuspended()) {
        // make sure there are no pending entries
        LOG.info("Resuming journal applier.");
        mJournalApplier.resume();
      }
    } catch (IOException e) {
      throw new IllegalStateException("State machine pause failed", e);
    }
    getLifeCycle().transition(LifeCycle.State.PAUSED);
    LOG.info("Raft state machine is paused.");
  }

  /**
   * Unpause the StateMachine. This should be done after uploading new state to the StateMachine.
   */
  public synchronized void unpause() {
    LOG.info("Unpausing raft state machine.");
    if (mJournalApplier.isSuspended()) {
      LOG.warn("Journal should not be suspended while state machine is paused.");
    }
    getLifeCycle().startAndTransition(() -> {
      // nothing to do - just use this method to transition from PAUSE to RUNNING state
    });
    LOG.info("Raft state machine is unpaused.");
  }

  /**
   * Applies a journal entry commit to the state machine.
   * @param commit the commit
   */
  public synchronized void applyJournalEntryCommand(TransactionContext commit) {
    JournalEntry entry;
    try {
      entry = JournalEntry.parseFrom(
          commit.getStateMachineLogEntry().getLogData().asReadOnlyByteBuffer());
    } catch (Exception e) {
      ProcessUtils.fatalError(LOG, e,
          "Encountered invalid journal entry in commit: %s.", commit);
      System.exit(-1);
      throw new IllegalStateException(e); // We should never reach here.
    }
    try {
      applyEntry(entry);
    } finally {
      Preconditions.checkState(commit.getLogEntry().getIndex() > mLastAppliedCommitIndex);
      mLastAppliedCommitIndex = commit.getLogEntry().getIndex();
    }
  }

  /**
   * Applies the journal entry, ignoring empty entries and expanding multi-entries.
   *
   * @param entry the entry to apply
   */
  private void applyEntry(JournalEntry entry) {
    if (LOG.isDebugEnabled()) {
      // This check is put behind the debug flag as the call to getAllFields creates
      // a map and is very expensive
      Preconditions.checkState(
          entry.getAllFields().size() <= 2
              || (entry.getAllFields().size() == 3 && entry.hasSequenceNumber()),
          "Raft journal entries should never set multiple fields in addition to sequence "
              + "number, but found %s",
          entry);
    }
    if (entry.getJournalEntriesCount() > 0) {
      // This entry aggregates multiple entries.
      for (JournalEntry e : entry.getJournalEntriesList()) {
        applyEntry(e);
      }
    } else if (entry.getSequenceNumber() < 0) {
      // Negative sequence numbers indicate special entries used to indicate that a new primary is
      // starting to serve.
      mLastPrimaryStartSequenceNumber = entry.getSequenceNumber();
    } else if (entry.toBuilder().clearSequenceNumber().build()
        .equals(JournalEntry.getDefaultInstance())) {
      // Ignore empty entries, they are created during snapshotting.
    } else {
      applySingleEntry(entry);
    }
  }

  @SuppressFBWarnings(value = "VO_VOLATILE_INCREMENT",
      justification = "All calls to applyJournalEntryCommand() are synchronized by ratis")
  private void applySingleEntry(JournalEntry entry) {
    if (mClosed) {
      return;
    }
    long newSN = entry.getSequenceNumber();
    if (newSN < mNextSequenceNumberToRead) {
      // This can happen due to retried writes. For example, if flushing [3, 4] fails, we will
      // retry, and the log may end up looking like [1, 2, 3, 4, 3, 4] if the original request
      // eventually succeeds. Once we've read the first "4", we must ignore the next two entries.
      LOG.debug("Ignoring duplicate journal entry with SN {} when next SN is {}", newSN,
          mNextSequenceNumberToRead);
      return;
    }
    if (newSN > mNextSequenceNumberToRead) {
      ProcessUtils.fatalError(LOG,
          "Unexpected journal entry. The next expected SN is %s, but"
              + " encountered an entry with SN %s. Full journal entry: %s",
          mNextSequenceNumberToRead, newSN, entry);
    }

    mNextSequenceNumberToRead++;
    if (!mIgnoreApplys) {
      mJournalApplier.processJournalEntry(entry);
    }
  }

  /**
   * Takes a snapshot of local state machine.
   * @param hasStateLock indicates whether this method call is guarded by a state lock
   * @return the index of last included entry, or {@link RaftLog#INVALID_LOG_INDEX} if it fails
   */
  public synchronized long takeLocalSnapshot(boolean hasStateLock) {
    // Snapshot format is [snapshotId, name1, bytes1, name2, bytes2, ...].
    if (mClosed) {
      SAMPLING_LOG.info("Skip taking snapshot because state machine is closed.");
      return RaftLog.INVALID_LOG_INDEX;
    }
    RaftServer server = getServer().join(); // gets completed during initialization
    if (server.getLifeCycleState() != LifeCycle.State.RUNNING) {
      SAMPLING_LOG.info("Skip taking snapshot because raft server is not in running state: "
          + "current state is {}.", server.getLifeCycleState());
      return RaftLog.INVALID_LOG_INDEX;
    }
    if (mJournalApplier.isSuspended()) {
      SAMPLING_LOG.info("Skip taking snapshot while journal application is suspended.");
      return RaftLog.INVALID_LOG_INDEX;
    }
    // Recheck mIsLeader (even though it was checked in #takeSnapshot) because mIsLeader is volatile
    // synchronized call to #isSnapshotting will prevent gaining leadership while this method is
    // executing
    if (mIsLeader && !hasStateLock) {
      LOG.warn("Tried to take local snapshot as leader without state lock");
      return RaftLog.INVALID_LOG_INDEX;
    }
    LOG.debug("Calling snapshot");
    Preconditions.checkState(!mSnapshotting, "Cannot call snapshot multiple times concurrently");
    mSnapshotting = true;
    TermIndex last = getLastAppliedTermIndex();

    File snapshotDir = getSnapshotDir(last.getTerm(), last.getIndex());
    if (!snapshotDir.isDirectory() && !snapshotDir.mkdir()) {
      return RaftLog.INVALID_LOG_INDEX;
    }
    try (Timer.Context ctx = MetricsSystem.timer(
             MetricKey.MASTER_EMBEDDED_JOURNAL_SNAPSHOT_GENERATE_TIMER.getName()).time()) {
      Instant start = Instant.now();
      long snapshotId = mNextSequenceNumberToRead - 1;
      SingleEntryJournaled idWriter = new SnapshotIdJournaled();
      idWriter.processJournalEntry(JournalEntry.newBuilder().setSequenceNumber(snapshotId).build());
      CompletableFuture.allOf(Stream.concat(Stream.of(idWriter), getStateMachines().stream())
          .map(journaled -> journaled.writeToCheckpoint(snapshotDir, mJournalPool))
          .toArray(CompletableFuture[]::new))
          .join();
      mStorage.loadLatestSnapshot();
      mStorage.signalNewSnapshot();

      mLastSnapshotDurationMs = Duration.between(start, Instant.now()).toMillis();
      mLastSnapshotEntriesCount = mNextSequenceNumberToRead;
      return last.getIndex();
    } catch (Exception e) {
      LOG.error("error taking snapshot", e);
      return RaftLog.INVALID_LOG_INDEX;
    } finally {
      mSnapshotting = false;
    }
  }

  private void install(SnapshotInfo snapshot) {
    if (mClosed) {
      return;
    }
    if (mIgnoreApplys) {
      LOG.warn("Unexpected request to install a snapshot on a read-only journal state machine");
      return;
    }

    File snapshotDir = getSnapshotDir(snapshot.getTerm(), snapshot.getIndex());
    long snapshotId = 0L;
    try (Timer.Context ctx = MetricsSystem.timer(MetricKey
        .MASTER_EMBEDDED_JOURNAL_SNAPSHOT_REPLAY_TIMER.getName()).time()) {
      Instant start = Instant.now();
      if (snapshotDir.isFile()) {
        LOG.info("Restoring from snapshot {} in old format", snapshot.getTermIndex());
        try (DataInputStream stream =  new DataInputStream(new FileInputStream(snapshotDir))) {
          snapshotId = stream.readLong();
          JournalUtils.restoreFromCheckpoint(new CheckpointInputStream(stream), getStateMachines());
        }
      } else {
        SingleEntryJournaled idReader = new SnapshotIdJournaled();
        CompletableFuture.allOf(Stream.concat(Stream.of(idReader), getStateMachines().stream())
                .map(journaled -> journaled.restoreFromCheckpoint(snapshotDir, mJournalPool))
                .toArray(CompletableFuture[]::new))
            .join();
        snapshotId = idReader.getEntry().getSequenceNumber();
      }
      mLastSnapshotReplayDurationMs = Duration.between(start, Instant.now()).toMillis();
    } catch (Exception e) {
      JournalUtils.handleJournalReplayFailure(LOG, e, "Failed to install snapshot: %s",
          snapshot.getTermIndex());
      if (Configuration.getBoolean(PropertyKey.MASTER_JOURNAL_TOLERATE_CORRUPTION)) {
        return;
      }
    }

    if (snapshotId < mNextSequenceNumberToRead - 1) {
      LOG.warn("Installed snapshot for SN {} but next SN to read is {}", snapshotId,
          mNextSequenceNumberToRead);
    }
    mNextSequenceNumberToRead = snapshotId + 1;
    mLastSnapshotReplayEntriesCount = mNextSequenceNumberToRead;
    LOG.info("Successfully installed snapshot up to SN {}", snapshotId);
  }

  private File getSnapshotDir(long term, long index) {
    String dirName = SimpleStateMachineStorage.getSnapshotFileName(term, index);
    return new File(mStorage.getSnapshotDir(), dirName);
  }

  /**
   * Suspends applying to masters.
   *
   * When using suspend, the caller needs to provide a callback method as parameter. This callback
   * is invoked when the journal needs to reload and thus cannot suspend the state changes any
   * more. The callback should cancel any tasks that access the master states. After the callback
   * returns, the journal assumes that the states is no longer being accessed and will reload
   * immediately.
   *
   * @param interruptCallback a callback function to be called when the suspend is interrupted
   * @throws IOException if suspension fails
   */
  public synchronized void suspend(Runnable interruptCallback) throws IOException {
    LOG.info("Suspending raft state machine.");
    if (!getLifeCycleState().isRunning()) {
      throw new UnavailableException("Cannot suspend journal when state machine is paused.");
    }
    mJournalApplier.suspend();
    mInterruptCallback = interruptCallback;
    LOG.info("Raft state machine is suspended.");
  }

  /**
   * Resumes applying to masters.
   *
   * @throws IOException if resuming fails
   */
  public synchronized void resume() throws IOException {
    LOG.info("Resuming raft state machine");
    mInterruptCallback = null;
    if (mJournalApplier.isSuspended()) {
      mJournalApplier.resume();
      LOG.info("Raft state machine resumed");
    } else {
      LOG.info("Raft state machine is already resumed");
    }
  }

  /**
   * Initiates catching up of masters to given sequence.
   *
   * @param sequence the target sequence
   * @return the future to track when catching up is done
   */
  public synchronized CatchupFuture catchup(long sequence) {
    return mJournalApplier.catchup(sequence);
  }

  private List<Journaled> getStateMachines() {
    return StreamUtils.map(RaftJournal::getStateMachine, mJournals.values());
  }

  private synchronized void resetState() {
    if (mClosed) {
      return;
    }
    if (mIgnoreApplys) {
      LOG.warn("Unexpected call to resetState() on a read-only journal state machine");
      return;
    }
    mJournalApplier.close();
    mJournalApplier = new BufferedJournalApplier(mJournals,
        () -> mJournalSystem.getJournalSinks(null));
    for (RaftJournal journal : mJournals.values()) {
      journal.getStateMachine().resetState();
    }
  }

  /**
   * Upgrades the journal state machine to primary mode.
   *
   * @return the last sequence number read while in standby mode
   */
  public synchronized long upgrade() {
    // Resume the journal applier if was suspended.
    if (mJournalApplier.isSuspended()) {
      try {
        resume();
      } catch (IOException e) {
        ProcessUtils.fatalError(LOG, e, "State-machine failed to catch up after suspension.");
      }
    }
    mIgnoreApplys = true;
    return mNextSequenceNumberToRead - 1;
  }

  /**
   * @return the sequence number of the last entry applied to the state machine
   */
  public long getLastAppliedSequenceNumber() {
    return mNextSequenceNumberToRead - 1;
  }

  /**
   * @return the last primary term start sequence number applied to this state machine
   */
  public long getLastPrimaryStartSequenceNumber() {
    return mLastPrimaryStartSequenceNumber;
  }

  /**
   * @return the last raft log index which was applied to the state machine
   */
  public long getLastAppliedCommitIndex() {
    return mLastAppliedCommitIndex;
  }

  /**
   * @return whether the state machine is in the process of taking a snapshot
   */
  public synchronized boolean isSnapshotting() {
    return mSnapshotting;
  }

  @Override
  public void notifyLeaderChanged(RaftGroupMemberId groupMemberId, RaftPeerId raftPeerId) {
    if (getGroupId() == groupMemberId.getGroupId()) {
      mIsLeader = groupMemberId.getPeerId() == raftPeerId;
      mJournalSystem.notifyLeadershipStateChanged(mIsLeader);
    } else {
      LOG.warn("Received notification for unrecognized group {}, current group is {}",
          groupMemberId.getGroupId(), getGroupId());
    }
  }

  /**
   * @return whether the journal is suspended
   */
  public synchronized boolean isSuspended() {
    return mJournalApplier.isSuspended();
  }
}
