/**
 * 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 com.twitter.distributedlog;

import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;
import com.twitter.distributedlog.bk.QuorumConfig;
import com.twitter.distributedlog.feature.DefaultFeatureProvider;
import com.twitter.distributedlog.namespace.DistributedLogNamespaceBuilder;
import com.twitter.distributedlog.net.DNSResolverForRacks;
import com.twitter.distributedlog.net.DNSResolverForRows;
import org.apache.bookkeeper.conf.ClientConfiguration;
import org.apache.bookkeeper.feature.FeatureProvider;
import org.apache.bookkeeper.net.DNSToSwitchMapping;
import org.apache.bookkeeper.stats.StatsLogger;
import org.apache.bookkeeper.util.ReflectionUtils;
import org.apache.commons.configuration.CompositeConfiguration;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.configuration.SystemConfiguration;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.URL;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Set;

/**
 * DistributedLog Configuration.
 * <p>
 * DistributedLog configuration is basically a properties based configuration, which extends from
 * Apache commons {@link CompositeConfiguration}. All the DL settings are in camel case and prefixed
 * with a meaningful component name. for example, `zkSessionTimeoutSeconds` means <i>SessionTimeoutSeconds</i>
 * for component `zk`.
 *
 * <h3>BookKeeper Configuration</h3>
 *
 * BookKeeper client configuration settings could be loaded via DistributedLog configuration. All those
 * settings are prefixed with <i>`bkc.`</i>. For example, <i>bkc.zkTimeout</i> in distributedlog configuration
 * will be applied as <i>`zkTimeout`</i> in bookkeeper client configuration.
 *
 * <h3>How to load configuration</h3>
 *
 * The default distributedlog configuration is constructed by instantiated a new instance. This
 * distributedlog configuration will automatically load the settings that specified via
 * {@link SystemConfiguration}.
 *
 * <pre>
 *      DistributedLogConfiguration conf = new DistributedLogConfiguration();
 * </pre>
 *
 * The recommended way is to load configuration from URL that points to a configuration file
 * ({@link #loadConf(URL)}).
 *
 * <pre>
 *      String configFile = "/path/to/distributedlog/conf/file";
 *      DistributedLogConfiguration conf = new DistributedLogConfiguration();
 *      conf.loadConf(new File(configFile).toURI().toURL());
 * </pre>
 *
 * @see org.apache.bookkeeper.conf.ClientConfiguration
 */
public class DistributedLogConfiguration extends CompositeConfiguration {
    static final Logger LOG = LoggerFactory.getLogger(DistributedLogConfiguration.class);

    private static ClassLoader defaultLoader;

    static {
        defaultLoader = Thread.currentThread().getContextClassLoader();
        if (null == defaultLoader) {
            defaultLoader = DistributedLogConfiguration.class.getClassLoader();
        }
    }

    //
    // ZooKeeper Related Settings
    //

    public static final String BKDL_ZK_ACL_ID = "zkAclId";
    public static final String BKDL_ZK_ACL_ID_DEFAULT = null;
    public static final String BKDL_ZK_SESSION_TIMEOUT_SECONDS = "zkSessionTimeoutSeconds";
    public static final int BKDL_ZK_SESSION_TIMEOUT_SECONDS_DEFAULT = 30;
    public static final String BKDL_ZK_REQUEST_RATE_LIMIT = "zkRequestRateLimit";
    public static final double BKDL_ZK_REQUEST_RATE_LIMIT_DEFAULT = 0;
    public static final String BKDL_ZK_NUM_RETRIES = "zkNumRetries";
    public static final int BKDL_ZK_NUM_RETRIES_DEFAULT = 3;
    public static final String BKDL_ZK_RETRY_BACKOFF_START_MILLIS = "zkRetryStartBackoffMillis";
    public static final int BKDL_ZK_RETRY_BACKOFF_START_MILLIS_DEFAULT = 5000;
    public static final String BKDL_ZK_RETRY_BACKOFF_MAX_MILLIS = "zkRetryMaxBackoffMillis";
    public static final int BKDL_ZK_RETRY_BACKOFF_MAX_MILLIS_DEFAULT = 30000;
    public static final String BKDL_ZKCLIENT_NUM_RETRY_THREADS = "zkcNumRetryThreads";
    public static final int BKDL_ZKCLIENT_NUM_RETRY_THREADS_DEFAULT = 1;

    //
    // BookKeeper Related Settings
    //

