/*
 * 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.gobblin.compaction.mapreduce;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.URI;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.Job;
import org.joda.time.DateTime;
import org.joda.time.DateTimeZone;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Functions;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.base.Stopwatch;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.io.Closer;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;

import org.apache.gobblin.compaction.Compactor;
import org.apache.gobblin.compaction.dataset.Dataset;
import org.apache.gobblin.compaction.dataset.DatasetsFinder;
import org.apache.gobblin.compaction.dataset.TimeBasedSubDirDatasetsFinder;
import org.apache.gobblin.compaction.event.CompactionSlaEventHelper;
import org.apache.gobblin.compaction.listeners.CompactorCompletionListener;
import org.apache.gobblin.compaction.listeners.CompactorCompletionListenerFactory;
import org.apache.gobblin.compaction.listeners.CompactorListener;
import org.apache.gobblin.compaction.verify.DataCompletenessVerifier;
import org.apache.gobblin.compaction.verify.DataCompletenessVerifier.Results;
import org.apache.gobblin.configuration.ConfigurationKeys;
import org.apache.gobblin.configuration.State;
import org.apache.gobblin.metrics.GobblinMetrics;
import org.apache.gobblin.metrics.MetricReporterException;
import org.apache.gobblin.metrics.MultiReporterException;
import org.apache.gobblin.metrics.Tag;
import org.apache.gobblin.metrics.event.EventSubmitter;
import org.apache.gobblin.util.ClassAliasResolver;
import org.apache.gobblin.util.ClusterNameTags;
import org.apache.gobblin.util.DatasetFilterUtils;
import org.apache.gobblin.util.ExecutorsUtils;
import org.apache.gobblin.util.FileListUtils;
import org.apache.gobblin.util.HadoopUtils;
import org.apache.gobblin.util.recordcount.CompactionRecordCountProvider;
import org.apache.gobblin.util.recordcount.IngestionRecordCountProvider;
import org.apache.gobblin.util.reflection.GobblinConstructorUtils;

import static org.apache.gobblin.compaction.dataset.Dataset.DatasetState.COMPACTION_COMPLETE;
import static org.apache.gobblin.compaction.dataset.Dataset.DatasetState.GIVEN_UP;
import static org.apache.gobblin.compaction.dataset.Dataset.DatasetState.UNVERIFIED;
import static org.apache.gobblin.compaction.dataset.Dataset.DatasetState.VERIFIED;
import static org.apache.gobblin.compaction.mapreduce.MRCompactorJobRunner.Status.ABORTED;
import static org.apache.gobblin.compaction.mapreduce.MRCompactorJobRunner.Status.COMMITTED;

/**
 * MapReduce-based {@link org.apache.gobblin.compaction.Compactor}. Compaction will run on each qualified {@link Dataset}
 * under {@link #COMPACTION_INPUT_DIR}.
 *
 * @author Ziyang Liu
 * @deprecated Please use {@link org.apache.gobblin.compaction.mapreduce.MRCompactionTask}
 *  and {@link org.apache.gobblin.compaction.source.CompactionSource} to launch MR instead.
 *  The new way enjoys simpler logic to trigger the compaction flow and more reliable verification criteria,
 *  instead of using timestamp only before.
 */

@Deprecated
public class MRCompactor implements Compactor {

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

  public static final String COMPACTION_PREFIX = "compaction.";

  /**
   * Basic compaction properties.
   */
  public static final String COMPACTION_THREAD_POOL_SIZE = COMPACTION_PREFIX + "thread.pool.size";
  public static final int DEFAULT_COMPACTION_THREAD_POOL_SIZE = 30;
  public static final String COMPACTION_INPUT_DIR = COMPACTION_PREFIX + "input.dir";

  // The subdir name of input dataset paths, e.g., "hourly" in "/data/input/PasswordChangeEvent/hourly/2015/09/06".
  public static final String COMPACTION_INPUT_SUBDIR = COMPACTION_PREFIX + "input.subdir";
  public static final String DEFAULT_COMPACTION_INPUT_SUBDIR = "hourly";

  public static final String COMPACTION_DEST_DIR = COMPACTION_PREFIX + "dest.dir";

  // The subdir name of output dataset paths, e.g., "daily" in "/data/input/PasswordChangeEvent/daily/2015/09/06".
  public static final String COMPACTION_DEST_SUBDIR = COMPACTION_PREFIX + "dest.subdir";
  public static final String DEFAULT_COMPACTION_DEST_SUBDIR = "daily";

