/**
 * 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.tez.runtime.library.api;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.classification.InterfaceAudience.Public;
import org.apache.hadoop.classification.InterfaceStability.Evolving;
import org.apache.hadoop.classification.InterfaceStability.Unstable;
import org.apache.hadoop.conf.Configuration;
import org.apache.tez.common.annotation.ConfigurationClass;
import org.apache.tez.common.annotation.ConfigurationProperty;
import org.apache.tez.runtime.library.conf.OrderedPartitionedKVOutputConfig.SorterImpl;

/**
 * Meant for user configurable job properties.
 * <p/>
 * Note for developers: Whenever a new key is added to this file, it must also be added to the set of
 * known tezRuntimeKeys.
 * @see <a href="../../../../../../configs/TezRuntimeConfiguration.html">Detailed Configuration Information</a>
 * @see <a href="../../../../../configs/tez-runtime-default-template.xml">XML-based Config Template</a>
 */

// TODO EVENTUALLY A description for each property.
@Public
@Evolving
@ConfigurationClass(templateFileName = "tez-runtime-default-template.xml")
public class TezRuntimeConfiguration {

  private static final String TEZ_RUNTIME_PREFIX = "tez.runtime.";

  private static final Set<String> tezRuntimeKeys = new HashSet<String>();
  private static Set<String> umnodifiableTezRuntimeKeySet;
  private static final Set<String> otherKeys = new HashSet<String>();
  private static Set<String> unmodifiableOtherKeySet;
  private static Configuration defaultConf = new Configuration(false);
  private static final Map<String, String> tezRuntimeConfMap = new HashMap<String, String>();
  private static final Map<String, String> otherConfMap = new HashMap<String, String>();

  /**
   * Prefixes from Hadoop configuration which are allowed.
   */
  private static final List<String> allowedPrefixes = new ArrayList<String>();
  private static List<String> unmodifiableAllowedPrefixes;


  static {
    Configuration.addDeprecation("tez.runtime.sort.threads",
        TezRuntimeConfiguration.TEZ_RUNTIME_PIPELINED_SORTER_SORT_THREADS);
  }

  /**
   * Configuration key to enable/disable IFile readahead.
   */
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_IFILE_READAHEAD = TEZ_RUNTIME_PREFIX +
      "ifile.readahead";
  public static final boolean TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT = true;

  /**
   * Configuration key to set the IFile readahead length in bytes.
   */
  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_IFILE_READAHEAD_BYTES = TEZ_RUNTIME_PREFIX +
      "ifile.readahead.bytes";
  public static final int TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT =
      4 * 1024 * 1024;

  public static final int TEZ_RUNTIME_IFILE_BUFFER_SIZE_DEFAULT = -1;

  /**
   * This is copy of io.file.buffer.size from Hadoop, which is used in several places such
   * as compression codecs, buffer sizes in IFile, while fetching etc.
   * Variable exists so that it can be referenced, instead of using the string name directly.
   */
  public static final String TEZ_RUNTIME_IO_FILE_BUFFER_SIZE = "io.file.buffer.size";


  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_IO_SORT_FACTOR = TEZ_RUNTIME_PREFIX +
      "io.sort.factor";
  public static final int TEZ_RUNTIME_IO_SORT_FACTOR_DEFAULT = 100;


  @ConfigurationProperty(type = "float")
  public static final String TEZ_RUNTIME_SORT_SPILL_PERCENT = TEZ_RUNTIME_PREFIX +
      "sort.spill.percent";
  public static final float TEZ_RUNTIME_SORT_SPILL_PERCENT_DEFAULT = 0.8f;


  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_IO_SORT_MB = TEZ_RUNTIME_PREFIX + "io.sort.mb";
  public static final int TEZ_RUNTIME_IO_SORT_MB_DEFAULT = 100;


  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES = TEZ_RUNTIME_PREFIX +
      "index.cache.memory.limit.bytes";
  public static final int TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES_DEFAULT =
      1024 * 1024;


  // TODO Use the default value
  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_COMBINE_MIN_SPILLS = TEZ_RUNTIME_PREFIX +
      "combine.min.spills";
  public static final int TEZ_RUNTIME_COMBINE_MIN_SPILLS_DEFAULT = 3;