    // BookKeeper zookeeper settings
    public static final String BKDL_BKCLIENT_ZK_SESSION_TIMEOUT = "bkcZKSessionTimeoutSeconds";
    public static final int BKDL_BKCLIENT_ZK_SESSION_TIMEOUT_DEFAULT = 30;
    public static final String BKDL_BKCLIENT_ZK_REQUEST_RATE_LIMIT = "bkcZKRequestRateLimit";
    public static final double BKDL_BKCLIENT_ZK_REQUEST_RATE_LIMIT_DEFAULT = 0;
    public static final String BKDL_BKCLIENT_ZK_NUM_RETRIES = "bkcZKNumRetries";
    public static final int BKDL_BKCLIENT_ZK_NUM_RETRIES_DEFAULT = 3;
    public static final String BKDL_BKCLIENT_ZK_RETRY_BACKOFF_START_MILLIS = "bkcZKRetryStartBackoffMillis";
    public static final int BKDL_BKCLIENT_ZK_RETRY_BACKOFF_START_MILLIS_DEFAULT = 5000;
    public static final String BKDL_BKCLIENT_ZK_RETRY_BACKOFF_MAX_MILLIS = "bkcZKRetryMaxBackoffMillis";
    public static final int BKDL_BKCLIENT_ZK_RETRY_BACKOFF_MAX_MILLIS_DEFAULT = 30000;

    // Bookkeeper ensemble placement settings
    // Bookkeeper ensemble size
    public static final String BKDL_BOOKKEEPER_ENSEMBLE_SIZE = "bkcEnsembleSize";
    // @Deprecated
    public static final String BKDL_BOOKKEEPER_ENSEMBLE_SIZE_OLD = "ensemble-size";
    public static final int BKDL_BOOKKEEPER_ENSEMBLE_SIZE_DEFAULT = 3;
    // Bookkeeper write quorum size
    public static final String BKDL_BOOKKEEPER_WRITE_QUORUM_SIZE = "bkcWriteQuorumSize";
    // @Deprecated
    public static final String BKDL_BOOKKEEPER_WRITE_QUORUM_SIZE_OLD = "write-quorum-size";
    public static final int BKDL_BOOKKEEPER_WRITE_QUORUM_SIZE_DEFAULT = 3;
    // Bookkeeper ack quorum size
    public static final String BKDL_BOOKKEEPER_ACK_QUORUM_SIZE = "bkcAckQuorumSize";
    // @Deprecated
    public static final String BKDL_BOOKKEEPER_ACK_QUORUM_SIZE_OLD = "ack-quorum-size";
    public static final int BKDL_BOOKKEEPER_ACK_QUORUM_SIZE_DEFAULT = 2;
    public static final String BKDL_ROW_AWARE_ENSEMBLE_PLACEMENT = "bkRowAwareEnsemblePlacement";
    public static final String BKDL_ROW_AWARE_ENSEMBLE_PLACEMENT_OLD = "row-aware-ensemble-placement";
    public static final boolean BKDL_ROW_AWARE_ENSEMBLE_PLACEMENT_DEFAULT = false;
    public static final String BKDL_ENSEMBLE_PLACEMENT_DNS_RESOLVER_CLASS = "bkEnsemblePlacementDnsResolverClass";
    public static final String BKDL_ENSEMBLE_PLACEMENT_DNS_RESOLVER_CLASS_DEFAULT =
            DNSResolverForRacks.class.getName();
    public static final String BKDL_BK_DNS_RESOLVER_OVERRIDES = "dnsResolverOverrides";
    public static final String BKDL_BK_DNS_RESOLVER_OVERRIDES_DEFAULT = "";

    // General Settings
    // @Deprecated
    public static final String BKDL_BOOKKEEPER_DIGEST_PW = "digestPw";
    public static final String BKDL_BOOKKEEPER_DIGEST_PW_DEFAULT = "";
    public static final String BKDL_BKCLIENT_NUM_IO_THREADS = "bkcNumIOThreads";
    public static final String BKDL_TIMEOUT_TIMER_TICK_DURATION_MS = "timerTickDuration";
    public static final long BKDL_TIMEOUT_TIMER_TICK_DURATION_MS_DEFAULT = 100;
    public static final String BKDL_TIMEOUT_TIMER_NUM_TICKS = "timerNumTicks";
    public static final int BKDL_TIMEOUT_TIMER_NUM_TICKS_DEFAULT = 1024;

    //
    // Deprecated BookKeeper Settings (in favor of "bkc." style bookkeeper settings)
    //

    public static final String BKDL_BKCLIENT_READ_TIMEOUT = "bkcReadTimeoutSeconds";
    public static final int BKDL_BKCLIENT_READ_TIMEOUT_DEFAULT = 10;
    public static final String BKDL_BKCLIENT_WRITE_TIMEOUT = "bkcWriteTimeoutSeconds";
    public static final int BKDL_BKCLIENT_WRITE_TIMEOUT_DEFAULT = 10;
    public static final String BKDL_BKCLIENT_NUM_WORKER_THREADS = "bkcNumWorkerThreads";
    public static final int BKDL_BKCLEINT_NUM_WORKER_THREADS_DEFAULT = 1;

    //
    // DL General Settings
    //

