/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hbase.replication.regionserver;

import static org.apache.hadoop.hbase.replication.ReplicationUtils.OFFSET_UPDATE_INTERVAL_MS_KEY;
import static org.apache.hadoop.hbase.replication.ReplicationUtils.OFFSET_UPDATE_SIZE_THRESHOLD_KEY;
import static org.apache.hadoop.hbase.replication.ReplicationUtils.getAdaptiveTimeout;
import static org.apache.hadoop.hbase.replication.ReplicationUtils.sleepForRetries;

import java.io.IOException;
import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.wal.WAL.Entry;
import org.apache.hadoop.hbase.wal.WALEdit;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.BulkLoadDescriptor;
import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos.StoreDescriptor;

/**
 * This thread reads entries from a queue and ships them. Entries are placed onto the queue by
 * ReplicationSourceWALReaderThread
 */
@InterfaceAudience.Private
public class ReplicationSourceShipper extends Thread {
  private static final Logger LOG = LoggerFactory.getLogger(ReplicationSourceShipper.class);

  // Hold the state of a replication worker thread
  public enum WorkerState {
    RUNNING,
    STOPPED,
    FINISHED, // The worker is done processing a queue
  }

  private final Configuration conf;
  final String walGroupId;
  private final ReplicationSource source;

  // Last position in the log that we sent to ZooKeeper
  // It will be accessed by the stats thread so make it volatile
  private volatile long currentPosition = -1;
  // Path of the current log
  private Path currentPath;
  // Current state of the worker thread
  private volatile WorkerState state;
  final ReplicationSourceWALReader entryReader;

  // How long should we sleep for each retry
  private final long sleepForRetries;
  // Maximum number of retries before taking bold actions
  private final int maxRetriesMultiplier;
  private final int DEFAULT_TIMEOUT = 20000;
  private final int getEntriesTimeout;
  private final int shipEditsTimeout;
  private long accumulatedSizeSinceLastUpdate = 0L;
  private long lastOffsetUpdateTime = EnvironmentEdgeManager.currentTime();
  private final long offsetUpdateIntervalMs;
  private final long offsetUpdateSizeThresholdBytes;
  private WALEntryBatch lastShippedBatch;

  private static final long DEFAULT_OFFSET_UPDATE_INTERVAL_MS = Long.MAX_VALUE;
  private static final long DEFAULT_OFFSET_UPDATE_SIZE_THRESHOLD = -1L;

  public ReplicationSourceShipper(Configuration conf, String walGroupId, ReplicationSource source,
    ReplicationSourceWALReader walReader) {
    this.conf = conf;
    this.walGroupId = walGroupId;
    this.source = source;
    this.entryReader = walReader;
    // 1 second
    this.sleepForRetries = this.conf.getLong("replication.source.sleepforretries", 1000);
    // 5 minutes @ 1 sec per
    this.maxRetriesMultiplier = this.conf.getInt("replication.source.maxretriesmultiplier", 300);
    // 20 seconds
    this.getEntriesTimeout =
      this.conf.getInt("replication.source.getEntries.timeout", DEFAULT_TIMEOUT);
    this.shipEditsTimeout = this.conf.getInt(HConstants.REPLICATION_SOURCE_SHIPEDITS_TIMEOUT,
      HConstants.REPLICATION_SOURCE_SHIPEDITS_TIMEOUT_DFAULT);
    this.offsetUpdateIntervalMs =
      conf.getLong(OFFSET_UPDATE_INTERVAL_MS_KEY, DEFAULT_OFFSET_UPDATE_INTERVAL_MS);
    this.offsetUpdateSizeThresholdBytes =
      conf.getLong(OFFSET_UPDATE_SIZE_THRESHOLD_KEY, DEFAULT_OFFSET_UPDATE_SIZE_THRESHOLD);
  }