  // The output dir for compaction MR job, which will be moved to the final output dir for data publishing.
  public static final String COMPACTION_TMP_DEST_DIR = COMPACTION_PREFIX + "tmp.dest.dir";
  public static final String DEFAULT_COMPACTION_TMP_DEST_DIR = "/tmp/gobblin-compaction";
  public static final String COMPACTION_JOB_DIR = COMPACTION_PREFIX + "tmp.job.dir";
  public static final String COMPACTION_LATE_DIR_SUFFIX = "_late";

  public static final String COMPACTION_BLACKLIST = COMPACTION_PREFIX + "blacklist";
  public static final String COMPACTION_WHITELIST = COMPACTION_PREFIX + "whitelist";
  public static final String COMPACTION_HIGH_PRIORITY_TOPICS = COMPACTION_PREFIX + "high.priority.topics";
  public static final String COMPACTION_NORMAL_PRIORITY_TOPICS = COMPACTION_PREFIX + "normal.priority.topics";

  public static final String COMPACTION_JOB_RUNNER_CLASS = COMPACTION_PREFIX + "job.runner.class";
  public static final String DEFAULT_COMPACTION_JOB_RUNNER_CLASS =
      "org.apache.gobblin.compaction.mapreduce.avro.MRCompactorAvroKeyDedupJobRunner";
  public static final String COMPACTION_TIMEZONE = COMPACTION_PREFIX + "timezone";
  public static final String DEFAULT_COMPACTION_TIMEZONE = ConfigurationKeys.PST_TIMEZONE_NAME;
  public static final String COMPACTION_FILE_SYSTEM_URI = COMPACTION_PREFIX + "file.system.uri";
  public static final String COMPACTION_MR_JOB_TIMEOUT_MINUTES = COMPACTION_PREFIX + "mr.job.timeout.minutes";
  public static final long DEFAULT_COMPACTION_MR_JOB_TIMEOUT_MINUTES = Long.MAX_VALUE;

  // Dataset finder to find datasets for compaction.
  public static final String COMPACTION_DATASETS_FINDER = COMPACTION_PREFIX + "datasets.finder";
  public static final String DEFAULT_COMPACTION_DATASETS_FINDER = TimeBasedSubDirDatasetsFinder.class.getName();

  public static final String COMPACTION_DATASETS_MAX_COUNT = COMPACTION_PREFIX + "datasets.max.count";
  public static final double DEFUALT_COMPACTION_DATASETS_MAX_COUNT = 1000000000;

  // Rename source directories as a compaction complete indication
  // Compaction jobs using this completion mode can't share input sources
  public static final String COMPACTION_RENAME_SOURCE_DIR_ENABLED = COMPACTION_PREFIX + "rename.source.dir.enabled";
  public static final boolean DEFAULT_COMPACTION_RENAME_SOURCE_DIR_ENABLED = false;
  public static final String COMPACTION_RENAME_SOURCE_DIR_SUFFIX = "_COMPLETE";


  //The provider that provides event counts for the compaction input files.
  public static final String COMPACTION_INPUT_RECORD_COUNT_PROVIDER = COMPACTION_PREFIX + "input.record.count.provider";
  public static final String DEFAULT_COMPACTION_INPUT_RECORD_COUNT_PROVIDER =
      IngestionRecordCountProvider.class.getName();

  //The provider that provides event counts for the compaction output files.
  public static final String COMPACTION_OUTPUT_RECORD_COUNT_PROVIDER =
      COMPACTION_PREFIX + "output.record.count.provider";
  public static final String DEFAULT_COMPACTION_OUTPUT_RECORD_COUNT_PROVIDER =
      CompactionRecordCountProvider.class.getName();

  // If a dataset has already been compacted and new (late) data is found, whether recompact this dataset.
  public static final String COMPACTION_RECOMPACT_FROM_INPUT_FOR_LATE_DATA =
      COMPACTION_PREFIX + "recompact.from.input.for.late.data";
  public static final boolean DEFAULT_COMPACTION_RECOMPACT_FROM_INPUT_FOR_LATE_DATA = false;