  /**
   * Tries to allocate @link{#TEZ_RUNTIME_IO_SORT_MB} in chunks specified in
   * this parameter.
   */
  @ConfigurationProperty(type = "integer")
  public static final String
      TEZ_RUNTIME_PIPELINED_SORTER_MIN_BLOCK_SIZE_IN_MB = TEZ_RUNTIME_PREFIX +
      "pipelined.sorter.min-block.size.in.mb";
  public static final int
      TEZ_RUNTIME_PIPELINED_SORTER_MIN_BLOCK_SIZE_IN_MB_DEFAULT = 2000;

  /**
   * Setting this to true would enable sorter
   * to auto-allocate memory on need basis in progressive fashion.
   *
   * Setting to false would allocate all available memory during
   * initialization of sorter. In such cases,@link{#TEZ_RUNTIME_PIPELINED_SORTER_MIN_BLOCK_SIZE_IN_MB}
   * would be honored and memory specified in @link{#TEZ_RUNTIME_IO_SORT_MB}
   * would be initialized upfront.
   */
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_PIPELINED_SORTER_LAZY_ALLOCATE_MEMORY = TEZ_RUNTIME_PREFIX +
      "pipelined.sorter.lazy-allocate.memory";
  public static final boolean
      TEZ_RUNTIME_PIPELINED_SORTER_LAZY_ALLOCATE_MEMORY_DEFAULT = false;

  /**
   * String value.
   * Which sorter implementation to use.
   * Valid values:
   *    - LEGACY
   *    - PIPELINED ( default )
   *    {@link org.apache.tez.runtime.library.conf.OrderedPartitionedKVOutputConfig.SorterImpl}
   */
  @ConfigurationProperty
  public static final String TEZ_RUNTIME_SORTER_CLASS = TEZ_RUNTIME_PREFIX +
      "sorter.class";
  public static final String TEZ_RUNTIME_SORTER_CLASS_DEFAULT = SorterImpl.PIPELINED.name();

  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_PIPELINED_SORTER_SORT_THREADS = TEZ_RUNTIME_PREFIX +
      "pipelined.sorter.sort.threads";
  public static final int TEZ_RUNTIME_PIPELINED_SORTER_SORT_THREADS_DEFAULT = 2;

  /**
   * Integer value. Percentage of buffer to be filled before we spill to disk. Default value is 0,
   * which will spill for every buffer.
   */
  @ConfigurationProperty(type="int")
  public static final String TEZ_RUNTIME_UNORDERED_PARTITIONED_KVWRITER_BUFFER_MERGE_PERCENT =
      TEZ_RUNTIME_PREFIX + "unordered-partitioned-kvwriter.buffer-merge-percent";
  public static final int TEZ_RUNTIME_UNORDERED_PARTITIONED_KVWRITER_BUFFER_MERGE_PERCENT_DEFAULT =
      0;

  /**
   * Report partition statistics (e.g better scheduling in ShuffleVertexManager). TEZ-2496
   * This can be enabled/disabled at vertex level.
   * {@link org.apache.tez.runtime.library.api.TezRuntimeConfiguration.ReportPartitionStats}
   * defines the list of values that can be specified.
   * TODO TEZ-3303 Given ShuffleVertexManager doesn't consume precise stats
   * yet. So do not set the value to "precise" yet when ShuffleVertexManager is used.
   */
  @ConfigurationProperty
  public static final String TEZ_RUNTIME_REPORT_PARTITION_STATS =
      TEZ_RUNTIME_PREFIX + "report.partition.stats";
  public static final String TEZ_RUNTIME_REPORT_PARTITION_STATS_DEFAULT =
      ReportPartitionStats.MEMORY_OPTIMIZED.getType();

  /**
   * Size of the buffer to use if not writing directly to disk.
   */
  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB = TEZ_RUNTIME_PREFIX +
      "unordered.output.buffer.size-mb";
  public static final int TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB_DEFAULT = 100;