  @Override
  public final void run() {
    setWorkerState(WorkerState.RUNNING);
    LOG.info("Running ReplicationSourceShipper Thread for wal group: {}", this.walGroupId);
    // Loop until we close down
    while (isActive()) {
      // Sleep until replication is enabled again
      if (!source.isPeerEnabled()) {
        // The peer enabled check is in memory, not expensive, so do not need to increase the
        // sleep interval as it may cause a long lag when we enable the peer.
        sleepForRetries("Replication is disabled", sleepForRetries, 1, maxRetriesMultiplier);
        continue;
      }
      try {
        // check time-based offset persistence
        if (shouldPersistLogPosition()) {
          persistLogPosition();
        }

        long pollTimeout = getEntriesTimeout;
        if (offsetUpdateIntervalMs != Long.MAX_VALUE) {
          long elapsed = EnvironmentEdgeManager.currentTime() - lastOffsetUpdateTime;
          long remaining = offsetUpdateIntervalMs - elapsed;
          if (remaining > 0) {
            pollTimeout = Math.min(getEntriesTimeout, remaining);
          }
        }
        WALEntryBatch entryBatch = entryReader.poll(pollTimeout);
        LOG.debug("Shipper from source {} got entry batch from reader: {}", source.getQueueId(),
          entryBatch);

        if (entryBatch == null) {
          continue;
        }
        // the NO_MORE_DATA instance has no path so do not call shipEdits
        if (entryBatch == WALEntryBatch.NO_MORE_DATA) {
          noMoreData();
        } else {
          shipEdits(entryBatch);
        }
      } catch (InterruptedException | ReplicationRuntimeException e) {
        LOG.warn("Interrupted while waiting for next replication entry batch", e);
        Thread.currentThread().interrupt();
      } catch (Exception e) {
        // During source shutdown / peer removal we can see interrupted IOEs
        // from replication queue updates. Do not restart in this case.
        if (!source.isSourceActive() || isInterrupted() || !source.isPeerEnabled()) {
          LOG.info("Ignoring persist failure during shutdown for walGroupId={}", walGroupId, e);
          break;
        }
        LOG.error("Shipper {} failed to persist offset, restarting", walGroupId, e);
        abortAndRestart(e);
        return;
      }
    }

    // If the worker exits run loop without finishing its task, mark it as stopped.
    if (!isFinished()) {
      try {
        persistLogPosition();
      } catch (Exception e) {
        LOG.error("Failed persisting final offset for walGroupId={}", walGroupId, e);
      }
      setWorkerState(WorkerState.STOPPED);
    } else {
      source.removeWorker(this);
      postFinish();
    }
  }

  private void noMoreData() throws IOException {
    // Flush any outstanding replication offset before marking the queue finished.
    // Offset persistence may be delayed by size/time thresholds, so ensure the
    // latest replicated position is stored before transitioning to FINISHED state.
    persistLogPosition();

    if (source.isRecovered()) {
      LOG.debug("Finished recovering queue for group {} of peer {}", walGroupId,
        source.getQueueId());
      source.getSourceMetrics().incrCompletedRecoveryQueue();
    } else {
      LOG.debug("Finished queue for group {} of peer {}", walGroupId, source.getQueueId());
    }
    setWorkerState(WorkerState.FINISHED);
  }

  // To be implemented by recovered shipper
  protected void postFinish() {
  }

  /**
   * Do the shipping logic
   */
  void shipEdits(WALEntryBatch entryBatch) throws IOException {
    List<Entry> entries = entryBatch.getWalEntries();
    int sleepMultiplier = 0;
    int currentSize = (int) entryBatch.getHeapSize();
    MetricsSource metrics = source.getSourceMetrics();
    if (metrics != null && !entries.isEmpty()) {
      metrics.setTimeStampNextToReplicate(entries.get(entries.size() - 1).getKey().getWriteTime());
    }
    if (entries.isEmpty()) {
      // empty batch may mean WAL boundary advancement
      lastShippedBatch = entryBatch;
      persistLogPosition();
      return;
    }
    while (isActive()) {
      try {
        try {
          source.tryThrottle(currentSize);
        } catch (InterruptedException e) {
          LOG.debug("Interrupted while sleeping for throttling control");
          Thread.currentThread().interrupt();
          // current thread might be interrupted to terminate
          // directly go back to while() for confirm this
          continue;
        }
        // create replicateContext here, so the entries can be GC'd upon return from this call
        // stack
        ReplicationEndpoint.ReplicateContext replicateContext =
          new ReplicationEndpoint.ReplicateContext();
        replicateContext.setEntries(entries).setSize(currentSize);
        replicateContext.setWalGroupId(walGroupId);
        replicateContext.setTimeout(getAdaptiveTimeout(this.shipEditsTimeout, sleepMultiplier));

        long startTimeNs = System.nanoTime();
        // send the edits to the endpoint. Will block until the edits are shipped and acknowledged
        boolean replicated = source.getReplicationEndpoint().replicate(replicateContext);
        long endTimeNs = System.nanoTime();

        if (!replicated) {
          continue;
        } else {
          sleepMultiplier = Math.max(sleepMultiplier - 1, 0);
        }
        // Clean up hfile references
        for (Entry entry : entries) {
          cleanUpHFileRefs(entry.getEdit());
          LOG.trace("shipped entry {}: ", entry);
        }

        // offsets totalBufferUsed by deducting shipped batchSize (excludes bulk load size)
        // this sizeExcludeBulkLoad has to use same calculation that when calling
        // acquireBufferQuota() in ReplicationSourceWALReader because they maintain
        // same variable: totalBufferUsed
        source.postShipEdits(entries, entryBatch.getUsedBufferSize());
        // FIXME check relationship between wal group and overall
        source.getSourceMetrics().shipBatch(entryBatch.getNbOperations(), currentSize,
          entryBatch.getNbHFiles());
        source.getSourceMetrics().setAgeOfLastShippedOp(
          entries.get(entries.size() - 1).getKey().getWriteTime(), walGroupId);
        source.getSourceMetrics().updateTableLevelMetrics(entryBatch.getWalEntriesWithSize());

        if (LOG.isTraceEnabled()) {
          LOG.debug("Replicated {} entries or {} operations in {} ms", entries.size(),
            entryBatch.getNbOperations(), (endTimeNs - startTimeNs) / 1000000);
        }
        break;
      } catch (Exception ex) {
        source.getSourceMetrics().incrementFailedBatches();
        LOG.warn("{} threw unknown exception:",
          source.getReplicationEndpoint().getClass().getName(), ex);
        if (
          sleepForRetries("ReplicationEndpoint threw exception", sleepForRetries, sleepMultiplier,
            maxRetriesMultiplier)
        ) {
          sleepMultiplier++;
        }
      }
    }

    accumulatedSizeSinceLastUpdate += currentSize;
    lastShippedBatch = entryBatch;
    if (shouldPersistLogPosition()) {
      persistLogPosition();
    }
  }