    // Executor Parameters
    public static final String BKDL_NUM_WORKER_THREADS = "numWorkerThreads";
    public static final String BKDL_NUM_READAHEAD_WORKER_THREADS = "numReadAheadWorkerThreads";
    public static final String BKDL_NUM_LOCKSTATE_THREADS = "numLockStateThreads";
    public static final String BKDL_NUM_RESOURCE_RELEASE_THREADS = "numResourceReleaseThreads";
    public static final String BKDL_SCHEDULER_SHUTDOWN_TIMEOUT_MS = "schedulerShutdownTimeoutMs";
    public static final int BKDL_SCHEDULER_SHUTDOWN_TIMEOUT_MS_DEFAULT = 5000;
    public static final String BKDL_USE_DAEMON_THREAD = "useDaemonThread";
    public static final boolean BKDL_USE_DAEMON_THREAD_DEFAULT = false;

    // Metadata Parameters
    public static final String BKDL_LEDGER_METADATA_LAYOUT_VERSION = "ledgerMetadataLayoutVersion";
    public static final String BKDL_LEDGER_METADATA_LAYOUT_VERSION_OLD = "ledger-metadata-layout";
    public static final int BKDL_LEDGER_METADATA_LAYOUT_VERSION_DEFAULT =
            LogSegmentMetadata.LogSegmentMetadataVersion.VERSION_V5_SEQUENCE_ID.value;
    public static final String BKDL_LEDGER_METADATA_SKIP_MIN_VERSION_CHECK = "ledgerMetadataSkipMinVersionCheck";
    public static final boolean BKDL_LEDGER_METADATA_SKIP_MIN_VERSION_CHECK_DEFAULT = false;
    public static final String BKDL_FIRST_LOGSEGMENT_SEQUENCE_NUMBER = "firstLogsegmentSequenceNumber";
    public static final String BKDL_FIRST_LOGSEGMENT_SEQUENCE_NUMBER_OLD = "first-logsegment-sequence-number";
    public static final long BKDL_FIRST_LOGSEGMENT_SEQUENCE_NUMBER_DEFAULT =
            DistributedLogConstants.FIRST_LOGSEGMENT_SEQNO;
    public static final String BKDL_LOGSEGMENT_SEQUENCE_NUMBER_VALIDATION_ENABLED = "logSegmentSequenceNumberValidationEnabled";
    public static final boolean BKDL_LOGSEGMENT_SEQUENCE_NUMBER_VALIDATION_ENABLED_DEFAULT = true;
    public static final String BKDL_ENABLE_RECORD_COUNTS = "enableRecordCounts";
    public static final boolean BKDL_ENABLE_RECORD_COUNTS_DEFAULT = true;
    public static final String BKDL_MAXID_SANITYCHECK = "maxIdSanityCheck";
    public static final boolean BKDL_MAXID_SANITYCHECK_DEFAULT = true;
    public static final String BKDL_ENCODE_REGION_ID_IN_VERSION = "encodeRegionIDInVersion";
    public static final boolean BKDL_ENCODE_REGION_ID_IN_VERSION_DEFAULT = false;
    // (@Deprecated)
    public static final String BKDL_LOGSEGMENT_NAME_VERSION = "logSegmentNameVersion";
    public static final int BKDL_LOGSEGMENT_NAME_VERSION_DEFAULT = DistributedLogConstants.LOGSEGMENT_NAME_VERSION;
    // (@Derepcated) Name for the default (non-partitioned) stream
    public static final String BKDL_UNPARTITIONED_STREAM_NAME = "unpartitionedStreamName";
    public static final String BKDL_UNPARTITIONED_STREAM_NAME_DEFAULT = "<default>";

    //
    // DL Writer Settings
    //

    // General Settings
    public static final String BKDL_CREATE_STREAM_IF_NOT_EXISTS = "createStreamIfNotExists";
    public static final boolean BKDL_CREATE_STREAM_IF_NOT_EXISTS_DEFAULT = true;
    public static final String BKDL_LOG_FLUSH_TIMEOUT = "logFlushTimeoutSeconds";
    public static final int BKDL_LOG_FLUSH_TIMEOUT_DEFAULT = 30;
    /**
     *  CompressionCodec.Type     String to use (See CompressionUtils)
     *  ---------------------     ------------------------------------
     *          NONE               none
     *          LZ4                lz4
     *          UNKNOWN            any other instance of String.class
     */
    public static final String BKDL_COMPRESSION_TYPE = "compressionType";
    public static final String BKDL_COMPRESSION_TYPE_DEFAULT = "none";
    public static final String BKDL_FAILFAST_ON_STREAM_NOT_READY = "failFastOnStreamNotReady";
    public static final boolean BKDL_FAILFAST_ON_STREAM_NOT_READY_DEFAULT = false;
    public static final String BKDL_DISABLE_ROLLING_ON_LOG_SEGMENT_ERROR = "disableRollingOnLogSegmentError";
    public static final boolean BKDL_DISABLE_ROLLING_ON_LOG_SEGMENT_ERROR_DEFAULT = false;

    // Durability Settings
    public static final String BKDL_IS_DURABLE_WRITE_ENABLED = "isDurableWriteEnabled";
    public static final boolean BKDL_IS_DURABLE_WRITE_ENABLED_DEFAULT = true;

