/*
 * 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.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.JournalClosedException;
import alluxio.master.journal.JournalWriter;
import alluxio.proto.journal.Journal.JournalEntry;
import alluxio.util.FormatUtils;

import com.google.common.base.Preconditions;
import org.apache.ratis.protocol.Message;
import org.apache.ratis.protocol.RaftClientReply;
import org.apache.ratis.thirdparty.com.google.protobuf.UnsafeByteOperations;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Objects;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import javax.annotation.concurrent.NotThreadSafe;

/**
 * Class for writing entries to the Raft journal. Written entries are aggregated until flush is
 * called, then they are submitted as a single unit.
 */
@NotThreadSafe
public class RaftJournalWriter implements JournalWriter {
  private static final Logger LOG = LoggerFactory.getLogger(RaftJournalWriter.class);
  // How long to wait for a response from the cluster before giving up and trying again.
  private static final long MASTER_EMBEDDED_JOURNAL_WRITE_TIMEOUT =
      Configuration.getMs(PropertyKey.MASTER_EMBEDDED_JOURNAL_WRITE_TIMEOUT);
  private static final long MASTER_EMBEDDED_JOURNAL_ENTRY_SIZE_MAX =
      Configuration.getBytes(PropertyKey.MASTER_EMBEDDED_JOURNAL_ENTRY_SIZE_MAX);
  // journal entry size max is the hard limit set by underlying ratis
  // we use a smaller value to guarantee we don't pass the hard limit
  private static final long FLUSH_BATCH_SIZE = MASTER_EMBEDDED_JOURNAL_ENTRY_SIZE_MAX / 3;

  private final AtomicLong mNextSequenceNumberToWrite;
  private final AtomicLong mLastSubmittedSequenceNumber = new AtomicLong(-1);
  private final AtomicLong mLastCommittedSequenceNumber = new AtomicLong(-1);

  private final RaftJournalAppender mClient;

  private volatile boolean mClosed = false;
  private JournalEntry.Builder mJournalEntryBuilder; // gets build across successive writes
  private final AtomicLong mCurrentJournalEntrySize = new AtomicLong(0);

  /**
   * @param nextSequenceNumberToWrite the sequence number for the writer to begin writing at
   * @param client client for writing entries to the journal; the constructed journal writer owns
   *               this client and is responsible for closing it
   */
  public RaftJournalWriter(long nextSequenceNumberToWrite, RaftJournalAppender client) {
    LOG.debug("Journal writer created starting at SN#{}", nextSequenceNumberToWrite);
    mNextSequenceNumberToWrite = new AtomicLong(nextSequenceNumberToWrite);
    mClient = Objects.requireNonNull(client);
  }

  @Override
  public void write(JournalEntry entry) throws IOException, JournalClosedException {
    if (mClosed) {
      throw new JournalClosedException("Cannot write to journal. Journal writer has been closed");
    }
    Preconditions.checkState(entry.getAllFields().size() <= 2,
        "Raft journal entries should never set multiple fields, but found %s", entry);
    if (mCurrentJournalEntrySize.get() > FLUSH_BATCH_SIZE) {
      flush();
    }
    if (mJournalEntryBuilder == null) {
      mJournalEntryBuilder = JournalEntry.newBuilder();
      mCurrentJournalEntrySize.set(0);
    }
    LOG.trace("Writing entry {}: {}", mNextSequenceNumberToWrite, entry);
    mJournalEntryBuilder.addJournalEntries(entry.toBuilder()
        .setSequenceNumber(mNextSequenceNumberToWrite.getAndIncrement()).build());
    long size = entry.getSerializedSize();
    if (size > MASTER_EMBEDDED_JOURNAL_ENTRY_SIZE_MAX) {
      LOG.error("Journal entry size ({}) is bigger than the max allowed size ({}) defined by {}",
          FormatUtils.getSizeFromBytes(size),
          FormatUtils.getSizeFromBytes(MASTER_EMBEDDED_JOURNAL_ENTRY_SIZE_MAX),
          PropertyKey.MASTER_EMBEDDED_JOURNAL_ENTRY_SIZE_MAX.getName());
    }
    mCurrentJournalEntrySize.addAndGet(size);
  }

  @Override
  public void flush() throws IOException, JournalClosedException {
    if (mClosed) {
      throw new JournalClosedException("Cannot flush. Journal writer has been closed");
    }
    if (mJournalEntryBuilder != null) {
      long flushSN = mNextSequenceNumberToWrite.get() - 1;
      try {
        // It is ok to submit the same entries multiple times because we de-duplicate by sequence
        // number when applying them. This could happen if submit fails and we re-submit the same
        // entry on retry.
        JournalEntry entry = mJournalEntryBuilder.build();
        Message message = Message.valueOf(UnsafeByteOperations.unsafeWrap(entry.toByteArray()));
        mLastSubmittedSequenceNumber.set(flushSN);
        LOG.trace("Flushing entry {} ({})", entry, message);
        RaftClientReply reply = mClient.sendAsync(message)
            .get(MASTER_EMBEDDED_JOURNAL_WRITE_TIMEOUT, TimeUnit.MILLISECONDS);
        if (reply.getException() != null) {
          throw reply.getException();
        }
        mLastCommittedSequenceNumber.set(flushSN);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new IOException(e);
      } catch (ExecutionException e) {
        throw new IOException(e.getCause());
      } catch (TimeoutException e) {
        throw new IOException(String.format(
            "Timed out after waiting %s milliseconds for journal entries to be processed",
            MASTER_EMBEDDED_JOURNAL_WRITE_TIMEOUT), e);
      }
      mJournalEntryBuilder = null;
    }
  }

  @Override
  public void close() {
    if (mClosed) {
      return;
    }
    mClosed = true;
    LOG.info("Closing journal writer. Last sequence numbers written/submitted/committed: {}/{}/{}",
        mNextSequenceNumberToWrite.get() - 1, mLastSubmittedSequenceNumber.get(),
        mLastCommittedSequenceNumber.get());
    closeClient();
  }

  /**
   * @return the next sequence number that will be written by this writer
   */
  public long getNextSequenceNumberToWrite() {
    return mNextSequenceNumberToWrite.get();
  }

  private void closeClient() {
    try {
      mClient.close();
    } catch (IOException e) {
      LOG.warn("Failed to close raft client: {}", e.toString());
    }
  }
}
