alluxio-master-2.x/core/server/common/src/main/java/alluxio/master/journal/ufs/UfsJournal.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 alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.JournalClosedException;
import alluxio.exception.status.CancelledException;
import alluxio.exception.status.UnavailableException;
import alluxio.master.Master;
import alluxio.master.journal.AbstractCatchupThread;
import alluxio.master.journal.AsyncJournalWriter;
import alluxio.master.journal.CatchupFuture;
import alluxio.master.journal.Journal;
import alluxio.master.journal.JournalContext;
import alluxio.master.journal.JournalReader;
import alluxio.master.journal.JournalUtils;
import alluxio.master.journal.MasterJournalContext;
import alluxio.master.journal.sink.JournalSink;
import alluxio.metrics.MetricKey;
import alluxio.metrics.MetricsSystem;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.resource.CloseableResource;
import alluxio.retry.ExponentialTimeBoundedRetry;
import alluxio.retry.RetryPolicy;
import alluxio.underfs.UfsStatus;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.UnderFileSystemConfiguration;
import alluxio.underfs.options.DeleteOptions;
import alluxio.util.URIUtils;
import alluxio.util.UnderFileSystemUtils;

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

import java.io.IOException;
import java.net.URI;
import java.time.Duration;
import java.util.Arrays;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Implementation of UFS-based journal.
 *
 * The journal is made up of 2 components:
 * - The checkpoint:  a snapshot of the master state
 * - The log entries: incremental entries to apply to the checkpoint.
 *
 * The journal log entries must be self-contained. Checkpoint is considered as a compaction of
 * a set of journal log entries. If the master does not do any checkpoint, the journal should
 * still be sufficient.
 *
 * Journal file structure:
 * journal_folder/version/logs/StartSequenceNumber-EndSequenceNumber
 * journal_folder/version/checkpoints/0-EndSequenceNumber
 * journal_folder/version/.tmp/random_id
 */
@ThreadSafe
public class UfsJournal implements Journal {
  private static final Logger LOG = LoggerFactory.getLogger(UfsJournal.class);
  /**
   * This is set to Long.MAX_VALUE such that the current log can be sorted after any other
   * completed logs.
   */
  public static final long UNKNOWN_SEQUENCE_NUMBER = Long.MAX_VALUE;
  /** The journal version. */
  public static final String VERSION = "v1";

  /** Directory for journal edit logs including the incomplete log file. */
  private static final String LOG_DIRNAME = "logs";
  /** Directory for committed checkpoints. */
  private static final String CHECKPOINT_DIRNAME = "checkpoints";
  /** Directory for temporary files. */
  private static final String TMP_DIRNAME = ".tmp";

  private final URI mLogDir;
  private final URI mCheckpointDir;
  private final URI mTmpDir;

  /** The location where this journal is stored. */
  private final URI mLocation;
  /** The master managed by this journal. */
  private final Master mMaster;
  /** The UFS where the journal is being written to. */
  private final UnderFileSystem mUfs;
  /** The amount of time to wait to pass without seeing a new journal entry when gaining primacy. */
  private final long mQuietPeriodMs;
  /** The current log writer. Null when in standby mode. */
  private UfsJournalLogWriter mWriter;
  /** Asynchronous journal writer. */
  private volatile AsyncJournalWriter mAsyncWriter;
  /**
   * Thread for tailing the journal, taking snapshots, and applying updates to the state machine.
   * Null when in primary mode.
   */
  private UfsJournalCheckpointThread mTailerThread;

  /** Whether the journal is suspended. */
  private volatile boolean mSuspended = false;
  /** Store where the journal was suspended. */
  private volatile long mSuspendSequence = -1;
  /** Used to store latest catch-up task. */
  private volatile AbstractCatchupThread mCatchupThread;
  /** Used to stop catching up when cancellation requested.  */
  private volatile boolean mStopCatchingUp = false;

  private long mLastCheckPointTime = -1;
  private long mEntriesSinceLastCheckPoint = 0;

  private enum State {
    STANDBY, PRIMARY, CLOSED
  }

  private final AtomicReference<State> mState = new AtomicReference<>(State.STANDBY);