  /**
   * Maximum size for individual buffers used in the UnsortedPartitionedOutput.
   * This is only meant to be used by unit tests for now.
   */
  @Private
  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES =
      TEZ_RUNTIME_PREFIX +
          "unordered.output.max-per-buffer.size-bytes";

  /**
   * Specifies a partitioner class, which is used in Tez Runtime components
   * like OnFileSortedOutput
   */
  @ConfigurationProperty
  public static final String TEZ_RUNTIME_PARTITIONER_CLASS =
      TEZ_RUNTIME_PREFIX + "partitioner.class";

  /**
   * Specifies a combiner class (primarily for Shuffle)
   */
  @ConfigurationProperty
  public static final String TEZ_RUNTIME_COMBINER_CLASS = TEZ_RUNTIME_PREFIX + "combiner.class";

  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES = TEZ_RUNTIME_PREFIX +
      "shuffle.parallel.copies";
  public static final int TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES_DEFAULT = 20;

  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT = TEZ_RUNTIME_PREFIX +
      "shuffle.fetch.failures.limit";
  public static final int TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT_DEFAULT = 5;

  /**
   * Specifies in milliseconds the maximum delay a penalized host can have before being retried,
   * defaults to 10 minutes.
   */
  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_HOST_PENALTY_TIME_LIMIT_MS = TEZ_RUNTIME_PREFIX +
      "shuffle.host.penalty.time.limit";
  public static final int TEZ_RUNTIME_SHUFFLE_HOST_PENALTY_TIME_LIMIT_MS_DEFAULT = 600000;

  @Private
  @Unstable
  @ConfigurationProperty(type = "integer")
  /**
   * Expert setting made available only for debugging. Do not change it. Sets
   * the number of retries before giving up on downloading from source
   * attempt by consumer. Code internally handles the threshold if set to -1.
   */
  public static final String
      TEZ_RUNTIME_SHUFFLE_SOURCE_ATTEMPT_ABORT_LIMIT =
      TEZ_RUNTIME_PREFIX + "shuffle.src-attempt.abort.limit";
  public static final int
      TEZ_RUNTIME_SHUFFLE_SOURCE_ATTEMPT_ABORT_LIMIT_DEFAULT = -1;

  @Private
  @Unstable
  @ConfigurationProperty(type = "float")
  /**
   * Expert setting made available only for debugging. Do not change it. Setting
   * to determine if failures happened across a percentage of nodes. This
   * helps in determining if the consumer has to be restarted on continuous
   * failures. Setting it to lower value can make consumer restarts more
   * aggressive on failures.
   */
  public static final String
      TEZ_RUNTIME_SHUFFLE_ACCEPTABLE_HOST_FETCH_FAILURE_FRACTION =
      TEZ_RUNTIME_PREFIX + "shuffle.acceptable.host-fetch.failure.fraction";
  public static final float
      TEZ_RUNTIME_SHUFFLE_ACCEPTABLE_HOST_FETCH_FAILURE_FRACTION_DEFAULT = 0.2f;

  @Private
  @Unstable
  @ConfigurationProperty(type = "integer")
  /**
   * Expert setting made available only for debugging. Do not change it. Setting
   * to determine if the consumer has to be restarted on continuous
   * failures across nodes. Used along with {@link
   * TEZ_RUNTIME_SHUFFLE_ACCEPTABLE_HOST_FETCH_FAILURE_FRACTION}.
   */
  public static final String
      TEZ_RUNTIME_SHUFFLE_MIN_FAILURES_PER_HOST =
      TEZ_RUNTIME_PREFIX + "shuffle.min.failures.per.host";
  public static final int TEZ_RUNTIME_SHUFFLE_MIN_FAILURES_PER_HOST_DEFAULT = 4;

  @Private
  @Unstable
  @ConfigurationProperty(type = "float")
  /**
   * Expert setting made available only for debugging. Do not change it.
   * Maximum percentage of time (compared to overall progress), the fetcher is
   * allowed before concluding that it is stalled.
   */
  public static final String TEZ_RUNTIME_SHUFFLE_MAX_STALL_TIME_FRACTION =
      TEZ_RUNTIME_PREFIX + "shuffle.max.stall.time.fraction";
  public static final float
      TEZ_RUNTIME_SHUFFLE_MAX_STALL_TIME_FRACTION_DEFAULT = 0.5f;