  // The threshold of new(late) data that will trigger recompaction per dataset.
  // It follows the pattern DATASET_NAME_REGEX:THRESHOLD;DATASET_NAME_REGEX:THRESHOLD, e.g., A.*,B.*:0.2; C.*,D.*:0.3.
  // Dataset names that match A.* or B.* will have threshold 0.2. Dataset names that match C.* or D.* will have threshold 0.3.
  public static final String COMPACTION_LATEDATA_THRESHOLD_FOR_RECOMPACT_PER_DATASET =
      COMPACTION_PREFIX + "latedata.threshold.for.recompact.per.topic";
  public static final double DEFAULT_COMPACTION_LATEDATA_THRESHOLD_FOR_RECOMPACT_PER_DATASET = 1.0;

  // The threshold of new (late) files that will trigger compaction per dataset.
  // The trigger is based on the file numbers in the late output directory
  public static final String COMPACTION_LATEDATA_THRESHOLD_FILE_NUM =
      COMPACTION_PREFIX + "latedata.threshold.file.num";
  public static final int DEFAULT_COMPACTION_LATEDATA_THRESHOLD_FILE_NUM = 1000;

  // The threshold of new (late) files that will trigger compaction per dataset.
  // The trigger is based on how long the file has been in the late output directory.
  public static final String COMPACTION_LATEDATA_THRESHOLD_DURATION =
      COMPACTION_PREFIX + "latedata.threshold.duration";
  public static final String DEFAULT_COMPACTION_LATEDATA_THRESHOLD_DURATION = "24h";


  public static final String COMPACTION_RECOMPACT_CONDITION = COMPACTION_PREFIX + "recompact.condition";
  public static final String DEFAULT_COMPACTION_RECOMPACT_CONDITION = "RecompactBasedOnRatio";

  public static final String COMPACTION_RECOMPACT_COMBINE_CONDITIONS = COMPACTION_PREFIX + "recompact.combine.conditions";
  public static final String COMPACTION_RECOMPACT_COMBINE_CONDITIONS_OPERATION = COMPACTION_PREFIX + "recompact.combine.conditions.operation";
  public static final String DEFAULT_COMPACTION_RECOMPACT_COMBINE_CONDITIONS_OPERATION = "or";

  public static final String COMPACTION_COMPLETE_LISTERNER = COMPACTION_PREFIX + "complete.listener";
  public static final String DEFAULT_COMPACTION_COMPLETE_LISTERNER = "SimpleCompactorCompletionHook";

  // Whether the input data for the compaction is deduplicated.
  public static final String COMPACTION_INPUT_DEDUPLICATED = COMPACTION_PREFIX + "input.deduplicated";
  public static final boolean DEFAULT_COMPACTION_INPUT_DEDUPLICATED = false;

  // Whether the output of the compaction should be deduplicated.
  public static final String COMPACTION_OUTPUT_DEDUPLICATED = COMPACTION_PREFIX + "output.deduplicated";
  public static final boolean DEFAULT_COMPACTION_OUTPUT_DEDUPLICATED = true;

  public static final String COMPACTION_COMPLETENESS_VERIFICATION_PREFIX =
      COMPACTION_PREFIX + "completeness.verification.";

  public static final String COMPACTION_RECOMPACT_FROM_DEST_PATHS = COMPACTION_PREFIX + "recompact.from.dest.paths";
  public static final String COMPACTION_RECOMPACT_ALL_DATA = COMPACTION_PREFIX + "recompact.all.data";
  public static final boolean DEFAULT_COMPACTION_RECOMPACT_FROM_DEST_PATHS = false;
  public static final boolean DEFAULT_COMPACTION_RECOMPACT_ALL_DATA = true;

  /**
   * Configuration properties related to data completeness verification.
   */
  public static final String COMPACTION_COMPLETENESS_VERIFICATION_BLACKLIST =
      COMPACTION_COMPLETENESS_VERIFICATION_PREFIX + "blacklist";
  public static final String COMPACTION_COMPLETENESS_VERIFICATION_WHITELIST =
      COMPACTION_COMPLETENESS_VERIFICATION_PREFIX + "whitelist";
  public static final String COMPACTION_VERIFICATION_TIMEOUT_MINUTES =
      COMPACTION_COMPLETENESS_VERIFICATION_PREFIX + "timeout.minutes";
  public static final long DEFAULT_COMPACTION_VERIFICATION_TIMEOUT_MINUTES = 30;
  public static final String COMPACTION_COMPLETENESS_VERIFICATION_ENABLED =
      COMPACTION_COMPLETENESS_VERIFICATION_PREFIX + "enabled";
  public static final boolean DEFAULT_COMPACTION_COMPLETENESS_VERIFICATION_ENABLED = false;