    // Transmit Settings
    public static final String BKDL_OUTPUT_BUFFER_SIZE = "writerOutputBufferSize";
    public static final String BKDL_OUTPUT_BUFFER_SIZE_OLD = "output-buffer-size";
    public static final int BKDL_OUTPUT_BUFFER_SIZE_DEFAULT = 1024;
    public static final String BKDL_PERIODIC_FLUSH_FREQUENCY_MILLISECONDS = "periodicFlushFrequencyMilliSeconds";
    public static final int BKDL_PERIODIC_FLUSH_FREQUENCY_MILLISECONDS_DEFAULT = 0;
    public static final String BKDL_ENABLE_IMMEDIATE_FLUSH = "enableImmediateFlush";
    public static final boolean BKDL_ENABLE_IMMEDIATE_FLUSH_DEFAULT = false;
    public static final String BKDL_MINIMUM_DELAY_BETWEEN_IMMEDIATE_FLUSH_MILLISECONDS = "minimumDelayBetweenImmediateFlushMilliSeconds";
    public static final int BKDL_MINIMUM_DELAY_BETWEEN_IMMEDIATE_FLUSH_MILLISECONDS_DEFAULT = 0;

    // Retention/Truncation Settings
    public static final String BKDL_RETENTION_PERIOD_IN_HOURS = "logSegmentRetentionHours";
    public static final String BKDL_RETENTION_PERIOD_IN_HOURS_OLD = "retention-size";
    public static final int BKDL_RETENTION_PERIOD_IN_HOURS_DEFAULT = 72;
    public static final String BKDL_EXPLICIT_TRUNCATION_BY_APPLICATION = "explicitTruncationByApp";
    public static final boolean BKDL_EXPLICIT_TRUNCATION_BY_APPLICATION_DEFAULT = false;

    // Log Segment Rolling Settings
    public static final String BKDL_ROLLING_INTERVAL_IN_MINUTES = "logSegmentRollingMinutes";
    public static final String BKDL_ROLLING_INTERVAL_IN_MINUTES_OLD = "rolling-interval";
    public static final int BKDL_ROLLING_INTERVAL_IN_MINUTES_DEFAULT = 120;
    public static final String BKDL_MAX_LOGSEGMENT_BYTES = "maxLogSegmentBytes";
    public static final int BKDL_MAX_LOGSEGMENT_BYTES_DEFAULT = 256 * 1024 * 1024; // default 256MB
    public static final String BKDL_LOGSEGMENT_ROLLING_CONCURRENCY = "logSegmentRollingConcurrency";
    public static final int BKDL_LOGSEGMENT_ROLLING_CONCURRENCY_DEFAULT = 1;

    // Lock Settings
    public static final String BKDL_WRITE_LOCK_ENABLED = "writeLockEnabled";
    public static final boolean BKDL_WRITE_LOCK_ENABLED_DEFAULT = true;
    public static final String BKDL_LOCK_TIMEOUT = "lockTimeoutSeconds";
    public static final long BKDL_LOCK_TIMEOUT_DEFAULT = 30;
    public static final String BKDL_LOCK_REACQUIRE_TIMEOUT = "lockReacquireTimeoutSeconds";
    public static final long BKDL_LOCK_REACQUIRE_TIMEOUT_DEFAULT = DistributedLogConstants.LOCK_REACQUIRE_TIMEOUT_DEFAULT;
    public static final String BKDL_LOCK_OP_TIMEOUT = "lockOpTimeoutSeconds";
    public static final long BKDL_LOCK_OP_TIMEOUT_DEFAULT = DistributedLogConstants.LOCK_OP_TIMEOUT_DEFAULT;

    // Ledger Allocator Settings
    public static final String BKDL_ENABLE_LEDGER_ALLOCATOR_POOL = "enableLedgerAllocatorPool";
    public static final boolean BKDL_ENABLE_LEDGER_ALLOCATOR_POOL_DEFAULT = false;
    public static final String BKDL_LEDGER_ALLOCATOR_POOL_PATH = "ledgerAllocatorPoolPath";
    public static final String BKDL_LEDGER_ALLOCATOR_POOL_PATH_DEFAULT = DistributedLogConstants.ALLOCATION_POOL_NODE;
    public static final String BKDL_LEDGER_ALLOCATOR_POOL_NAME = "ledgerAllocatorPoolName";
    public static final String BKDL_LEDGER_ALLOCATOR_POOL_NAME_DEFAULT = null;
    public static final String BKDL_LEDGER_ALLOCATOR_POOL_CORE_SIZE = "ledgerAllocatorPoolCoreSize";
    public static final int BKDL_LEDGER_ALLOCATOR_POOL_CORE_SIZE_DEFAULT = 20;