  /** A supplier of journal sinks for this journal. */
  private final Supplier<Set<JournalSink>> mJournalSinks;

  /**
   * @return the ufs configuration to use for the journal operations
   */
  public static UnderFileSystemConfiguration getJournalUfsConf() {
    Map<String, Object> ufsConf =
        Configuration.getNestedProperties(PropertyKey.MASTER_JOURNAL_UFS_OPTION);
    return UnderFileSystemConfiguration.defaults(Configuration.global())
               .createMountSpecificConf(ufsConf);
  }

  /**
   * Creates a new instance of {@link UfsJournal}.
   *
   * @param location the location for this journal
   * @param master the master to manage
   * @param quietPeriodMs the amount of time to wait to pass without seeing a new journal entry when
   *        gaining primacy
   * @param journalSinks a supplier for journal sinks
   */
  public UfsJournal(URI location, Master master, long quietPeriodMs,
      Supplier<Set<JournalSink>> journalSinks) {
    try (CloseableResource<UnderFileSystem> ufs =
             master.getMasterContext().getUfsManager().getJournal(location).acquireUfsResource()) {
      mLocation = URIUtils.appendPathOrDie(location, VERSION);
      mMaster = master;
      mUfs = ufs.get();
      mQuietPeriodMs = quietPeriodMs;

      mLogDir = URIUtils.appendPathOrDie(mLocation, LOG_DIRNAME);
      mCheckpointDir = URIUtils.appendPathOrDie(mLocation, CHECKPOINT_DIRNAME);
      mTmpDir = URIUtils.appendPathOrDie(mLocation, TMP_DIRNAME);
      mJournalSinks = journalSinks;
      init();
    }
  }

  @VisibleForTesting
  UfsJournal(URI location, Master master, UnderFileSystem ufs,
      long quietPeriodMs, Supplier<Set<JournalSink>> journalSinks) {
    mLocation = URIUtils.appendPathOrDie(location, VERSION);
    mMaster = master;
    mUfs = ufs;
    mQuietPeriodMs = quietPeriodMs;

    mLogDir = URIUtils.appendPathOrDie(mLocation, LOG_DIRNAME);
    mCheckpointDir = URIUtils.appendPathOrDie(mLocation, CHECKPOINT_DIRNAME);
    mTmpDir = URIUtils.appendPathOrDie(mLocation, TMP_DIRNAME);
    mJournalSinks = journalSinks;
    init();
  }

