/*
 * 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.accumulo.tserver.compactions;

import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;

import org.apache.accumulo.core.spi.compaction.CompactionExecutorId;
import org.apache.accumulo.core.spi.compaction.CompactionJob;
import org.apache.accumulo.core.spi.compaction.CompactionServiceId;
import org.apache.accumulo.core.util.NamingThreadFactory;
import org.apache.accumulo.core.util.compaction.CompactionJobPrioritizer;
import org.apache.accumulo.core.util.ratelimit.RateLimiter;
import org.apache.accumulo.tserver.metrics.CompactionExecutorsMetrics;
import org.apache.htrace.wrappers.TraceExecutorService;
import org.apache.htrace.wrappers.TraceRunnable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;

public class CompactionExecutor {

  private static final Logger log = LoggerFactory.getLogger(CompactionExecutor.class);

  private PriorityBlockingQueue<Runnable> queue;
  private ExecutorService executor;
  private final CompactionExecutorId ceid;
  private AtomicLong cancelCount = new AtomicLong();
  private ThreadPoolExecutor rawExecutor;

  // This exist to provide an accurate count of queued compactions for metrics. The PriorityQueue is
  // not used because it size may be off because it contains cancelled compactions. The collection
  // below should not contain cancelled compactions. A conncurrent set was not used because those do
  // not have constant time size operations.
  private Set<CompactionTask> queuedTask = Collections.synchronizedSet(new HashSet<>());

  private AutoCloseable metricCloser;

  private RateLimiter readLimiter;
  private RateLimiter writeLimiter;

  private class CompactionTask extends SubmittedJob implements Runnable {

    private AtomicReference<Status> status = new AtomicReference<>(Status.QUEUED);
    private Compactable compactable;
    private CompactionServiceId csid;
    private Consumer<Compactable> completionCallback;

    public CompactionTask(CompactionJob job, Compactable compactable, CompactionServiceId csid,
        Consumer<Compactable> completionCallback) {
      super(job);
      this.compactable = compactable;
      this.csid = csid;
      this.completionCallback = completionCallback;
      queuedTask.add(this);
    }

    @Override
    public void run() {

      try {
        if (status.compareAndSet(Status.QUEUED, Status.RUNNING)) {
          queuedTask.remove(this);
          compactable.compact(csid, getJob(), readLimiter, writeLimiter);
          completionCallback.accept(compactable);
        }
      } catch (Exception e) {
        log.warn("Compaction failed for {} on {}", compactable.getExtent(), getJob(), e);
        status.compareAndSet(Status.RUNNING, Status.FAILED);
      } finally {
        status.compareAndSet(Status.RUNNING, Status.COMPLETE);
      }
    }

    @Override
    public Status getStatus() {
      return status.get();
    }

    @Override
    public boolean cancel(Status expectedStatus) {

      boolean canceled = false;

      if (expectedStatus == Status.QUEUED) {
        canceled = status.compareAndSet(expectedStatus, Status.CANCELED);
      }

      if (canceled)
        queuedTask.remove(this);

      if (canceled && cancelCount.incrementAndGet() % 1024 == 0) {
        // Need to occasionally clean the queue which could have canceled task with low priority that
        // hang around. Avoid cleaning the queue every time something is canceled as that could be
        // expensive.
        queue.removeIf(runnable -> ((CompactionTask) runnable).getStatus() == Status.CANCELED);
      }

      return canceled;
    }

  }

  private static CompactionJob getJob(Runnable r) {
    if (r instanceof TraceRunnable) {
      return getJob(((TraceRunnable) r).getRunnable());
    }

    if (r instanceof CompactionTask) {
      return ((CompactionTask) r).getJob();
    }

    throw new IllegalArgumentException("Unknown runnable type " + r.getClass().getName());
  }

  CompactionExecutor(CompactionExecutorId ceid, int threads, CompactionExecutorsMetrics ceMetrics,
      RateLimiter readLimiter, RateLimiter writeLimiter) {
    this.ceid = ceid;
    var comparator =
        Comparator.comparing(CompactionExecutor::getJob, CompactionJobPrioritizer.JOB_COMPARATOR);

    queue = new PriorityBlockingQueue<Runnable>(100, comparator);

    ThreadPoolExecutor tp = new ThreadPoolExecutor(threads, threads, 60, TimeUnit.SECONDS, queue,
        new NamingThreadFactory("compaction." + ceid));
    tp.allowCoreThreadTimeOut(true);

    rawExecutor = tp;
    executor = new TraceExecutorService(tp);

    metricCloser = ceMetrics.addExecutor(ceid, () -> tp.getActiveCount(), () -> queuedTask.size());

    this.readLimiter = readLimiter;
    this.writeLimiter = writeLimiter;

    log.debug("Created compaction executor {} with {} threads", ceid, threads);
  }

  public SubmittedJob submit(CompactionServiceId csid, CompactionJob job, Compactable compactable,
      Consumer<Compactable> completionCallback) {
    Preconditions.checkArgument(job.getExecutor().equals(ceid));
    var ctask = new CompactionTask(job, compactable, csid, completionCallback);
    executor.execute(ctask);
    return ctask;
  }

  public void setThreads(int numThreads) {

    var tp = rawExecutor;

    int coreSize = tp.getCorePoolSize();

    if (numThreads < coreSize) {
      tp.setCorePoolSize(numThreads);
      tp.setMaximumPoolSize(numThreads);
    } else if (numThreads > coreSize) {
      tp.setMaximumPoolSize(numThreads);
      tp.setCorePoolSize(numThreads);
    }

    if (numThreads != coreSize) {
      log.debug("Adjusted compaction executor {} threads from {} to {}", ceid, coreSize,
          numThreads);
    }
  }

  public int getCompactionsRunning() {
    return rawExecutor.getActiveCount();
  }

  public int getCompactionsQueued() {
    return queuedTask.size();
  }

  public void stop() {
    executor.shutdownNow();
    log.debug("Stopped compaction executor {}", ceid);
    try {
      metricCloser.close();
    } catch (Exception e) {
      log.warn("Failed to close metrics {}", ceid, e);
    }
  }
}

@@ -4532,21 +4532,16 @@
 priority
- that
 %0A       
@@ -4543,16 +4543,21 @@
       //
+ that
  hang ar
@@ -4633,19 +4633,16 @@
 at could
- be
 %0A       
@@ -4644,16 +4644,19 @@
       //
+ be
  expensi

fix comment #1609