    // Write Limit Settings
    public static final String BKDL_PER_WRITER_OUTSTANDING_WRITE_LIMIT = "perWriterOutstandingWriteLimit";
    public static final int BKDL_PER_WRITER_OUTSTANDING_WRITE_LIMIT_DEFAULT = -1;
    public static final String BKDL_GLOBAL_OUTSTANDING_WRITE_LIMIT = "globalOutstandingWriteLimit";
    public static final int BKDL_GLOBAL_OUTSTANDING_WRITE_LIMIT_DEFAULT = -1;
    public static final String BKDL_OUTSTANDING_WRITE_LIMIT_DARKMODE = "outstandingWriteLimitDarkmode";
    public static final boolean BKDL_OUTSTANDING_WRITE_LIMIT_DARKMODE_DEFAULT = true;

    //
    // DL Reader Settings
    //

    // General Settings
    public static final String BKDL_READLAC_OPTION = "readLACLongPoll";
    public static final int BKDL_READLAC_OPTION_DEFAULT = 3; //BKLogPartitionReadHandler.ReadLACOption.READENTRYPIGGYBACK_SEQUENTIAL.value
    public static final String BKDL_READLACLONGPOLL_TIMEOUT = "readLACLongPollTimeout";
    public static final int BKDL_READLACLONGPOLL_TIMEOUT_DEFAULT = 1000;
    public static final String BKDL_DESERIALIZE_RECORDSET_ON_READS = "deserializeRecordSetOnReads";
    public static final boolean BKDL_DESERIALIZE_RECORDSET_ON_READS_DEFAULT = true;

    // Idle reader settings
    public static final String BKDL_READER_IDLE_WARN_THRESHOLD_MILLIS = "readerIdleWarnThresholdMillis";
    public static final int BKDL_READER_IDLE_WARN_THRESHOLD_MILLIS_DEFAULT = 120000;
    public static final String BKDL_READER_IDLE_ERROR_THRESHOLD_MILLIS = "readerIdleErrorThresholdMillis";
    public static final int BKDL_READER_IDLE_ERROR_THRESHOLD_MILLIS_DEFAULT = Integer.MAX_VALUE;

    // Reader constraint settings
    public static final String BKDL_READER_IGNORE_TRUNCATION_STATUS = "ignoreTruncationStatus";
    public static final boolean BKDL_READER_IGNORE_TRUNCATION_STATUS_DEFAULT = false;
    public static final String BKDL_READER_ALERT_POSITION_ON_TRUNCATED = "alertPositionOnTruncated";
    public static final boolean BKDL_READER_ALERT_POSITION_ON_TRUNCATED_DEFAULT = true;
    public static final String BKDL_READER_POSITION_GAP_DETECTION_ENABLED = "positionGapDetectionEnabled";
    public static final boolean BKDL_READER_POSITION_GAP_DETECTION_ENABLED_DEFAULT = false;

    // Read ahead related parameters
    public static final String BKDL_ENABLE_READAHEAD = "enableReadAhead";
    public static final boolean BKDL_ENABLE_READAHEAD_DEFAULT = true;
    public static final String BKDL_ENABLE_FORCEREAD = "enableForceRead";
    public static final boolean BKDL_ENABLE_FORCEREAD_DEFAULT = true;
    public static final String BKDL_READAHEAD_MAX_RECORDS = "readAheadMaxRecords";
    public static final String BKDL_READAHEAD_MAX_RECORDS_OLD = "ReadAheadMaxEntries";
    public static final int BKDL_READAHEAD_MAX_RECORDS_DEFAULT = 10;
    public static final String BKDL_READAHEAD_BATCHSIZE = "readAheadBatchSize";
    public static final String BKDL_READAHEAD_BATCHSIZE_OLD = "ReadAheadBatchSize";
    public static final int BKDL_READAHEAD_BATCHSIZE_DEFAULT = 2;
    public static final String BKDL_READAHEAD_WAITTIME = "readAheadWaitTime";
    public static final String BKDL_READAHEAD_WAITTIME_OLD = "ReadAheadWaitTime";
    public static final int BKDL_READAHEAD_WAITTIME_DEFAULT = 200;
    public static final String BKDL_READAHEAD_WAITTIME_ON_ENDOFSTREAM = "readAheadWaitTimeOnEndOfStream";
    public static final String BKDL_READAHEAD_WAITTIME_ON_ENDOFSTREAM_OLD = "ReadAheadWaitTimeOnEndOfStream";
    public static final int BKDL_READAHEAD_WAITTIME_ON_ENDOFSTREAM_DEFAULT = 10000;
    public static final String BKDL_READAHEAD_NOSUCHLEDGER_EXCEPTION_ON_READLAC_ERROR_THRESHOLD_MILLIS =
            "readAheadNoSuchLedgerExceptionOnReadLACErrorThresholdMillis";
    public static final int BKDL_READAHEAD_NOSUCHLEDGER_EXCEPTION_ON_READLAC_ERROR_THRESHOLD_MILLIS_DEFAULT = 10000;
    public static final String BKDL_READAHEAD_SKIP_BROKEN_ENTRIES = "readAheadSkipBrokenEntries";
    public static final boolean BKDL_READAHEAD_SKIP_BROKEN_ENTRIES_DEFAULT = false;