  private boolean shouldPersistLogPosition() {
    LOG.debug(
      "Persist decision: accumulatedSizeSinceLastUpdate={} threshold={} elapsed={} interval={}",
      accumulatedSizeSinceLastUpdate, offsetUpdateSizeThresholdBytes,
      EnvironmentEdgeManager.currentTime() - lastOffsetUpdateTime, offsetUpdateIntervalMs);
    if (lastShippedBatch == null) {
      return false;
    }
    // Default behaviour to update offset immediately after replicate()
    if (offsetUpdateSizeThresholdBytes == -1 && offsetUpdateIntervalMs == Long.MAX_VALUE) {
      return true;
    }

    return (accumulatedSizeSinceLastUpdate >= offsetUpdateSizeThresholdBytes)
      || (EnvironmentEdgeManager.currentTime() - lastOffsetUpdateTime >= offsetUpdateIntervalMs);
  }

  void persistLogPosition() throws IOException {
    if (lastShippedBatch == null) {
      return;
    }

    if (!source.isSourceActive() || isInterrupted() || !source.isPeerEnabled()) {
      LOG.debug("Skip persistLogPosition for inactive/stopping source");
      return;
    }

    ReplicationEndpoint endpoint = source.getReplicationEndpoint();
    if (endpoint != null) {
      endpoint.beforePersistingReplicationOffset();
    }

    // Log and clean up WAL logs
    updateLogPosition(lastShippedBatch);
    accumulatedSizeSinceLastUpdate = 0;
    lastShippedBatch = null;
    lastOffsetUpdateTime = EnvironmentEdgeManager.currentTime();
  }

  void cleanUpHFileRefs(WALEdit edit) throws IOException {
    String peerId = source.getPeerId();
    if (peerId.contains("-")) {
      // peerClusterZnode will be in the form peerId + "-" + rsZNode.
      // A peerId will not have "-" in its name, see HBASE-11394
      peerId = peerId.split("-")[0];
    }
    List<Cell> cells = edit.getCells();
    int totalCells = cells.size();
    for (int i = 0; i < totalCells; i++) {
      Cell cell = cells.get(i);
      if (CellUtil.matchingQualifier(cell, WALEdit.BULK_LOAD)) {
        BulkLoadDescriptor bld = WALEdit.getBulkLoadDescriptor(cell);
        List<StoreDescriptor> stores = bld.getStoresList();
        int totalStores = stores.size();
        for (int j = 0; j < totalStores; j++) {
          List<String> storeFileList = stores.get(j).getStoreFileList();
          source.getSourceManager().cleanUpHFileRefs(peerId, storeFileList);
          source.getSourceMetrics().decrSizeOfHFileRefsQueue(storeFileList.size());
        }
      }
    }
  }