  @Private
  @Unstable
  @ConfigurationProperty(type = "float")
  /**
   * Expert setting made available only for debugging. Do not change it.
   * Fraction to determine whether the shuffle has progressed enough or not
   * If it has not progressed enough, it could be qualified for the consumer.
   */
  public static final String
      TEZ_RUNTIME_SHUFFLE_MIN_REQUIRED_PROGRESS_FRACTION =
      TEZ_RUNTIME_PREFIX + "shuffle.min.required.progress.fraction";
  public static final float
      TEZ_RUNTIME_SHUFFLE_MIN_REQUIRED_PROGRESS_FRACTION_DEFAULT = 0.5f;

  @Private
  @Unstable
  @ConfigurationProperty(type = "float")
  /**
   * Expert setting made available only for debugging. Do not change it.
   * Provides threshold for determining whether fetching has to be marked
   * unhealthy based on the ratio of (failures/(failures+completed))
   */
  public static final String
      TEZ_RUNTIME_SHUFFLE_MAX_ALLOWED_FAILED_FETCH_ATTEMPT_FRACTION =
      TEZ_RUNTIME_PREFIX + "shuffle.max.allowed.failed.fetch.fraction";
  public static final float
      TEZ_RUNTIME_SHUFFLE_MAX_ALLOWED_FAILED_FETCH_ATTEMPT_FRACTION_DEFAULT = 0.5f;

  @Private
  @Unstable
  @ConfigurationProperty(type = "boolean")
  /**
   * Expert setting made available only for debugging. Do not change it.
   * Provides threshold for determining whether fetching has to be marked
   * unhealthy based on the ratio of (failures/(failures+completed))
   */
  public static final String
      TEZ_RUNTIME_SHUFFLE_FAILED_CHECK_SINCE_LAST_COMPLETION =
      TEZ_RUNTIME_PREFIX + "shuffle.failed.check.since-last.completion";
  public static final boolean
      TEZ_RUNTIME_SHUFFLE_FAILED_CHECK_SINCE_LAST_COMPLETION_DEFAULT = true;


  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE =
      TEZ_RUNTIME_PREFIX +
          "shuffle.fetch.max.task.output.at.once";
  public final static int TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE_DEFAULT
      = 20;

  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR = TEZ_RUNTIME_PREFIX +
      "shuffle.notify.readerror";
  public static final boolean TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR_DEFAULT = true;

  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT = TEZ_RUNTIME_PREFIX +
      "shuffle.connect.timeout";
  public static final int TEZ_RUNTIME_SHUFFLE_STALLED_COPY_TIMEOUT_DEFAULT =
      3 * 60 * 1000;

  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED = TEZ_RUNTIME_PREFIX +
      "shuffle.keep-alive.enabled";
  public static final boolean TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED_DEFAULT = false;

  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS = TEZ_RUNTIME_PREFIX +
      "shuffle.keep-alive.max.connections";
  public static final int TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS_DEFAULT = 20;

  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT =
      TEZ_RUNTIME_PREFIX + "shuffle.read.timeout";
  public final static int TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT_DEFAULT =
      3 * 60 * 1000;

  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE =
      TEZ_RUNTIME_PREFIX + "shuffle.buffersize";
  public final static int TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE_DEFAULT =
      8 * 1024;

  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP = TEZ_RUNTIME_PREFIX +
      "shuffle.use.async.http";
  public static final boolean TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP_DEFAULT = false;

  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_SHUFFLE_ENABLE_SSL = TEZ_RUNTIME_PREFIX +
      "shuffle.ssl.enable";
  public static final boolean TEZ_RUNTIME_SHUFFLE_ENABLE_SSL_DEFAULT = false;

  /**
   * Controls verification of data checksums when fetching data directly to
   * disk. Enabling verification allows the fetcher to detect corrupted data
   * and report the failure against the upstream task before the data reaches
   * the Processor and causes the fetching task to fail.
   */
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM =
      TEZ_RUNTIME_PREFIX + "shuffle.fetch.verify-disk-checksum";
  public static final boolean TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM_DEFAULT = true;