  // Number of datasets to be passed to DataCompletenessVerifier together. By passing multiple datasets together,
  // some costs in DataCompletenessVerifier (e.g., submitting a SQL query) can be amortized.
  public static final String COMPACTION_COMPLETENESS_VERIFICATION_NUM_DATASETS_VERIFIED_TOGETHER =
      COMPACTION_COMPLETENESS_VERIFICATION_PREFIX + "num.datasets.verified.together";
  public static final int DEFAULT_COMPACTION_COMPLETENESS_VERIFICATION_NUM_DATASETS_VERIFIED_TOGETHER = 10;

  // Whether to compact and publish a datatset if its completeness cannot be verified.
  public static final String COMPACTION_COMPLETENESS_VERIFICATION_PUBLISH_DATA_IF_CANNOT_VERIFY =
      COMPACTION_COMPLETENESS_VERIFICATION_PREFIX + "publish.data.if.cannot.verify";
  public static final boolean DEFAULT_COMPACTION_COMPLETENESS_VERIFICATION_PUBLISH_DATA_IF_CANNOT_VERIFY = false;

  /**
   * Compaction configuration properties used internally.
   */
  public static final String COMPACTION_SHOULD_DEDUPLICATE = COMPACTION_PREFIX + "should.deduplicate";
  public static final String COMPACTION_JOB_DEST_PARTITION = COMPACTION_PREFIX + "job.dest.partition";
  public static final String COMPACTION_ENABLE_SUCCESS_FILE =
      COMPACTION_PREFIX + "fileoutputcommitter.marksuccessfuljobs";
  public static final String COMPACTION_JOB_LATE_DATA_MOVEMENT_TASK = COMPACTION_PREFIX + "job.late.data.movement.task";
  public static final String COMPACTION_JOB_LATE_DATA_FILES = COMPACTION_PREFIX + "job.late.data.files";
  public static final String COMPACTION_COMPLETE_FILE_NAME = "_COMPACTION_COMPLETE";
  public static final String COMPACTION_LATE_FILES_DIRECTORY = "late";
  public static final String COMPACTION_JARS = COMPACTION_PREFIX + "jars";
  public static final String COMPACTION_JAR_SUBDIR = "_gobblin_compaction_jars";
  public static final String COMPACTION_TRACKING_EVENTS_NAMESPACE = COMPACTION_PREFIX + "tracking.events";

  public static final String COMPACTION_INPUT_PATH_TIME = COMPACTION_PREFIX + "input.path.time";
  public static final String COMPACTION_FILE_EXTENSION =
      COMPACTION_PREFIX + "extension";

  private static final long COMPACTION_JOB_WAIT_INTERVAL_SECONDS = 10;
  private static final Map<Dataset, Job> RUNNING_MR_JOBS = Maps.newConcurrentMap();

  private final State state;
  private final List<? extends Tag<?>> tags;
  private final Configuration conf;
  private final String tmpOutputDir;
  private final FileSystem fs;
  private final JobRunnerExecutor jobExecutor;
  private final Set<Dataset> datasets;
  private final Map<Dataset, MRCompactorJobRunner> jobRunnables;
  private final Closer closer;
  private final Optional<DataCompletenessVerifier> verifier;
  private final Stopwatch stopwatch;
  private final GobblinMetrics gobblinMetrics;
  private final EventSubmitter eventSubmitter;
  private final Optional<CompactorListener> compactorListener;
  private final DateTime initilizeTime;
  private final long dataVerifTimeoutMinutes;
  private final long compactionTimeoutMinutes;
  private final boolean shouldVerifDataCompl;
  private final boolean shouldPublishDataIfCannotVerifyCompl;
  private final CompactorCompletionListener compactionCompleteListener;