    // Scan Settings
    public static final String BKDL_FIRST_NUM_ENTRIES_PER_READ_LAST_RECORD_SCAN = "firstNumEntriesEachPerLastRecordScan";
    public static final int BKDL_FIRST_NUM_ENTRIES_PER_READ_LAST_RECORD_SCAN_DEFAULT = 2;
    public static final String BKDL_MAX_NUM_ENTRIES_PER_READ_LAST_RECORD_SCAN = "maxNumEntriesPerReadLastRecordScan";
    public static final int BKDL_MAX_NUM_ENTRIES_PER_READ_LAST_RECORD_SCAN_DEFAULT = 16;

    // Log Existence Settings
    public static final String BKDL_CHECK_LOG_EXISTENCE_BACKOFF_START_MS = "checkLogExistenceBackoffStartMillis";
    public static final int BKDL_CHECK_LOG_EXISTENCE_BACKOFF_START_MS_DEFAULT = 200;
    public static final String BKDL_CHECK_LOG_EXISTENCE_BACKOFF_MAX_MS = "checkLogExistenceBackoffMaxMillis";
    public static final int BKDL_CHECK_LOG_EXISTENCE_BACKOFF_MAX_MS_DEFAULT = 1000;

    //
    // Tracing/Stats Settings
    //

    public static final String BKDL_TRACE_READAHEAD_DELIVERY_LATENCY = "traceReadAheadDeliveryLatency";
    public static final boolean BKDL_TRACE_READAHEAD_DELIVERY_LATENCY_DEFAULT = false;
    public static final String BKDL_METADATA_LATENCY_WARN_THRESHOLD_MS = "metadataLatencyWarnThresholdMs";
    public static final long BKDL_METADATA_LATENCY_WARN_THRESHOLD_MS_DEFAULT = DistributedLogConstants.LATENCY_WARN_THRESHOLD_IN_MILLIS;
    public static final String BKDL_DATA_LATENCY_WARN_THRESHOLD_MS = "dataLatencyWarnThresholdMs";
    public static final long BKDL_DATA_LATENCY_WARN_THRESHOLD_MS_DEFAULT = 2 * DistributedLogConstants.LATENCY_WARN_THRESHOLD_IN_MILLIS;
    public static final String BKDL_TRACE_READAHEAD_METADATA_CHANGES = "traceReadAheadMetadataChanges";
    public static final boolean BKDL_TRACE_READAHEAD_MEATDATA_CHANGES_DEFAULT = false;
    public final static String BKDL_ENABLE_TASK_EXECUTION_STATS = "enableTaskExecutionStats";
    public final static boolean BKDL_ENABLE_TASK_EXECUTION_STATS_DEFAULT = false;
    public final static String BKDL_TASK_EXECUTION_WARN_TIME_MICROS = "taskExecutionWarnTimeMicros";
    public final static long BKDL_TASK_EXECUTION_WARN_TIME_MICROS_DEFAULT = 100000;
    public static final String BKDL_ENABLE_PERSTREAM_STAT = "enablePerStreamStat";
    public static final boolean BKDL_ENABLE_PERSTREAM_STAT_DEFAULT = false;

    //
    // Settings for Feature Providers
    //

    public static final String BKDL_FEATURE_PROVIDER_CLASS = "featureProviderClass";

    //
    // Settings for Configuration Based Feature Provider
    //

    public static final String BKDL_FILE_FEATURE_PROVIDER_BASE_CONFIG_PATH = "fileFeatureProviderBaseConfigPath";
    public static final String BKDL_FILE_FEATURE_PROVIDER_BASE_CONFIG_PATH_DEFAULT = "decider.conf";
    public static final String BKDL_FILE_FEATURE_PROVIDER_OVERLAY_CONFIG_PATH = "fileFeatureProviderOverlayConfigPath";
    public static final String BKDL_FILE_FEATURE_PROVIDER_OVERLAY_CONFIG_PATH_DEFAULT = null;

    //
    // Settings for Namespaces
    //

    public static final String BKDL_FEDERATED_NAMESPACE_ENABLED = "federatedNamespaceEnabled";
    public static final boolean BKDL_FEDERATED_NAMESPACE_ENABLED_DEFAULT = false;
    public static final String BKDL_FEDERATED_MAX_LOGS_PER_SUBNAMESPACE = "federatedMaxLogsPerSubnamespace";
    public static final int BKDL_FEDERATED_MAX_LOGS_PER_SUBNAMESPACE_DEFAULT = 15000;
    public static final String BKDL_FEDERATED_CHECK_EXISTENCE_WHEN_CACHE_MISS = "federatedCheckExistenceWhenCacheMiss";
    public static final boolean BKDL_FEDERATED_CHECK_EXISTENCE_WHEN_CACHE_MISS_DEFAULT = true;