  @ConfigurationProperty(type = "float")
  public static final String TEZ_RUNTIME_SHUFFLE_FETCH_BUFFER_PERCENT = TEZ_RUNTIME_PREFIX +
      "shuffle.fetch.buffer.percent";
  public static final float TEZ_RUNTIME_SHUFFLE_FETCH_BUFFER_PERCENT_DEFAULT =
      0.90f;

  @ConfigurationProperty(type = "float")
  public static final String TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT = TEZ_RUNTIME_PREFIX +
      "shuffle.memory.limit.percent";
  public static final float TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT_DEFAULT =
      0.25f;

  // Rename to fraction
  @ConfigurationProperty(type = "float")
  public static final String TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT = TEZ_RUNTIME_PREFIX +
      "shuffle.merge.percent";
  public static final float TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT_DEFAULT = 0.90f;

  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_SHUFFLE_MEMTOMEM_SEGMENTS = TEZ_RUNTIME_PREFIX +
      "shuffle.memory-to-memory.segments";

  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_SHUFFLE_ENABLE_MEMTOMEM = TEZ_RUNTIME_PREFIX +
      "shuffle.memory-to-memory.enable";
  public static final boolean TEZ_RUNTIME_SHUFFLE_ENABLE_MEMTOMEM_DEFAULT =
      false;

  @Private
  @Unstable
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL = TEZ_RUNTIME_PREFIX +
      "shuffle.fetcher.use-shared-pool";
  public static final boolean TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL_DEFAULT = false;

  @ConfigurationProperty(type = "float")
  public static final String TEZ_RUNTIME_INPUT_POST_MERGE_BUFFER_PERCENT = TEZ_RUNTIME_PREFIX +
      "task.input.post-merge.buffer.percent";
  public static final float TEZ_RUNTIME_INPUT_BUFFER_PERCENT_DEFAULT = 0.0f;


  @ConfigurationProperty
  public static final String TEZ_RUNTIME_GROUP_COMPARATOR_CLASS = TEZ_RUNTIME_PREFIX +
      "group.comparator.class";

  @ConfigurationProperty
  public static final String TEZ_RUNTIME_INTERNAL_SORTER_CLASS = TEZ_RUNTIME_PREFIX +
      "internal.sorter.class";

  @ConfigurationProperty
  public static final String TEZ_RUNTIME_KEY_COMPARATOR_CLASS =
      TEZ_RUNTIME_PREFIX + "key.comparator.class";

  @ConfigurationProperty
  public static final String TEZ_RUNTIME_KEY_CLASS = TEZ_RUNTIME_PREFIX + "key.class";

  @ConfigurationProperty
  public static final String TEZ_RUNTIME_VALUE_CLASS = TEZ_RUNTIME_PREFIX + "value.class";

  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_COMPRESS = TEZ_RUNTIME_PREFIX + "compress";

  @ConfigurationProperty
  public static final String TEZ_RUNTIME_COMPRESS_CODEC = TEZ_RUNTIME_PREFIX + "compress.codec";

  // TODO Move this key to MapReduce
  @ConfigurationProperty
  public static final String TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS =
      TEZ_RUNTIME_PREFIX + "key.secondary.comparator.class";

  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED =
      TEZ_RUNTIME_PREFIX +
          "empty.partitions.info-via-events.enabled";
  public static final boolean TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED_DEFAULT = true;

  @Private
  public static final String TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_ENABLED =
          TEZ_RUNTIME_PREFIX + "transfer.data-via-events.enabled";
  @Private
  public static final boolean TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_ENABLED_DEFAULT = false;

  @Private
  public static final String TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_MAX_SIZE =
          TEZ_RUNTIME_PREFIX + "transfer.data-via-events.max-size";
  @Private
  public static final int TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_MAX_SIZE_DEFAULT = 512;

  @Private
  public static final String TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_SUPPORT_IN_MEM_FILE =
      TEZ_RUNTIME_PREFIX + "transfer.data-via-events.support.in-mem.file";

  @Private
  public static final boolean TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_SUPPORT_IN_MEM_FILE_DEFAULT
      = true;