  public MRCompactor(Properties props, List<? extends Tag<?>> tags, Optional<CompactorListener> compactorListener)
      throws IOException {
    this.state = new State();
    this.state.addAll(props);
    this.initilizeTime = getCurrentTime();
    this.tags = tags;
    this.conf = HadoopUtils.getConfFromState(this.state);
    this.tmpOutputDir = getTmpOutputDir();
    this.fs = getFileSystem();
    this.datasets = getDatasetsFinder().findDistinctDatasets();
    this.jobExecutor = createJobExecutor();
    this.jobRunnables = Maps.newConcurrentMap();
    this.closer = Closer.create();
    this.stopwatch = Stopwatch.createStarted();
    this.gobblinMetrics = initializeMetrics();
    this.eventSubmitter = new EventSubmitter.Builder(
        GobblinMetrics.get(this.state.getProp(ConfigurationKeys.JOB_NAME_KEY)).getMetricContext(),
        MRCompactor.COMPACTION_TRACKING_EVENTS_NAMESPACE).build();
    this.compactorListener = compactorListener;
    this.dataVerifTimeoutMinutes = getDataVerifTimeoutMinutes();
    this.compactionTimeoutMinutes = getCompactionTimeoutMinutes();
    this.shouldVerifDataCompl = shouldVerifyDataCompleteness();
    this.compactionCompleteListener = getCompactionCompleteListener();
    this.verifier =
        this.shouldVerifDataCompl ? Optional.of(this.closer.register(new DataCompletenessVerifier(this.state)))
            : Optional.<DataCompletenessVerifier> absent();
    this.shouldPublishDataIfCannotVerifyCompl = shouldPublishDataIfCannotVerifyCompl();
  }

  public DateTime getInitializeTime() {
    return this.initilizeTime;
  }

  private String getTmpOutputDir() {
    return this.state.getProp(COMPACTION_TMP_DEST_DIR, DEFAULT_COMPACTION_TMP_DEST_DIR);
  }

  private FileSystem getFileSystem() throws IOException {
    if (this.state.contains(COMPACTION_FILE_SYSTEM_URI)) {
      URI uri = URI.create(this.state.getProp(COMPACTION_FILE_SYSTEM_URI));
      return FileSystem.get(uri, this.conf);
    }
    return FileSystem.get(this.conf);
  }

  private DatasetsFinder getDatasetsFinder() {
    try {
      return (DatasetsFinder) Class
          .forName(this.state.getProp(COMPACTION_DATASETS_FINDER, DEFAULT_COMPACTION_DATASETS_FINDER))
          .getConstructor(State.class).newInstance(this.state);
    } catch (Exception e) {
      throw new RuntimeException("Failed to initiailize DatasetsFinder.", e);
    }
  }

  private DateTime getCurrentTime () {
    DateTimeZone timeZone = DateTimeZone
        .forID(this.state.getProp(MRCompactor.COMPACTION_TIMEZONE, MRCompactor.DEFAULT_COMPACTION_TIMEZONE));
    return new DateTime (timeZone);
  }

  private JobRunnerExecutor createJobExecutor() {
    int threadPoolSize = getThreadPoolSize();
    BlockingQueue<Runnable> queue = new PriorityBlockingQueue<>();
    return new JobRunnerExecutor(threadPoolSize, threadPoolSize, Long.MAX_VALUE, TimeUnit.NANOSECONDS, queue);
  }

  private int getThreadPoolSize() {
    return this.state.getPropAsInt(COMPACTION_THREAD_POOL_SIZE, DEFAULT_COMPACTION_THREAD_POOL_SIZE);
  }

  private GobblinMetrics initializeMetrics() {
    ImmutableList.Builder<Tag<?>> tags = ImmutableList.builder();
    tags.addAll(this.tags);
    tags.addAll(Tag.fromMap(ClusterNameTags.getClusterNameTags()));
    GobblinMetrics gobblinMetrics =
        GobblinMetrics.get(this.state.getProp(ConfigurationKeys.JOB_NAME_KEY), null, tags.build());
    try {
      gobblinMetrics.startMetricReporting(this.state.getProperties());
    } catch (MultiReporterException ex) {
      for (MetricReporterException e: ex.getExceptions()) {
        LOG.error("Failed to start {} {} reporter.", e.getSinkType().name(), e.getReporterType().name(), e);
      }
    }
    return gobblinMetrics;
  }