  protected void init() {
    mState.set(State.STANDBY);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_JOURNAL_ENTRIES_SINCE_CHECKPOINT.getName() + "." + mMaster.getName(),
        this::getEntriesSinceLastCheckPoint);
    MetricsSystem.registerGaugeIfAbsent(
        MetricKey.MASTER_JOURNAL_LAST_CHECKPOINT_TIME.getName() + "." + mMaster.getName(),
        this::getLastCheckPointTime);
  }

  @Override
  public URI getLocation() {
    return mLocation;
  }

  private synchronized long getEntriesSinceLastCheckPoint() {
    return mEntriesSinceLastCheckPoint;
  }

  private synchronized long getLastCheckPointTime() {
    return mLastCheckPointTime;
  }

  /**
   * Writes a journal entry.
   * This is only used by tests where journal entries are manually written.
   * In the real logic the flush is handled by the {@link AsyncJournalWriter}
   * and no direct access to the {@link UfsJournalLogWriter}.
   * @param entry an entry to write to the journal
   */
  @VisibleForTesting
  synchronized void write(JournalEntry entry) throws IOException, JournalClosedException {
    writer().write(entry);
    mEntriesSinceLastCheckPoint++;
  }

  /**
   * Flushes the journal.
   * This is only used by tests where journal entries are manually flushed.
   * In the real logic the flush is handled by the {@link AsyncJournalWriter}
   * and no direct access to the {@link UfsJournalLogWriter}.
   */
  @VisibleForTesting
  public synchronized void flush() throws IOException, JournalClosedException {
    writer().flush();
  }

  @Override
  public synchronized JournalContext createJournalContext()
      throws UnavailableException {
    if (mState.get() != State.PRIMARY) {
      // We throw UnavailableException here so that clients will retry with the next primary master.
      throw new UnavailableException(
          mMaster.getName() + ": Not allowed to write to journal in state: " + mState.get());
    }
    AsyncJournalWriter writer = mAsyncWriter;
    if (writer == null) {
      throw new UnavailableException(
          mMaster.getName() + ": Failed to write to journal: journal is shutdown.");
    }
    return new MasterJournalContext(writer);
  }

  private synchronized UfsJournalLogWriter writer() {
    Preconditions.checkState(mState.get() == State.PRIMARY,
        "Cannot write to the journal in state " + mState.get());
    return mWriter;
  }

  /**
   * Starts the journal in standby mode.
   */
  public synchronized void start() {
    mMaster.resetState();
    mTailerThread = new UfsJournalCheckpointThread(mMaster, this, mJournalSinks);
    mTailerThread.start();
  }

  /**
   * Transitions the journal from standby to primary mode. The journal will apply the latest
   * journal entries to the state machine, then begin to allow writes.
   */
  public synchronized void gainPrimacy() throws IOException {
    Preconditions.checkState(mWriter == null, "writer must be null in standby mode");
    Preconditions.checkState(mSuspended || mTailerThread != null,
        "tailer thread must not be null in standby mode");

    // Resume first if suspended.
    if (mSuspended) {
      resume();
    }

    // If the tailing thread has crashed, the standby master will crash here
    // instead of gaining primacy
    mTailerThread.awaitTermination(true);
    long nextSequenceNumber = mTailerThread.getNextSequenceNumber();
    mTailerThread = null;
    // Read all the rest of journal entries if any, in the main thread
    nextSequenceNumber = catchUp(nextSequenceNumber);
    mWriter = new UfsJournalLogWriter(this, nextSequenceNumber);
    mAsyncWriter = new AsyncJournalWriter(mWriter, mJournalSinks, mMaster.getName());
    mState.set(State.PRIMARY);
    LOG.info("{}: journal switched to primary mode. location: {}", mMaster.getName(), mLocation);
  }

  /**
   * Notifies this journal that it is no longer primary. After this returns, the journal will not
   * allow any writes.
   *
   * The method {@link #awaitLosePrimacy()} must be called afterwards to complete the transition
   * from primary.
   */
  public synchronized void signalLosePrimacy() {
    Preconditions
        .checkState(mState.get() == State.PRIMARY, "unexpected journal state " + mState.get());
    mState.set(State.STANDBY);
    LOG.info("{}: journal switched to standby mode, starting transition. location: {}",
        mMaster.getName(), mLocation);
  }

  /**
   * Transitions the journal from primary to standby mode. The journal will no longer allow
   * writes, and the state machine is rebuilt from the journal and kept up to date.
   *
   * This must be called after {@link #signalLosePrimacy()} to finish the transition from primary.
   */
  public synchronized void awaitLosePrimacy() {
    Preconditions.checkState(mState.get() == State.STANDBY,
        "Should already be set to STANDBY state. unexpected state: " + mState.get());
    Preconditions.checkState(mWriter != null, "writer thread must not be null in primary mode");
    Preconditions.checkState(mTailerThread == null, "tailer thread must be null in primary mode");

    // Close async writer first to flush pending entries.
    mAsyncWriter.close();
    mAsyncWriter = null;
    mWriter.close();
    mWriter = null;
    mMaster.resetState();
    mTailerThread = new UfsJournalCheckpointThread(mMaster, this, mJournalSinks);
    mTailerThread.start();
  }

  /**
   * Suspends applying this journal until resumed.
   */
  public synchronized void suspend() {
    Preconditions.checkState(!mSuspended, "journal is already suspended");
    Preconditions.checkState(mState.get() == State.STANDBY, "unexpected state " + mState.get());
    Preconditions.checkState(mSuspendSequence == -1, "suspend sequence already set");
    // The standby suspends first in order to take a snapshot/backup
    // So if the tailing thread has crashed before that,
    // the crash will propagate and kill the standby master here
    mTailerThread.awaitTermination(false);
    mSuspendSequence = mTailerThread.getNextSequenceNumber() - 1;
    mTailerThread = null;
    mSuspended = true;
  }

  /**
   * Initiates catching up of the journal up to given sequence.
   * Note: Journal should have been suspended prior to calling this.
   *
   * @param sequence sequence to catch up
   * @return the catch-up task
   */
  public synchronized CatchupFuture catchup(long sequence) {
    Preconditions.checkState(mSuspended, "journal is not suspended");
    Preconditions.checkState(mState.get() == State.STANDBY, "unexpected state %s", mState.get());
    Preconditions.checkState(mTailerThread == null, "tailer is not null");
    Preconditions.checkState(sequence >= mSuspendSequence, "can't catch-up before suspend");
    Preconditions.checkState(mCatchupThread == null || !mCatchupThread.isAlive(),
        "Catch-up thread active");

    // Return completed if already at target.
    if (sequence == mSuspendSequence) {
      return CatchupFuture.completed();
    }

    // Create an async task to catch up to target sequence.
    // The thread is closed when the Future is completed
    mCatchupThread = new UfsJournalCatchupThread(mSuspendSequence + 1, sequence);
    mCatchupThread.start();
    return new CatchupFuture(mCatchupThread);
  }

  /**
   * Resumes the journal.
   * Note: Journal should have been suspended prior to calling this.
   */
  public synchronized void resume() {
    Preconditions.checkState(mSuspended, "journal is not suspended");
    Preconditions.checkState(mState.get() == State.STANDBY, "unexpected state " + mState.get());
    Preconditions.checkState(mTailerThread == null, "tailer is not null");

    // Cancel and wait for active catch-up thread.
    // If the catch-up thread crashed silently, the next catchup() will just create another one
    if (mCatchupThread != null && mCatchupThread.isAlive()) {
      mCatchupThread.cancel();
      mCatchupThread.waitTermination();
      mCatchupThread = null;
      mStopCatchingUp = false;
    }

    mTailerThread =
        new UfsJournalCheckpointThread(mMaster, this, mSuspendSequence + 1, mJournalSinks);
    mTailerThread.start();
    mSuspendSequence = -1;
    mSuspended = false;
  }

  /**
   * @return the quiet period for this journal
   */
  public long getQuietPeriodMs() {
    return mQuietPeriodMs;
  }

  /**
   * @param readIncompleteLogs whether the reader should read the latest incomplete log
   * @return a reader for reading from the start of the journal
   */
  public UfsJournalReader getReader(boolean readIncompleteLogs) {
    return new UfsJournalReader(this, readIncompleteLogs);
  }

  /**
   * @param checkpointSequenceNumber the next sequence number after the checkpoint
   * @return a writer for writing a checkpoint
   */
  public UfsJournalCheckpointWriter getCheckpointWriter(long checkpointSequenceNumber)
      throws IOException {
    return UfsJournalCheckpointWriter.create(this, checkpointSequenceNumber);
  }

  /**
   * @return the next sequence number to write
   */
  public long getNextSequenceNumberToWrite() {
    return writer().getNextSequenceNumber();
  }

  /**
   * @return the first log sequence number that hasn't yet been checkpointed
   */
  public long getNextSequenceNumberToCheckpoint() throws IOException {
    return UfsJournalSnapshot.getNextLogSequenceNumberToCheckpoint(this);
  }

  /**
   * @return whether the journal has been formatted
   */
  public boolean isFormatted() {
    try {
      UfsStatus[] files = mUfs.listStatus(mLocation.toString());
      if (files == null) {
        return false;
      }
      // Search for the format file.
      String filePrefix = Configuration.getString(PropertyKey.MASTER_FORMAT_FILE_PREFIX);
      return Arrays.stream(files).anyMatch(file -> file.getName().startsWith(filePrefix));
    } catch (IOException e) {
      return false;
    }
  }

  /**
   * @return true if the journal is allowed to be written to
   */
  public boolean isWritable() {
    return mState.get() == State.PRIMARY;
  }

  /**
   * Formats the journal.
   */
  public void format() throws IOException {
    URI location = getLocation();
    LOG.info("Formatting {}", location);
    if (mUfs.isDirectory(location.toString())) {
      for (UfsStatus status : Objects.requireNonNull(mUfs.listStatus(location.toString()))) {
        String childPath = URIUtils.appendPathOrDie(location, status.getName()).toString();
        if (status.isDirectory()
            && !mUfs.deleteDirectory(childPath, DeleteOptions.defaults().setRecursive(true))
            || status.isFile() && !mUfs.deleteFile(childPath)) {
          throw new IOException(String.format("Failed to delete %s", childPath));
        }
      }
    } else if (!mUfs.mkdirs(location.toString())) {
      throw new IOException(String.format("Failed to create %s", location));
    }

    // Create a breadcrumb that indicates that the journal folder has been formatted.
    UnderFileSystemUtils.touch(mUfs, URIUtils.appendPathOrDie(location,
        Configuration.getString(
            PropertyKey.MASTER_FORMAT_FILE_PREFIX) + System.currentTimeMillis())
        .toString());
  }

  /**
   * Creates a checkpoint in this ufs journal.
   */
  public synchronized void checkpoint() throws IOException {
    long nextSequenceNumber = getNextSequenceNumberToWrite();
    if (nextSequenceNumber == getNextSequenceNumberToCheckpoint()) {
      LOG.info("{}: No entries have been written since the last checkpoint.",
          mMaster.getName());
      return;
    }
    try (UfsJournalCheckpointWriter journalWriter
             = getCheckpointWriter(nextSequenceNumber)) {
      LOG.info("{}: Writing checkpoint [sequence number {}].",
          mMaster.getName(), nextSequenceNumber);
      mMaster.writeToCheckpoint(journalWriter);
      LOG.info("{}: Finished checkpoint [sequence number {}].",
          mMaster.getName(), nextSequenceNumber);
      mEntriesSinceLastCheckPoint = 0;
      mLastCheckPointTime = System.currentTimeMillis();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw new CancelledException(mMaster.getName() + ": Checkpoint is interrupted");
    }
  }

  /**
   * @return the state of the master process journal catchup
   */
  public synchronized UfsJournalCheckpointThread.CatchupState getCatchupState() {
    if (mTailerThread == null) {
      // tailer thread not active yet
      return UfsJournalCheckpointThread.CatchupState.NOT_STARTED;
    }
    return mTailerThread.getCatchupState();
  }

  /**
   * @return the log directory location
   */
  @VisibleForTesting
  public URI getLogDir() {
    return mLogDir;
  }

  /**
   * @return the checkpoint directory location
   */
  URI getCheckpointDir() {
    return mCheckpointDir;
  }

  /**
   * @return the temporary directory location
   */
  URI getTmpDir() {
    return mTmpDir;
  }

  /**
   * @return the under file system instance
   */
  UnderFileSystem getUfs() {
    return mUfs;
  }

  /**
   * Reads and applies all journal entries starting from the specified sequence number.
   *
   * @param nextSequenceNumber the sequence number to continue catching up from
   * @return the next sequence number after the final sequence number read
   */
  private synchronized long catchUp(long nextSequenceNumber) {
    return catchUp(nextSequenceNumber, -1);
  }

  /**
   * Reads and applies journal entries starting from the specified sequence number upto given limit.
   *
   * @param nextSequenceNumber the sequence number to continue catching up from
   * @param endSequenceNumber the inclusive sequence number to end catching up
   * @return the next sequence number after the final sequence number read
   */
  private long catchUp(long nextSequenceNumber, long endSequenceNumber) {
    JournalReader journalReader = new UfsJournalReader(this, nextSequenceNumber, true);
    try {
      return catchUp(journalReader, endSequenceNumber);
    } finally {
      try {
        journalReader.close();
      } catch (IOException e) {
        LOG.warn("Failed to close journal reader: {}", e.toString());
      }
    }
  }

  /**
   * This method only throws unchecked exceptions like {@code RuntimeException}.
   * If journal replay fails due to corrupted journal contents, this method will crash the master.
   * If the journal cannot be accessed due to IOException, this method will retry forever.
   * If an exception is thrown from this method that suggests an uncaught exception
   * or the master is failing over or shutting down.
   */
  private long catchUp(JournalReader journalReader, long limit) {
    RetryPolicy retry =
        ExponentialTimeBoundedRetry.builder()
            .withInitialSleep(Duration.ofSeconds(1))
            .withMaxSleep(Duration.ofSeconds(10))
            .withMaxDuration(Duration.ofDays(365))
            .build();
    while (true) {
      // Finish catching up, if reader is beyond given limit.
      if (limit != -1 && journalReader.getNextSequenceNumber() > limit) {
        return journalReader.getNextSequenceNumber();
      }
      if (mStopCatchingUp) {
        return journalReader.getNextSequenceNumber();
      }
      try {
        switch (journalReader.advance()) {
          case CHECKPOINT:
            mMaster.restoreFromCheckpoint(journalReader.getCheckpoint());
            break;
          case LOG:
            JournalEntry entry = journalReader.getEntry();
            try {
              if (!mMaster.processJournalEntry(entry)) {
                JournalUtils
                    .handleJournalReplayFailure(LOG, null, "%s: Unrecognized journal entry: %s",
                        mMaster.getName(), entry);
              } else {
                JournalUtils.sinkAppend(mJournalSinks, entry);
              }
            }  catch (Throwable t) {
              JournalUtils.handleJournalReplayFailure(LOG, t,
                    "%s: Failed to process journal entry %s", mMaster.getName(), entry);
            }
            break;
          default:
            return journalReader.getNextSequenceNumber();
        }
      } catch (IOException e) {
        LOG.warn("{}: Failed to read from journal: {}", mMaster.getName(), e);
        if (retry.attempt()) {
          continue;
        }
        throw new RuntimeException(
            String.format("%s: failed to catch up journal", mMaster.getName()), e);
      }
    }
  }

  @Override
  public String toString() {
    return "UfsJournal(" + mLocation + ")";
  }

  @Override
  public synchronized void close() {
    if (mState.get() == State.PRIMARY && mWriter != null) {
      LOG.info("Closing journal {}, state {} last journal location {}, next sequence number {}",
          this, mState, mWriter.currentLogName(), mWriter.getNextSequenceNumber());
    }
    if (mAsyncWriter != null) {
      mAsyncWriter.close();
      mAsyncWriter = null;
    }
    if (mWriter != null) {
      mWriter.close();
      mWriter = null;
    }
    if (mTailerThread != null) {
      try {
        // If the tailing thread has crashed before the close,
        // an exception will be thrown, containing what has originally caused the crash
        mTailerThread.awaitTermination(false);
        if (mState.get() == State.STANDBY) {
          LOG.info("Closing journal {}, state {}, next sequence number {}",
              this, this.mState, mTailerThread.getNextSequenceNumber());
        }
      } catch (Throwable t) {
        // We want to let the thread finish normally, however this call might throw if it already
        // finished exceptionally. We do not rethrow as we want the shutdown sequence to be smooth
        // (aka not throw exceptions).
        LOG.warn("exception caught when closing {}'s journal", mMaster.getName(), t);
      }
      mTailerThread = null;
    }
    mState.set(State.CLOSED);
  }

  /**
   * UFS implementation for {@link AbstractCatchupThread}.
   * This thread is only used on the standby master to catch up before taking a backup.
   */
  class UfsJournalCatchupThread extends AbstractCatchupThread {
    /** Where to start catching up. */
    private final long mCatchUpStartSequence;
    /** Where to end catching up. */
    private final long mCatchUpEndSequence;

    /**
     * Creates UFS catch-up thread for given range.
     *
     * @param start start sequence (inclusive)
     * @param end end sequence (inclusive)
     */
    public UfsJournalCatchupThread(long start, long end) {
      mCatchUpStartSequence = start;
      mCatchUpEndSequence = end;
      setName(String.format("ufs-catchup-thread-%s", mMaster.getName()));
    }

    @Override
    public void cancel() {
      // Used by catchup() to bail early.
      mStopCatchingUp = true;
    }

    @Override
    protected void runCatchup() {
      // Update suspended sequence after catch-up is finished.
      mSuspendSequence = catchUp(mCatchUpStartSequence, mCatchUpEndSequence) - 1;
    }
  }
}