  /**
   * If the shuffle input is on the local host bypass the http fetch and access the files directly
   */
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH = TEZ_RUNTIME_PREFIX + "optimize.local.fetch";
  public static final boolean TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH_DEFAULT = true;

  /**
   * Expert level setting. Enable pipelined shuffle in ordered outputs and in unordered
   * partitioned outputs. In ordered cases, it works with PipelinedSorter.
   * set tez.runtime.sort.threads to greater than 1 to enable pipelinedsorter.
   * Ensure to set tez.runtime.enable.final-merge.in.output=false.
   * Speculative execution needs to be turned off when using this parameter. //TODO: TEZ-2132
   */
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED =
      TEZ_RUNTIME_PREFIX + "pipelined-shuffle.enabled";
  public static final boolean TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED_DEFAULT = false;

  /**
   * Expert level setting. Enable final merge in ordered (defaultsorter/pipelinedsorter) outputs.
   * Speculative execution needs to be turned off when disabling this parameter. //TODO: TEZ-2132
   */
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT =
      TEZ_RUNTIME_PREFIX + "enable.final-merge.in.output";
  public static final boolean TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT_DEFAULT = true;

  /**
   * Expert level setting. How long should @link{ShuffleManager} wait for batching
   * before sending the events in milliseconds. Set to -1 to not wait.
   */
  @ConfigurationProperty(type = "integer")
  public static final String  TEZ_RUNTIME_SHUFFLE_BATCH_WAIT =
      TEZ_RUNTIME_PREFIX + "shuffle.batch.wait";
  public static final int TEZ_RUNTIME_SHUFFLE_BATCH_WAIT_DEFAULT = -1;


  /**
   * Share data fetched between tasks running on the same host if applicable
   */
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_OPTIMIZE_SHARED_FETCH = TEZ_RUNTIME_PREFIX
      + "optimize.shared.fetch";

  /**
   * shared mode bypassing the http fetch is not enabled by default till we have unit tests in.
   */
  public static final boolean TEZ_RUNTIME_OPTIMIZE_SHARED_FETCH_DEFAULT = false;

  /**
   * Used only for internal testing. Strictly not recommended to be used elsewhere. This
   * parameter could be changed/dropped later.
   */
  @Unstable
  @Private
  @ConfigurationProperty(type = "boolean")
  public static final String TEZ_RUNTIME_CLEANUP_FILES_ON_INTERRUPT = TEZ_RUNTIME_PREFIX
      + "cleanup.files.on.interrupt";
  public static final boolean TEZ_RUNTIME_CLEANUP_FILES_ON_INTERRUPT_DEFAULT = false;

  // TODO TEZ-1233 - allow this property to be set per vertex
  // TODO TEZ-1231 - move these properties out since they are not relevant for Inputs / Outputs

  /**
   * Value: Boolean
   * Whether to publish configuration information to History logger. Default false.
   */
  @ConfigurationProperty
  public static final String TEZ_RUNTIME_CONVERT_USER_PAYLOAD_TO_HISTORY_TEXT =
      TEZ_RUNTIME_PREFIX + "convert.user-payload.to.history-text";
  public static final boolean TEZ_RUNTIME_CONVERT_USER_PAYLOAD_TO_HISTORY_TEXT_DEFAULT = false;

  @Unstable
  @Private
  @ConfigurationProperty(type = "integer")
  public static final String TEZ_RUNTIME_RECORDS_BEFORE_PROGRESS = TEZ_RUNTIME_PREFIX +
      "merge.progress.records";
  public static final long TEZ_RUNTIME_RECORDS_BEFORE_PROGRESS_DEFAULT = 10000;

