content: /*
   * 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.util.URIUtils;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.MoreObjects;
  import com.google.common.base.Objects;
  import com.google.common.base.Preconditions;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.net.URI;
  import java.util.UUID;
  import javax.annotation.Nullable;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * A simple data structure that represents a journal file.
   * There are four types of files (SN is short for sequence number):
   * 1. Temporary file: the file name is a random UUID.
   * 2. Completed log file: [start SN]-[end SN] (the start SN is inclusive, the end SN is exclusive).
   * 3. Incomplete log file (a.k.a the current log): [start SN]-[LONG.MAX_VALUE] (the start SN is
   *    inclusive).
   * 4. Checkpoint file: [0]-[end SN] (the end SN is exclusive).
   *
   * This data structure implements {@link Comparable} such that journal files can be sorted by the
   * end SNs.
   *
   * Note that the natural ordering of the class may not necessarily imply equality of objects. The
   * {@link #compareTo(UfsJournalFile)} implementation should not be used to determine equality.
   */
  @ThreadSafe
  @VisibleForTesting
  public final class UfsJournalFile implements Comparable<UfsJournalFile> {
    private static final Logger LOG = LoggerFactory.getLogger(UfsJournalFile.class);
  
    /** The location of the file. */
    private final URI mLocation;
    /** The start journal log entry sequence number (inclusive). */
    private final long mStart;
    /** The end journal log entry sequence number (exclusive). */
    private final long mEnd;
    /**
     * Whether this is a checkpoint file (the temporary checkpoint file is not considered as a
     * checkpoint file here).
     */
    private final boolean mIsCheckpoint;
  
    /**
     * Creates a journal file.
     *
     * @param location the file location
     * @param start the start sequence number (inclusive)
     * @param end the end sequence number (exclusive)
     * @param isCheckpoint whether this is a committed checkpoint file
     */
    private UfsJournalFile(URI location, long start, long end, boolean isCheckpoint) {
      mLocation = location;
      mStart = start;
      mEnd = end;
      mIsCheckpoint = isCheckpoint;
    }
  
    /**
     * Creates a committed checkpoint file.
     *
     * @param location the file location
     * @param end the end sequence number (exclusive)
     * @return the file
     */
    static UfsJournalFile createCheckpointFile(URI location, long end) {
      return new UfsJournalFile(location, 0, end, true);
    }
  
    /**
     * Creates a journal log file.
     *
     * @param location the file location
     * @param start the start sequence number (inclusive)
     * @param end the end sequence number (exclusive)
     * @return the file
     */
    static UfsJournalFile createLogFile(URI location, long start, long end) {
      return new UfsJournalFile(location, start, end, false);
    }
  
    /**
     * Creates a temporary checkpoint file.
     *
     * @param location the file location
     * @return the file
     */
    static UfsJournalFile createTmpCheckpointFile(URI location) {
      return new UfsJournalFile(location, UfsJournal.UNKNOWN_SEQUENCE_NUMBER,
          UfsJournal.UNKNOWN_SEQUENCE_NUMBER, false);
    }
  
    /**
     * Encodes a checkpoint location under the checkpoint directory.
     *
     * @param journal the UFS journal instance
     * @param end the end sequence number (exclusive)
     * @return the location
     */
    static URI encodeCheckpointFileLocation(UfsJournal journal, long end) {
      String filename = String.format("0x%x-0x%x", 0, end);
      URI location = URIUtils.appendPathOrDie(journal.getCheckpointDir(), filename);
      return location;
    }
  
    /**
     * Encodes a log location under the log directory.
     *
     * @param journal the UFS journal instance
     * @param start the start sequence number (inclusive)
     * @param end the end sequence number (exclusive)
     * @return the location
     */
    static URI encodeLogFileLocation(UfsJournal journal, long start, long end) {
      String filename = String.format("0x%x-0x%x", start, end);
      URI location = URIUtils.appendPathOrDie(journal.getLogDir(), filename);
      return location;
    }
  
    /**
     * Encodes a temporary location under the temporary directory.
     *
     * @param journal the UFS journal instance*
     * @return the location
     */
    static URI encodeTemporaryCheckpointFileLocation(UfsJournal journal) {
      return URIUtils.appendPathOrDie(journal.getTmpDir(), UUID.randomUUID().toString());
    }
  
    /**
     * Decodes a checkpoint or a log file name into a {@link UfsJournalFile}.
     *
     * @param journal the UFS journal instance
     * @param filename the filename
     * @return the instance of {@link UfsJournalFile}, null if the file invalid
     */
    @Nullable
    static UfsJournalFile decodeLogFile(UfsJournal journal, String filename) {
      URI location = URIUtils.appendPathOrDie(journal.getLogDir(), filename);
      try {
        String[] parts = filename.split("-");
  
        // There can be temporary files in logs directory. Skip them.
        if (parts.length != 2) {
          return null;
        }
        long start = Long.decode(parts[0]);
        long end = Long.decode(parts[1]);
        return UfsJournalFile.createLogFile(location, start, end);
      } catch (IllegalStateException e) {
        LOG.error("Illegal journal file {}.", location);
        throw e;
      } catch (NumberFormatException e) {
        // There can be temporary files (e.g. created for rename).
        return null;
      }
    }
  
    /**
     * Decodes a checkpoint file name into a {@link UfsJournalFile}.
     *
     * @param journal the UFS journal instance
     * @param filename the filename
     * @return the instance of {@link UfsJournalFile}, null if the file invalid
     */
    @Nullable
    static UfsJournalFile decodeCheckpointFile(UfsJournal journal, String filename) {
      URI location = URIUtils.appendPathOrDie(journal.getCheckpointDir(), filename);
      try {
        String[] parts = filename.split("-");
  
        // There can be temporary files in logs directory. Skip them.
        if (parts.length != 2) {
          return null;
        }
        long start = Long.decode(parts[0]);
        long end = Long.decode(parts[1]);
  
        Preconditions.checkState(start == 0);
        return UfsJournalFile.createCheckpointFile(location, end);
      } catch (IllegalStateException e) {
        LOG.error("Illegal journal file {}.", location);
        throw e;
      } catch (NumberFormatException e) {
        // There can be temporary files (e.g. created for rename).
        return null;
      }
    }
  
    /**
     * Decodes a temporary checkpoint file name into a {@link UfsJournalFile}.
     *
     * @param journal the UFS journal instance
     * @param filename the temporary checkpoint file name
     * @return the instance of {@link UfsJournalFile}
     */
    static UfsJournalFile decodeTemporaryCheckpointFile(UfsJournal journal, String filename) {
      URI location = URIUtils.appendPathOrDie(journal.getTmpDir(), filename);
      return UfsJournalFile.createTmpCheckpointFile(location);
    }
  
    /**
     * @return the file location
     */
    @VisibleForTesting
    public URI getLocation() {
      return mLocation;
    }
  
    /**
     * @return the start sequence number (inclusive)
     */
    long getStart() {
      return mStart;
    }
  
    /**
     * @return the end sequence number (exclusive)
     */
    long getEnd() {
      return mEnd;
    }
  
    /**
     * @return whether it is a committed checkpoint file
     */
    boolean isCheckpoint() {
      return mIsCheckpoint;
    }
  
    /**
     * @return whether it is a completed log
     */
    boolean isCompletedLog() {
      return !isCheckpoint() && mEnd != UfsJournal.UNKNOWN_SEQUENCE_NUMBER;
    }
  
    /**
     * @return whether it is incomplete log (a.k.a. the current log)
     */
    boolean isIncompleteLog() {
      return mStart != UfsJournal.UNKNOWN_SEQUENCE_NUMBER
          && mEnd == UfsJournal.UNKNOWN_SEQUENCE_NUMBER;
    }
  
    /**
     * @return whether it is a temporary checkpoint
     */
    boolean isTmpCheckpoint() {
      return mStart == UfsJournal.UNKNOWN_SEQUENCE_NUMBER
          && mEnd == UfsJournal.UNKNOWN_SEQUENCE_NUMBER;
    }
  
    @Override
    public String toString() {
      return MoreObjects.toStringHelper(this).add("location", mLocation).add("start", mStart)
          .add("end", mEnd).add("isCheckpoint", mIsCheckpoint).toString();
    }
  
    @Override
    public int compareTo(UfsJournalFile other) {
      long diff = mEnd - other.mEnd;
      if (diff < 0) {
        return -1;
      } else if (diff == 0) {
        return 0;
      } else {
        return 1;
      }
    }
  
    @Override
    public boolean equals(Object o) {
      if (o == null) {
        return false;
      }
  
      if (!(o instanceof UfsJournalFile)) {
        return false;
      }
  
      UfsJournalFile other = (UfsJournalFile) o;
  
      return mLocation.equals(other.mLocation)
          && mIsCheckpoint == other.mIsCheckpoint
          && mStart == other.mStart
          && mEnd == other.mEnd;
    }
  
    @Override
    public int hashCode() {
      return Objects.hashCode(mLocation, mIsCheckpoint, mStart, mEnd);
    }
  }
  
  