  private boolean updateLogPosition(WALEntryBatch batch) {
    boolean updated = false;
    // if end of file is true, then the logPositionAndCleanOldLogs method will remove the file
    // record on zk, so let's call it. The last wal position maybe zero if end of file is true and
    // there is no entry in the batch. It is OK because that the queue storage will ignore the zero
    // position and the file will be removed soon in cleanOldLogs.
    if (
      batch.isEndOfFile() || !batch.getLastWalPath().equals(currentPath)
        || batch.getLastWalPosition() != currentPosition
    ) {
      source.logPositionAndCleanOldLogs(batch);
      updated = true;
    }
    // if end of file is true, then we can just skip to the next file in queue.
    // the only exception is for recovered queue, if we reach the end of the queue, then there will
    // no more files so here the currentPath may be null.
    if (batch.isEndOfFile()) {
      currentPath = entryReader.getCurrentPath();
      currentPosition = 0L;
    } else {
      currentPath = batch.getLastWalPath();
      currentPosition = batch.getLastWalPosition();
    }
    return updated;
  }

  public void startup(UncaughtExceptionHandler handler) {
    String name = Thread.currentThread().getName();
    Threads.setDaemonThreadRunning(this,
      name + ".replicationSource.shipper" + walGroupId + "," + source.getQueueId(),
      handler::uncaughtException);
  }

  Path getCurrentPath() {
    return entryReader.getCurrentPath();
  }

  long getCurrentPosition() {
    return currentPosition;
  }

  protected boolean isActive() {
    return source.isSourceActive() && state == WorkerState.RUNNING && !isInterrupted();
  }

  protected final void setWorkerState(WorkerState state) {
    this.state = state;
  }

  void stopWorker() {
    setWorkerState(WorkerState.STOPPED);
  }

  public boolean isFinished() {
    return state == WorkerState.FINISHED;
  }

  /**
   * Attempts to properly update <code>ReplicationSourceManager.totalBufferUser</code>, in case
   * there were unprocessed entries batched by the reader to the shipper, but the shipper didn't
   * manage to ship those because the replication source is being terminated. In that case, it
   * iterates through the batched entries and decrease the pending entries size from
   * <code>ReplicationSourceManager.totalBufferUser</code>
   * <p/>
   * <b>NOTES</b> 1) This method should only be called upon replication source termination. It
   * blocks waiting for both shipper and reader threads termination, to make sure no race conditions
   * when updating <code>ReplicationSourceManager.totalBufferUser</code>. 2) It <b>does not</b>
   * attempt to terminate reader and shipper threads. Those <b>must</b> have been triggered
   * interruption/termination prior to calling this method.
   */
  void clearWALEntryBatch() {
    long timeout = EnvironmentEdgeManager.currentTime() + this.shipEditsTimeout;
    while (this.isAlive() || this.entryReader.isAlive()) {
      try {
        if (EnvironmentEdgeManager.currentTime() >= timeout) {
          LOG.warn("Shipper clearWALEntryBatch method timed out while waiting reader/shipper "
            + "thread to stop. Not cleaning buffer usage. PeerId: {}; Shipper alive: {}; Reader "
            + "alive: {}", this.source.getPeerId(), this.isAlive(), this.entryReader.isAlive());
          return;
        } else {
          // Wait both shipper and reader threads to stop
          Thread.sleep(this.sleepForRetries);
        }
      } catch (InterruptedException e) {
        LOG.warn("{} Interrupted while waiting {} to stop on clearWALEntryBatch. "
          + "Not cleaning buffer usage: {}", this.source.getPeerId(), this.getName(), e);
        return;
      }
    }
    long totalReleasedBytes = 0;
    while (true) {
      WALEntryBatch batch = entryReader.entryBatchQueue.poll();
      if (batch == null) {
        break;
      }
      totalReleasedBytes += source.getSourceManager().releaseWALEntryBatchBufferQuota(batch);
    }
    if (LOG.isTraceEnabled()) {
      LOG.trace("Decrementing totalBufferUsed by {}B while stopping Replication WAL Readers.",
        totalReleasedBytes);
    }
  }

  long getSleepForRetries() {
    return sleepForRetries;
  }

  // Restart worker so replication resumes from last persisted offset.
  void abortAndRestart(Throwable cause) {
    LOG.warn("Restarting shipper for walGroupId={}", walGroupId, cause);
    if (!source.isSourceActive() || !source.isPeerEnabled() || isInterrupted()) {
      LOG.warn("abortAndRestart SKIPPED walGroupId={}, thread={}", walGroupId,
        Thread.currentThread().getName());
      return;
    }
    setWorkerState(WorkerState.STOPPED);
    source.restartShipper(walGroupId, this);
  }
}