alluxio-master-2.x/core/server/common/src/main/java/alluxio/master/journal/FileSystemMergeJournalContext.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;

import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.status.UnavailableException;
import alluxio.proto.journal.Journal.JournalEntry;

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

import java.util.List;
import javax.annotation.concurrent.ThreadSafe;

/**
 * Context for merging journal entries together for a wrapped journal context.
 *
 * This is used so that we can combine several journal entries into one using a merge
 * function. This helps us resolve the inconsistency between primary and standby and
 * decrease the chance for a standby to see intermediate state of a file system operation.
 *
 * This journal context is made thread-safe because metadata sync creates worker threads to fetch
 * metadata and reuses the same journal context.
 */
@ThreadSafe
public class FileSystemMergeJournalContext implements JournalContext {
  // It will log a warning if the number of buffered journal entries exceed 100
  private static final int MAX_LOGGING_ENTRIES
      = Configuration.getInt(
          PropertyKey.MASTER_MERGE_JOURNAL_CONTEXT_NUM_ENTRIES_LOGGING_THRESHOLD);

  private static final Logger LOG = LoggerFactory.getLogger(FileSystemMergeJournalContext.class);

  protected final JournalContext mJournalContext;
  protected final JournalEntryMerger mJournalEntryMerger;

  /**
   * Constructs a {@link FileSystemMergeJournalContext}.
   * @param journalContext the journal context to wrap
   * @param journalEntryMerger the merger which merges multiple journal entries into one
   */
  public FileSystemMergeJournalContext(JournalContext journalContext,
                             JournalEntryMerger journalEntryMerger) {
    Preconditions.checkNotNull(journalContext, "journalContext is null");
    Preconditions.checkNotNull(journalEntryMerger, "mergeOperator is null");
    mJournalContext = journalContext;
    mJournalEntryMerger = journalEntryMerger;
  }

  /**
   * Adds the new journal entry into the journal entry merger.
   * If the size of the outstanding journal entries are bigger than the threshold,
   * we will flush them immediately into the async journal writer.
   * This is used to prevent a recursive file system operation adding
   * too many journal entries without being flushed and taking huge memory.
   * Though it might create inconsistency between primary and standby if
   * the primary crashes between two flushes.
   * Tune PropertyKey.MASTER_MERGE_JOURNAL_CONTEXT_MAX_ENTRIES to trade off
   * @param entry the {@link JournalEntry} to append to the journal
   */
  @Override
  public synchronized void append(JournalEntry entry) {
    mJournalEntryMerger.add(entry);
    List<JournalEntry> journalEntries = mJournalEntryMerger.getMergedJournalEntries();
    if (journalEntries.size() >= MAX_LOGGING_ENTRIES) {
      LOG.warn("MergeJournalContext has " + journalEntries.size()
          + " entries, over the limit of " + MAX_LOGGING_ENTRIES
          + ", forcefully merging journal entries and add them to the async journal writer"
          + "\n Journal Entry: \n"
          + entry, new Exception("MergeJournalContext Stacktrace:"));
      appendMergedJournals();
    }
  }

  @Override
  public synchronized void close() throws UnavailableException {
    try {
      appendMergedJournals();
    } finally {
      mJournalContext.close();
    }
  }

  /**
   * Merges all journals and then flushes them.
   * The journal writer will commit these journals synchronously.
   */
  @Override
  public synchronized void flush() throws UnavailableException {
    // Skip flushing the journal if no journal entries to append
    if (appendMergedJournals()) {
      mJournalContext.flush();
    }
  }

  protected synchronized boolean appendMergedJournals() {
    List<JournalEntry> journalEntries = mJournalEntryMerger.getMergedJournalEntries();
    if (journalEntries.size() == 0) {
      return false;
    }

    journalEntries.forEach(mJournalContext::append);
    mJournalEntryMerger.clear();
    return true;
  }

  /**
   * @return the underlying journal context
   */
  public JournalContext getUnderlyingJournalContext() {
    return mJournalContext;
  }
}