    // Settings for Configurations

    public static final String BKDL_DYNAMIC_CONFIG_RELOAD_INTERVAL_SEC = "dynamicConfigReloadIntervalSec";
    public static final int BKDL_DYNAMIC_CONFIG_RELOAD_INTERVAL_SEC_DEFAULT = 60;
    public static final String BKDL_STREAM_CONFIG_ROUTER_CLASS = "streamConfigRouterClass";
    public static final String BKDL_STREAM_CONFIG_ROUTER_CLASS_DEFAULT = "com.twitter.distributedlog.service.config.IdentityConfigRouter";

    // Settings for RateLimit (used by distributedlog-service)

    public static final String BKDL_BPS_SOFT_WRITE_LIMIT = "bpsSoftWriteLimit";
    public static final int BKDL_BPS_SOFT_WRITE_LIMIT_DEFAULT = -1;
    public static final String BKDL_BPS_HARD_WRITE_LIMIT = "bpsHardWriteLimit";
    public static final int BKDL_BPS_HARD_WRITE_LIMIT_DEFAULT = -1;
    public static final String BKDL_RPS_SOFT_WRITE_LIMIT = "rpsSoftWriteLimit";
    public static final int BKDL_RPS_SOFT_WRITE_LIMIT_DEFAULT = -1;
    public static final String BKDL_RPS_HARD_WRITE_LIMIT = "rpsHardWriteLimit";
    public static final int BKDL_RPS_HARD_WRITE_LIMIT_DEFAULT = -1;

    // Rate and resource limits: per shard

    public static final String BKDL_RPS_SOFT_SERVICE_LIMIT = "rpsSoftServiceLimit";
    public static final int BKDL_RPS_SOFT_SERVICE_LIMIT_DEFAULT = -1;
    public static final String BKDL_RPS_HARD_SERVICE_LIMIT = "rpsHardServiceLimit";
    public static final int BKDL_RPS_HARD_SERVICE_LIMIT_DEFAULT = -1;
    public static final String BKDL_RPS_STREAM_ACQUIRE_SERVICE_LIMIT = "rpsStreamAcquireServiceLimit";
    public static final int BKDL_RPS_STREAM_ACQUIRE_SERVICE_LIMIT_DEFAULT = -1;
    public static final String BKDL_BPS_SOFT_SERVICE_LIMIT = "bpsSoftServiceLimit";
    public static final int BKDL_BPS_SOFT_SERVICE_LIMIT_DEFAULT = -1;
    public static final String BKDL_BPS_HARD_SERVICE_LIMIT = "bpsHardServiceLimit";
    public static final int BKDL_BPS_HARD_SERVICE_LIMIT_DEFAULT = -1;
    public static final String BKDL_BPS_STREAM_ACQUIRE_SERVICE_LIMIT = "bpsStreamAcquireServiceLimit";
    public static final int BKDL_BPS_STREAM_ACQUIRE_SERVICE_LIMIT_DEFAULT = -1;

    // Settings for Partitioning

    public static final String BKDL_MAX_ACQUIRED_PARTITIONS_PER_PROXY = "maxAcquiredPartitionsPerProxy";
    public static final int BKDL_MAX_ACQUIRED_PARTITIONS_PER_PROXY_DEFAULT = -1;

    public static final String BKDL_MAX_CACHED_PARTITIONS_PER_PROXY = "maxCachedPartitionsPerProxy";
    public static final int BKDL_MAX_CACHED_PARTITIONS_PER_PROXY_DEFAULT = -1;

    //
    // Settings for Error Injection
    //
    public static final String BKDL_EI_INJECT_WRITE_DELAY = "eiInjectWriteDelay";
    public static final boolean BKDL_EI_INJECT_WRITE_DELAY_DEFAULT = false;
    public static final String BKDL_EI_INJECTED_WRITE_DELAY_PERCENT = "eiInjectedWriteDelayPercent";
    public static final double BKDL_EI_INJECTED_WRITE_DELAY_PERCENT_DEFAULT = 0.0;
    public static final String BKDL_EI_INJECTED_WRITE_DELAY_MS = "eiInjectedWriteDelayMs";
    public static final int BKDL_EI_INJECTED_WRITE_DELAY_MS_DEFAULT = 0;
    public static final String BKDL_EI_INJECT_READAHEAD_STALL = "eiInjectReadAheadStall";
    public static final boolean BKDL_EI_INJECT_READAHEAD_STALL_DEFAULT = false;
    public static final String BKDL_EI_INJECT_READAHEAD_DELAY = "eiInjectReadAheadDelay";
    public static final boolean BKDL_EI_INJECT_READAHEAD_DELAY_DEFAULT = false;
    public static final String BKDL_EI_INJECT_MAX_READAHEAD_DELAY_MS = "eiInjectMaxReadAheadDelayMs";
    public static final int BKDL_EI_INJECT_MAX_READAHEAD_DELAY_MS_DEFAULT = 0;
    public static final String BKDL_EI_INJECT_READAHEAD_DELAY_PERCENT = "eiInjectReadAheadDelayPercent";
    public static final int BKDL_EI_INJECT_READAHEAD_DELAY_PERCENT_DEFAULT = 10;
    public static final String BKDL_EI_INJECT_READAHEAD_BROKEN_ENTRIES = "eiInjectReadAheadBrokenEntries";
    public static final boolean BKDL_EI_INJECT_READAHEAD_BROKEN_ENTRIES_DEFAULT = false;