  static {
    tezRuntimeKeys.add(TEZ_RUNTIME_IFILE_READAHEAD);
    tezRuntimeKeys.add(TEZ_RUNTIME_IFILE_READAHEAD_BYTES);
    tezRuntimeKeys.add(TEZ_RUNTIME_IO_FILE_BUFFER_SIZE);
    tezRuntimeKeys.add(TEZ_RUNTIME_IO_SORT_FACTOR);
    tezRuntimeKeys.add(TEZ_RUNTIME_SORT_SPILL_PERCENT);
    tezRuntimeKeys.add(TEZ_RUNTIME_IO_SORT_MB);
    tezRuntimeKeys.add(TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES);
    tezRuntimeKeys.add(TEZ_RUNTIME_COMBINE_MIN_SPILLS);
    tezRuntimeKeys.add(TEZ_RUNTIME_PIPELINED_SORTER_SORT_THREADS);
    tezRuntimeKeys.add(
        TEZ_RUNTIME_PIPELINED_SORTER_MIN_BLOCK_SIZE_IN_MB);
    tezRuntimeKeys.add(TEZ_RUNTIME_PIPELINED_SORTER_LAZY_ALLOCATE_MEMORY);
    tezRuntimeKeys.add(TEZ_RUNTIME_UNORDERED_OUTPUT_BUFFER_SIZE_MB);
    tezRuntimeKeys.add(TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES);
    tezRuntimeKeys.add(TEZ_RUNTIME_PARTITIONER_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_COMBINER_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_USE_ASYNC_HTTP);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_PARALLEL_COPIES);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_FETCH_FAILURES_LIMIT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_FETCH_MAX_TASK_OUTPUT_AT_ONCE);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_NOTIFY_READERROR);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_CONNECT_TIMEOUT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_ENABLED);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_KEEP_ALIVE_MAX_CONNECTIONS);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_READ_TIMEOUT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_BUFFER_SIZE);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_ENABLE_SSL);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_FETCH_VERIFY_DISK_CHECKSUM);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_FETCH_BUFFER_PERCENT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_MEMTOMEM_SEGMENTS);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_ENABLE_MEMTOMEM);
    tezRuntimeKeys.add
        (TEZ_RUNTIME_SHUFFLE_ACCEPTABLE_HOST_FETCH_FAILURE_FRACTION);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_MIN_FAILURES_PER_HOST);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_MAX_STALL_TIME_FRACTION);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_SOURCE_ATTEMPT_ABORT_LIMIT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_MAX_ALLOWED_FAILED_FETCH_ATTEMPT_FRACTION);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_MIN_REQUIRED_PROGRESS_FRACTION);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_FAILED_CHECK_SINCE_LAST_COMPLETION);
    tezRuntimeKeys.add(TEZ_RUNTIME_REPORT_PARTITION_STATS);
    tezRuntimeKeys.add(TEZ_RUNTIME_INPUT_POST_MERGE_BUFFER_PERCENT);
    tezRuntimeKeys.add(TEZ_RUNTIME_GROUP_COMPARATOR_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_INTERNAL_SORTER_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_KEY_COMPARATOR_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_KEY_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_VALUE_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_COMPRESS);
    tezRuntimeKeys.add(TEZ_RUNTIME_COMPRESS_CODEC);
    tezRuntimeKeys.add(TEZ_RUNTIME_KEY_SECONDARY_COMPARATOR_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_EMPTY_PARTITION_INFO_VIA_EVENTS_ENABLED);
    tezRuntimeKeys.add(TEZ_RUNTIME_PIPELINED_SHUFFLE_ENABLED);
    tezRuntimeKeys.add(TEZ_RUNTIME_ENABLE_FINAL_MERGE_IN_OUTPUT);
    tezRuntimeKeys.add(TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_ENABLED);
    tezRuntimeKeys.add(TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_MAX_SIZE);
    tezRuntimeKeys.add(TEZ_RUNTIME_TRANSFER_DATA_VIA_EVENTS_SUPPORT_IN_MEM_FILE);
    tezRuntimeKeys.add(TEZ_RUNTIME_RECORDS_BEFORE_PROGRESS);
    tezRuntimeKeys.add(TEZ_RUNTIME_OPTIMIZE_LOCAL_FETCH);
    tezRuntimeKeys.add(TEZ_RUNTIME_OPTIMIZE_SHARED_FETCH);
    tezRuntimeKeys.add(TEZ_RUNTIME_CONVERT_USER_PAYLOAD_TO_HISTORY_TEXT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SORTER_CLASS);
    tezRuntimeKeys.add(TEZ_RUNTIME_CLEANUP_FILES_ON_INTERRUPT);
    tezRuntimeKeys.add(TEZ_RUNTIME_UNORDERED_PARTITIONED_KVWRITER_BUFFER_MERGE_PERCENT);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_FETCHER_USE_SHARED_POOL);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_HOST_PENALTY_TIME_LIMIT_MS);
    tezRuntimeKeys.add(TEZ_RUNTIME_SHUFFLE_BATCH_WAIT);

    defaultConf.addResource("core-default.xml");
    defaultConf.addResource("core-site.xml");
    defaultConf.addResource("tez-site.xml");

    for (Map.Entry<String, String> confEntry : defaultConf) {
      if (tezRuntimeKeys.contains(confEntry.getKey())) {
        tezRuntimeConfMap.put(confEntry.getKey(), confEntry.getValue());
      } else {
        // TODO TEZ-1232 Filter out parameters from TezConfiguration, and Task specific confs
        otherConfMap.put(confEntry.getKey(), confEntry.getValue());
        otherKeys.add(confEntry.getKey());
      }
    }

    // Do NOT need all prefixes from the following list. Only specific ones are allowed
    // "hadoop.", "hadoop.security", "io.", "fs.", "ipc.", "net.", "file.", "dfs.", "ha.", "s3.", "nfs3.", "rpc.", "ssl."
    allowedPrefixes.add("io.");
    allowedPrefixes.add("file.");
    allowedPrefixes.add("fs.");
    allowedPrefixes.add("ssl.");

    umnodifiableTezRuntimeKeySet = Collections.unmodifiableSet(tezRuntimeKeys);
    unmodifiableOtherKeySet = Collections.unmodifiableSet(otherKeys);
    unmodifiableAllowedPrefixes = Collections.unmodifiableList(allowedPrefixes);
  }

  @Private
  public static Set<String> getRuntimeConfigKeySet() {
    return umnodifiableTezRuntimeKeySet;
  }

  @Private
  public static Set<String> getRuntimeAdditionalConfigKeySet() {
    return unmodifiableOtherKeySet;
  }

  @Private
  public static List<String> getAllowedPrefixes() {
    return allowedPrefixes;
  }

  @Private
  public static Map<String, String> getTezRuntimeConfigDefaults() {
    return Collections.unmodifiableMap(tezRuntimeConfMap);
  }

  @Private
  public static Map<String, String> getOtherConfigDefaults() {
    return Collections.unmodifiableMap(otherConfMap);
  }

  public enum ReportPartitionStats {
    @Deprecated
    /**
     * Don't report partition stats. It is the same as NONE.
     * It is defined to maintain backward compatibility given
     * Configuration @link{#TEZ_RUNTIME_REPORT_PARTITION_STATS} used
     * to be boolean type.
     */
    DISABLED("false"),

    @Deprecated
    /**
     * Report partition stats. It is the same as MEMORY_OPTIMIZED.
     * It is defined to maintain backward compatibility given
     * Configuration @link{#TEZ_RUNTIME_REPORT_PARTITION_STATS} used
     * to be boolean type.
     */
    ENABLED("true"),

    /**
     * Don't report partition stats.
     */
    NONE("none"),

    /**
     * Report partition stats with less precision to reduce
     * memory and CPU overhead
     */
    MEMORY_OPTIMIZED("memory_optimized"),

    /**
     * Report precise partition stats in MB.
     */
    PRECISE("precise");

    private final String type;

    private ReportPartitionStats(String type) {
      this.type = type;
    }

    public final String getType() {
      return type;
    }

    public boolean isEnabled() {
      return !equals(ReportPartitionStats.DISABLED) &&
              !equals(ReportPartitionStats.NONE);
    }

    public boolean isPrecise() {
      return equals(ReportPartitionStats.PRECISE);
    }

    public static ReportPartitionStats fromString(String type) {
      if (type != null) {
        for (ReportPartitionStats b : ReportPartitionStats.values()) {
          if (type.equalsIgnoreCase(b.type)) {
            return b;
          }
        }
      }
      throw new IllegalArgumentException("Invalid type " + type);
    }
  }
}