  @Override
  public void compact() throws IOException {
    try {
      copyDependencyJarsToHdfs();
      processDatasets();
      throwExceptionsIfAnyDatasetCompactionFailed();
      onCompactionCompletion();
    } catch (Throwable t) {

      // This throwable is logged here before propagated. Otherwise, if another throwable is thrown
      // in the finally-block, this throwable may be suppressed.
      LOG.error("Caught throwable during compaction", t);
      throw Throwables.propagate(t);
    } finally {
      try {
        shutdownExecutors();
        this.closer.close();
      } finally {
        deleteDependencyJars();
        this.gobblinMetrics.stopMetricsReporting();
      }
    }
  }

  private CompactorCompletionListener getCompactionCompleteListener () {
    ClassAliasResolver<CompactorCompletionListenerFactory> classAliasResolver = new ClassAliasResolver<>(CompactorCompletionListenerFactory.class);
    String listenerName= this.state.getProp(MRCompactor.COMPACTION_COMPLETE_LISTERNER,
        MRCompactor.DEFAULT_COMPACTION_COMPLETE_LISTERNER);
    try {
      CompactorCompletionListenerFactory factory = GobblinConstructorUtils.invokeFirstConstructor(
          classAliasResolver.resolveClass(listenerName), ImmutableList.of());
      return factory.createCompactorCompactionListener(this.state);
    } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException
        | ClassNotFoundException e) {
      throw new IllegalArgumentException(e);
    }
  }

  private void onCompactionCompletion() {
    this.compactionCompleteListener.onCompactionCompletion(this);
  }

  /**
   * Copy dependency jars from local fs to HDFS.
   */
  private void copyDependencyJarsToHdfs() throws IOException {
    if (!this.state.contains(ConfigurationKeys.JOB_JAR_FILES_KEY)) {
      return;
    }
    LocalFileSystem lfs = FileSystem.getLocal(this.conf);
    Path tmpJarFileDir = new Path(this.tmpOutputDir, "_gobblin_compaction_jars");
    this.state.setProp(COMPACTION_JARS, tmpJarFileDir.toString());
    this.fs.delete(tmpJarFileDir, true);
    for (String jarFile : this.state.getPropAsList(ConfigurationKeys.JOB_JAR_FILES_KEY)) {
      for (FileStatus status : lfs.globStatus(new Path(jarFile))) {
        Path tmpJarFile = new Path(this.fs.makeQualified(tmpJarFileDir), status.getPath().getName());
        this.fs.copyFromLocalFile(status.getPath(), tmpJarFile);
        LOG.info(String.format("%s will be added to classpath", tmpJarFile));
      }
    }
  }

  /**
   * Delete dependency jars from HDFS when job is done.
   */
  private void deleteDependencyJars() throws IllegalArgumentException, IOException {
    if (this.state.contains(COMPACTION_JARS)) {
      this.fs.delete(new Path(this.state.getProp(COMPACTION_JARS)), true);
    }
  }

  private void processDatasets() {
    createJobPropsForDatasets();
    processCompactionJobs();
  }

  /**
   * Create compaction job properties for {@link Dataset}s.
   */
  private void createJobPropsForDatasets() {
    final Set<Dataset> datasetsWithProps = Sets.newHashSet();
    for (Dataset dataset : this.datasets) {
      datasetsWithProps.addAll(createJobPropsForDataset(dataset));
    }

    this.datasets.clear();
    this.datasets.addAll(datasetsWithProps);
  }

  /**
   * Existing dataset in {@link #datasets} does not have job props.
   * Create compaction job properties for each given {@link Dataset}.
   * Update datasets based on the results of creating job props for them.
   */
  private List<Dataset> createJobPropsForDataset(Dataset dataset) {
    LOG.info("Creating compaction jobs for dataset " + dataset + " with priority " + dataset.priority());
    final MRCompactorJobPropCreator jobPropCreator = getJobPropCreator(dataset);
    List<Dataset> datasetsWithProps;
    try {
      datasetsWithProps = jobPropCreator.createJobProps();
    } catch (Throwable t) {
      // If a throwable is caught when creating job properties for a dataset, skip the topic and add the throwable
      // to the dataset.
      datasetsWithProps = ImmutableList.<Dataset> of(jobPropCreator.createFailedJobProps(t));
    }
    return datasetsWithProps;
  }

  /**
   * Get an instance of {@link MRCompactorJobPropCreator}.
   */
  MRCompactorJobPropCreator getJobPropCreator(Dataset dataset) {
    try {
      return new MRCompactorJobPropCreator.Builder().withDataset(dataset).withFileSystem(this.fs).withState(this.state)
          .build();
    } catch (Exception e) {
      throw new RuntimeException(e);
    }
  }

  public Set<Dataset> getDatasets() {
    return this.datasets;
  }

  private void processCompactionJobs() {
    if (this.shouldVerifDataCompl) {
      verifyDataCompleteness();
    } else {
      setAllDatasetStatesToVerified();
    }

    this.submitCompactionJobsAndWaitForCompletion();
  }

  private boolean shouldVerifyDataCompleteness() {
    return this.state.getPropAsBoolean(COMPACTION_COMPLETENESS_VERIFICATION_ENABLED,
        DEFAULT_COMPACTION_COMPLETENESS_VERIFICATION_ENABLED);
  }

  private void verifyDataCompleteness() {
    List<Pattern> blacklist =
        DatasetFilterUtils.getPatternList(this.state, COMPACTION_COMPLETENESS_VERIFICATION_BLACKLIST);
    List<Pattern> whitelist =
        DatasetFilterUtils.getPatternList(this.state, COMPACTION_COMPLETENESS_VERIFICATION_WHITELIST);
    int numDatasetsVerifiedTogether = getNumDatasetsVerifiedTogether();
    List<Dataset> datasetsToBeVerified = Lists.newArrayList();
    for (Dataset dataset : this.datasets) {
      if (dataset.state() != UNVERIFIED) {
        continue;
      }
      if (shouldVerifyCompletenessForDataset(dataset, blacklist, whitelist)) {
        datasetsToBeVerified.add(dataset);
        if (datasetsToBeVerified.size() >= numDatasetsVerifiedTogether) {
          ListenableFuture<Results> future = this.verifier.get().verify(datasetsToBeVerified);
          addCallback(datasetsToBeVerified, future);
          datasetsToBeVerified = Lists.newArrayList();
        }
      } else {
        dataset.setState(VERIFIED);
      }
    }

    if (!datasetsToBeVerified.isEmpty()) {
      ListenableFuture<Results> future = this.verifier.get().verify(datasetsToBeVerified);
      addCallback(datasetsToBeVerified, future);
    }
  }

  /**
   * A {@link Dataset} should be verified if its not already compacted, and it satisfies the blacklist and whitelist.
   */
  private boolean shouldVerifyCompletenessForDataset(Dataset dataset, List<Pattern> blacklist,
    List<Pattern> whitelist) {
    boolean renamingRequired = this.state.getPropAsBoolean(COMPACTION_RENAME_SOURCE_DIR_ENABLED, DEFAULT_COMPACTION_RENAME_SOURCE_DIR_ENABLED);

    LOG.info ("Should verify completeness with renaming source dir : " + renamingRequired);

    return !datasetAlreadyCompacted(this.fs, dataset, renamingRequired)
        && DatasetFilterUtils.survived(dataset.getName(), blacklist, whitelist);
  }

  /**
   * Get all the renamed directories from the given paths
   * They are deepest level containing directories whose name has a suffix {@link MRCompactor#COMPACTION_RENAME_SOURCE_DIR_SUFFIX}
   * Also each directory needs to contain at least one file so empty directories will be excluded from the result
   */
  public static Set<Path> getDeepestLevelRenamedDirsWithFileExistence (FileSystem fs, Set<Path> paths) throws IOException {
    Set<Path> renamedDirs = Sets.newHashSet();
    for (FileStatus fileStatus : FileListUtils.listFilesRecursively(fs, paths)) {
      if (fileStatus.getPath().getParent().toString().endsWith(MRCompactor.COMPACTION_RENAME_SOURCE_DIR_SUFFIX)) {
        renamedDirs.add(fileStatus.getPath().getParent());
      }
    }

    return renamedDirs;
  }

    /**
     * Get all the unrenamed directories from the given paths
     * They are deepest level containing directories whose name doesn't have a suffix {@link MRCompactor#COMPACTION_RENAME_SOURCE_DIR_SUFFIX}
     * Also each directory needs to contain at least one file so empty directories will be excluded from the result
     */
  public static Set<Path> getDeepestLevelUnrenamedDirsWithFileExistence (FileSystem fs, Set<Path> paths) throws IOException {
    Set<Path> unrenamed = Sets.newHashSet();
    for (FileStatus fileStatus : FileListUtils.listFilesRecursively(fs, paths)) {
      if (!fileStatus.getPath().getParent().toString().endsWith(MRCompactor.COMPACTION_RENAME_SOURCE_DIR_SUFFIX)) {
        unrenamed.add(fileStatus.getPath().getParent());
      }
    }

    return unrenamed;
  }

  /**
   * Rename all the source directories for a specific dataset
   */
  public static void renameSourceDirAsCompactionComplete (FileSystem fs, Dataset dataset) {
    try {
      for (Path path: dataset.getRenamePaths()) {
        Path newPath = new Path (path.getParent(), path.getName() + MRCompactor.COMPACTION_RENAME_SOURCE_DIR_SUFFIX);
        LOG.info("[{}] Renaming {} to {}", dataset.getDatasetName(), path, newPath);
        fs.rename(path, newPath);
      }
    } catch (Exception e) {
      LOG.error ("Rename input path failed", e);
    }
  }

  /**
   * A {@link Dataset} is considered already compacted if either condition is true:
   * 1) When completion file strategy is used, a compaction completion means there is a file named
   *    {@link MRCompactor#COMPACTION_COMPLETE_FILE_NAME} in its {@link Dataset#outputPath()}.
   * 2) When renaming source directory strategy is used, a compaction completion means source directories
   *    {@link Dataset#inputPaths()} contains at least one directory which has been renamed to something with
   *    {@link MRCompactor#COMPACTION_RENAME_SOURCE_DIR_SUFFIX}.
   */
  public static boolean datasetAlreadyCompacted(FileSystem fs, Dataset dataset, boolean renameSourceEnable) {
    if (renameSourceEnable) {
      return checkAlreadyCompactedBasedOnSourceDirName (fs, dataset);
    } else {
      return checkAlreadyCompactedBasedOnCompletionFile(fs, dataset);
    }
  }


  /**  When renaming source directory strategy is used, a compaction completion means source directories
   *    {@link Dataset#inputPaths()} contains at least one directory which has been renamed to something with
   *    {@link MRCompactor#COMPACTION_RENAME_SOURCE_DIR_SUFFIX}.
   */
  private static boolean checkAlreadyCompactedBasedOnSourceDirName (FileSystem fs, Dataset dataset) {
    try {
      Set<Path> renamedDirs = getDeepestLevelRenamedDirsWithFileExistence(fs, dataset.inputPaths());
      return !renamedDirs.isEmpty();
    } catch (IOException e) {
      LOG.error("Failed to get deepest directories from source", e);
      return false;
    }
  }

 /**
  *  When completion file strategy is used, a compaction completion means there is a file named
  *    {@link MRCompactor#COMPACTION_COMPLETE_FILE_NAME} in its {@link Dataset#outputPath()}.
  */
  private static boolean checkAlreadyCompactedBasedOnCompletionFile(FileSystem fs, Dataset dataset) {
    Path filePath = new Path(dataset.outputPath(), MRCompactor.COMPACTION_COMPLETE_FILE_NAME);
    try {
      return fs.exists(filePath);
    } catch (IOException e) {
      LOG.error("Failed to verify the existence of file " + filePath, e);
      return false;
    }
  }

  public static long readCompactionTimestamp(FileSystem fs, Path compactionOutputPath) throws IOException {
    Path completionFilePath = new Path(compactionOutputPath, COMPACTION_COMPLETE_FILE_NAME);
    try (FSDataInputStream completionFileStream = fs.open(completionFilePath)) {
      return completionFileStream.readLong();
    }
  }

  private void addCallback(final List<Dataset> datasetsToBeVerified, ListenableFuture<Results> future) {

    Futures.addCallback(future, new FutureCallback<Results>() {

      /**
       * On success, resubmit verification for the {@link Dataset}s that should be resubmitted
       * (i.e., verification didn't pass and it didn't timeout).
       */
      @Override
      public void onSuccess(Results results) {
        List<Dataset> datasetsToBeVerifiedAgain = Lists.newArrayList();
        for (Results.Result result : results) {
          Optional<MRCompactorJobRunner> jobRunner =
              Optional.fromNullable(MRCompactor.this.jobRunnables.get(result.dataset()));

          switch (result.status()) {
            case PASSED:
              LOG.info("Completeness verification for dataset " + result.dataset() + " passed.");