    // Whitelisted stream-level configuration settings.
    private static final Set<String> streamSettings = Sets.newHashSet(
        BKDL_READER_POSITION_GAP_DETECTION_ENABLED,
        BKDL_READER_IDLE_ERROR_THRESHOLD_MILLIS,
        BKDL_READER_IDLE_WARN_THRESHOLD_MILLIS,
        BKDL_PERIODIC_FLUSH_FREQUENCY_MILLISECONDS,
        BKDL_ENABLE_IMMEDIATE_FLUSH
    );

    /**
     * Construct distributedlog configuration with default settings.
     * It also loads the settings from system properties.
     */
    public DistributedLogConfiguration() {
        super();
        // add configuration for system properties
        addConfiguration(new SystemConfiguration());
    }

    /**
     * You can load configurations in precedence order. The first one takes
     * precedence over any loaded later.
     *
     * @param confURL Configuration URL
     */
    public void loadConf(URL confURL) throws ConfigurationException {
        Configuration loadedConf = new PropertiesConfiguration(confURL);
        addConfiguration(loadedConf);
    }

    /**
     * You can load configuration from other configuration
     *
     * @param baseConf Other Configuration
     */
    public void loadConf(DistributedLogConfiguration baseConf) {
        addConfiguration(baseConf);
    }

    /**
     * Load configuration from other configuration object
     *
     * @param otherConf Other configuration object
     */
    public void loadConf(Configuration otherConf) {
        addConfiguration(otherConf);
    }

    /**
     * Load whitelisted stream configuration from another configuration object
     *
     * @param streamConfiguration stream configuration overrides
     */
    public void loadStreamConf(Optional<DistributedLogConfiguration> streamConfiguration) {
        if (!streamConfiguration.isPresent()) {
            return;
        }
        ArrayList<Object> ignoredSettings = new ArrayList<Object>();
        Iterator iterator = streamConfiguration.get().getKeys();
        while (iterator.hasNext()) {
            Object setting = iterator.next();
            if (setting instanceof String && streamSettings.contains(setting)) {
                String settingStr = (String) setting;
                setProperty(settingStr, streamConfiguration.get().getProperty(settingStr));
            } else {
                ignoredSettings.add(setting);
            }
        }
        if (LOG.isWarnEnabled() && !ignoredSettings.isEmpty()) {
            LOG.warn("invalid stream configuration override(s): {}",
                StringUtils.join(ignoredSettings, ";"));
        }
    }

    //
    // ZooKeeper Related Settings
    //

    /**
     * Get all properties as a string.
     */
    public String getPropsAsString() {
        Iterator iterator = getKeys();
        StringBuilder builder = new StringBuilder();
        boolean appendNewline = false;
        while (iterator.hasNext()) {
            Object key = iterator.next();
            if (key instanceof String) {
                if (appendNewline) {
                    builder.append("\n");
                }
                Object value = getProperty((String)key);
                builder.append(key).append("=").append(value);
                appendNewline = true;
            }
        }
        return builder.toString();
    }

    /**
     * Get digest id used for ZK acl.
     *
     * @return zk acl id.
     */
    public String getZkAclId() {
        return getString(BKDL_ZK_ACL_ID, BKDL_ZK_ACL_ID_DEFAULT);
    }

    /**
     * Set digest id to use for ZK acl.
     *
     * @param zkAclId acl id.
     * @return distributedlog configuration
     * @see #getZkAclId()
     */
    public DistributedLogConfiguration setZkAclId(String zkAclId) {
        setProperty(BKDL_ZK_ACL_ID, zkAclId);
        return this;
    }

    /**
     * Get ZK Session timeout in seconds.
     * <p>
     * This is the session timeout applied for zookeeper client used by distributedlog.
     * Use {@link #getBKClientZKSessionTimeoutMilliSeconds()} for zookeeper client used
     * by bookkeeper client.
     *
     * @return zookeeeper session timeout in seconds.
     * @deprecated use {@link #getZKSessionTimeoutMilliseconds()}
     */
    public int getZKSessionTimeoutSeconds() {
        return this.getInt(BKDL_ZK_SESSION_TIMEOUT_SECONDS, BKDL_ZK_SESSION_TIMEOUT_SECONDS_DEFAULT);
    }

    /**
     * Get ZK Session timeout in milliseconds.
     * <p>
     * This is the session timeout applied for zookeeper client used by distributedlog.
