content: /*
   * 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.bookkeeper.conf;
  
  import static org.apache.bookkeeper.conf.ClientConfiguration.CLIENT_AUTH_PROVIDER_FACTORY_CLASS;
  
  import io.netty.buffer.PooledByteBufAllocator;
  import java.net.URL;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Map;
  import javax.net.ssl.SSLEngine;
  import lombok.extern.slf4j.Slf4j;
  import org.apache.bookkeeper.common.allocator.LeakDetectionPolicy;
  import org.apache.bookkeeper.common.allocator.OutOfMemoryPolicy;
  import org.apache.bookkeeper.common.allocator.PoolingPolicy;
  import org.apache.bookkeeper.common.conf.ConfigurationUtil;
  import org.apache.bookkeeper.common.util.JsonUtil;
  import org.apache.bookkeeper.common.util.JsonUtil.ParseJsonException;
  import org.apache.bookkeeper.common.util.ReflectionUtils;
  import org.apache.bookkeeper.feature.Feature;
  import org.apache.bookkeeper.meta.AbstractZkLedgerManagerFactory;
  import org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory;
  import org.apache.bookkeeper.meta.LedgerManagerFactory;
  import org.apache.bookkeeper.meta.LongHierarchicalLedgerManagerFactory;
  import org.apache.bookkeeper.util.EntryFormatter;
  import org.apache.bookkeeper.util.LedgerIdFormatter;
  import org.apache.bookkeeper.util.StringEntryFormatter;
  import org.apache.commons.configuration2.CompositeConfiguration;
  import org.apache.commons.configuration2.PropertiesConfiguration;
  import org.apache.commons.configuration2.convert.DefaultListDelimiterHandler;
  import org.apache.commons.configuration2.ex.ConfigurationException;
  import org.apache.commons.lang3.StringUtils;
  
  /**
   * Abstract configuration.
   */
  @Slf4j
  public abstract class AbstractConfiguration<T extends AbstractConfiguration>
      extends CompositeConfiguration {
  
      public static final String READ_SYSTEM_PROPERTIES_PROPERTY = "org.apache.bookkeeper.conf.readsystemproperties";
  
      /**
       * Enable the use of System Properties, which was the default behaviour till 4.4.0.
       */
      private static final boolean READ_SYSTEM_PROPERTIES = Boolean.getBoolean(READ_SYSTEM_PROPERTIES_PROPERTY);
  
      protected static final ClassLoader DEFAULT_LOADER;
      static {
          ClassLoader loader = Thread.currentThread().getContextClassLoader();
          if (null == loader) {
              loader = AbstractConfiguration.class.getClassLoader();
          }
          DEFAULT_LOADER = loader;
      }
  
      // Zookeeper Parameters
      protected static final String ZK_TIMEOUT = "zkTimeout";
      protected static final String ZK_SERVERS = "zkServers";
      protected static final String ZK_RETRY_BACKOFF_MAX_RETRIES = "zkRetryBackoffMaxRetries";
  
      // Ledger Manager
      protected static final String LEDGER_MANAGER_TYPE = "ledgerManagerType";
      protected static final String LEDGER_MANAGER_FACTORY_CLASS = "ledgerManagerFactoryClass";
      protected static final String LEDGER_METADATA_FORMAT_VERSION = "ledgerMetadataVersion";
      protected static final String ALLOW_SHADED_LEDGER_MANAGER_FACTORY_CLASS = "allowShadedLedgerManagerFactoryClass";
      protected static final String SHADED_LEDGER_MANAGER_FACTORY_CLASS_PREFIX = "shadedLedgerManagerFactoryClassPrefix";
      protected static final String METADATA_SERVICE_URI = "metadataServiceUri";
      protected static final String ZK_LEDGERS_ROOT_PATH = "zkLedgersRootPath";
      protected static final String ZK_REQUEST_RATE_LIMIT = "zkRequestRateLimit";
      protected static final String ZK_REPLICATION_TASK_RATE_LIMIT = "zkReplicationTaskRateLimit";
      protected static final String AVAILABLE_NODE = "available";
      protected static final String REREPLICATION_ENTRY_BATCH_SIZE = "rereplicationEntryBatchSize";
      protected static final String STORE_SYSTEMTIME_AS_LEDGER_UNDERREPLICATED_MARK_TIME =
              "storeSystemTimeAsLedgerUnderreplicatedMarkTime";
      protected static final String STORE_SYSTEMTIME_AS_LEDGER_CREATION_TIME = "storeSystemTimeAsLedgerCreationTime";
  
      protected static final String ENABLE_BUSY_WAIT = "enableBusyWait";
      protected static final String ENABLE_HEALTH_CHECK = "enableHealthCheck";
  
      // Metastore settings, only being used when LEDGER_MANAGER_FACTORY_CLASS is MSLedgerManagerFactory
      protected static final String METASTORE_IMPL_CLASS = "metastoreImplClass";
      protected static final String METASTORE_MAX_ENTRIES_PER_SCAN = "metastoreMaxEntriesPerScan";
  
      // Common TLS configuration
      // TLS Provider (JDK or OpenSSL)
      protected static final String TLS_PROVIDER = "tlsProvider";
  
      // TLS provider factory class name
      protected static final String TLS_PROVIDER_FACTORY_CLASS = "tlsProviderFactoryClass";
  
      protected static final String LEDGERID_FORMATTER_CLASS = "ledgerIdFormatterClass";
      protected static final String ENTRY_FORMATTER_CLASS = "entryFormatterClass";
  
      // Enable authentication of the other connection end point (mutual authentication)
      protected static final String TLS_CLIENT_AUTHENTICATION = "tlsClientAuthentication";
  
      // Preserve MDC or not for tasks in executor
      protected static final String PRESERVE_MDC_FOR_TASK_EXECUTION = "preserveMdcForTaskExecution";
  
      // Default formatter classes
      protected static final Class<? extends EntryFormatter> DEFAULT_ENTRY_FORMATTER = StringEntryFormatter.class;
      protected static final Class<? extends LedgerIdFormatter> DEFAULT_LEDGERID_FORMATTER =
              LedgerIdFormatter.LongLedgerIdFormatter.class;
  
      protected static final String TLS_CERT_FILES_REFRESH_DURATION_SECONDS = "tlsCertFilesRefreshDurationSeconds";
      /**
       * This list will be passed to {@link SSLEngine#setEnabledCipherSuites(java.lang.String[]) }.
       * Please refer to official JDK JavaDocs
      */
      protected static final String TLS_ENABLED_CIPHER_SUITES = "tlsEnabledCipherSuites";
  
      /**
       * This list will be passed to {@link SSLEngine#setEnabledProtocols(java.lang.String[]) }.
       * Please refer to official JDK JavaDocs
      */
      protected static final String TLS_ENABLED_PROTOCOLS = "tlsEnabledProtocols";
  
      /**
       * TLS KeyStore, TrustStore, Password files and Certificate Paths.
       */
      protected static final String TLS_KEYSTORE_TYPE = "tlsKeyStoreType";
      protected static final String TLS_KEYSTORE = "tlsKeyStore";
      protected static final String TLS_KEYSTORE_PASSWORD_PATH = "tlsKeyStorePasswordPath";
      protected static final String TLS_TRUSTSTORE_TYPE = "tlsTrustStoreType";
      protected static final String TLS_TRUSTSTORE = "tlsTrustStore";
      protected static final String TLS_TRUSTSTORE_PASSWORD_PATH = "tlsTrustStorePasswordPath";
      protected static final String TLS_CERTIFICATE_PATH = "tlsCertificatePath";
  
      //Netty configuration
      protected static final String NETTY_MAX_FRAME_SIZE = "nettyMaxFrameSizeBytes";
      protected static final int DEFAULT_NETTY_MAX_FRAME_SIZE = 5 * 1024 * 1024; // 5MB
  
      // Zookeeper ACL settings
      protected static final String ZK_ENABLE_SECURITY = "zkEnableSecurity";
  
      // Kluge for compatibility testing. Never set this outside tests.
      public static final String LEDGER_MANAGER_FACTORY_DISABLE_CLASS_CHECK = "ledgerManagerFactoryDisableClassCheck";
  
      // Validate bookie process user
      public static final String PERMITTED_STARTUP_USERS = "permittedStartupUsers";
  
      // minimum number of racks per write quorum
      public static final String MIN_NUM_RACKS_PER_WRITE_QUORUM = "minNumRacksPerWriteQuorum";
  
      // enforce minimum number of racks per write quorum
      public static final String ENFORCE_MIN_NUM_RACKS_PER_WRITE_QUORUM = "enforceMinNumRacksPerWriteQuorum";
  
      // enforce minimum number of fault domains for write
      public static final String ENFORCE_MIN_NUM_FAULT_DOMAINS_FOR_WRITE = "enforceMinNumFaultDomainsForWrite";
  
      // ignore usage of local node in the internal logic of placement policy
      public static final String IGNORE_LOCAL_NODE_IN_PLACEMENT_POLICY = "ignoreLocalNodeInPlacementPolicy";
  
      // minimum number of zones per write quorum in ZoneAwarePlacementPolicy
      public static final String MIN_NUM_ZONES_PER_WRITE_QUORUM = "minNumZonesPerWriteQuorum";
  
      // desired number of zones per write quorum in ZoneAwarePlacementPolicy
      public static final String DESIRED_NUM_ZONES_PER_WRITE_QUORUM = "desiredNumZonesPerWriteQuorum";
  
      // in ZoneawareEnsemblePlacementPolicy if strict placement is enabled then
      // minZones/desiredZones in writeQuorum would be maintained otherwise it
      // will pick nodes randomly.
      public static final String ENFORCE_STRICT_ZONEAWARE_PLACEMENT = "enforceStrictZoneawarePlacement";
  
      // Allocator configuration
      protected static final String ALLOCATOR_POOLING_POLICY = "allocatorPoolingPolicy";
      protected static final String ALLOCATOR_POOLING_CONCURRENCY = "allocatorPoolingConcurrency";
      protected static final String ALLOCATOR_OOM_POLICY = "allocatorOutOfMemoryPolicy";
      protected static final String ALLOCATOR_LEAK_DETECTION_POLICY = "allocatorLeakDetectionPolicy";
      protected static final String ALLOCATOR_EXIT_ON_OUT_OF_MEMORY = "allocatorExitOnOutOfMemory";
  
      // option to limit stats logging
      public static final String LIMIT_STATS_LOGGING = "limitStatsLogging";
  
      protected static final String REPLICATION_RATE_BY_BYTES = "replicationRateByBytes";
  
      protected AbstractConfiguration() {
          super();
          // configure list handling to behave in the same way as in commons-configuration 1.x
          setListDelimiterHandler(new DefaultListDelimiterHandler(','));
          if (READ_SYSTEM_PROPERTIES) {
              // add configuration for system properties
              addConfiguration(ConfigurationUtil.readSystemPropertiesAsConfiguration());
          }
      }
  
      /**
       * Limit who can start the application to prevent future permission errors.
       */
      public void setPermittedStartupUsers(String s) {
          setProperty(PERMITTED_STARTUP_USERS, s);
      }
  
      /**
       * Get array of users specified in this property.
       */
      public String[] getPermittedStartupUsers() {
          return getStringArray(PERMITTED_STARTUP_USERS);
      }
  
      /**
       * You can load configurations in precedence order. The first one takes
       * precedence over any loaded later.
       *
       * @param confURL
       *          Configuration URL
       */
      @SuppressWarnings("unchecked")
      public void loadConf(URL confURL) throws ConfigurationException {
          PropertiesConfiguration loadedConf =
                  ConfigurationUtil.newConfiguration(conf -> conf.propertiesBuilder(confURL));
          for (Iterator<String> iter = loadedConf.getKeys(); iter.hasNext(); ) {
              String key = iter.next();
              setProperty(key, loadedConf.getProperty(key));
          }
      }
  
      /**
       * You can load configuration from other configuration.
       *
       * @param baseConf
       *          Other Configuration
       */
      @SuppressWarnings("unchecked")
      public void loadConf(CompositeConfiguration baseConf) {
          for (Iterator<String> iter = baseConf.getKeys(); iter.hasNext(); ) {
              String key = iter.next();
              setProperty(key, baseConf.getProperty(key));
          }
      }
  
      /**
       * Get metadata service uri.
       *
       * <p><b>Warning:</b> this method silently converts checked exceptions to unchecked exceptions.
       * It is useful to use this method in lambda expressions. However it should not be used with places
       * which have logics to handle checked exceptions. In such cases use {@link #getMetadataServiceUri()} instead.
       *
       * @return metadata service uri
       * @throws UncheckedConfigurationException if the metadata service uri is invalid.
       */
      public String getMetadataServiceUriUnchecked() throws UncheckedConfigurationException {
          try {
              return getMetadataServiceUri();
          } catch (ConfigurationException e) {
              throw new UncheckedConfigurationException(e);
          }
      }
  
      /**
       * Get metadata service uri.
       *
       * @return metadata service uri.
       * @throws ConfigurationException if the metadata service uri is invalid.
       */
      public String getMetadataServiceUri() throws ConfigurationException {
          String serviceUri = getString(METADATA_SERVICE_URI);
          if (StringUtils.isBlank(serviceUri)) {
              // no service uri is defined, fallback to old settings
              String ledgerManagerType;
              ledgerManagerType = getLedgerManagerLayoutStringFromFactoryClass();
              String zkServers = getZkServers();
              if (null != zkServers) {
                  // URI doesn't accept ','
                  serviceUri = String.format(
                      "zk+%s://%s%s",
                      ledgerManagerType,
                      zkServers.replace(",", ";"),
                      getZkLedgersRootPath());
              }
          }
          return serviceUri;
      }
  
      /**
       * Set the metadata service uri.
       *
       * @param serviceUri the metadata service uri.
       * @return the configuration object.
       */
      public T setMetadataServiceUri(String serviceUri) {
          setProperty(METADATA_SERVICE_URI, serviceUri);
          return getThis();
      }
  
      /**
       * Get zookeeper servers to connect.
       *
       * <p>`zkServers` is deprecating, in favor of using `metadataServiceUri`
       *
       * @return zookeeper servers
       * @deprecated since 4.7.0
       */
      @Deprecated
      public String getZkServers() {
          List servers = getList(ZK_SERVERS, null);
          if (null == servers || 0 == servers.size()) {
              return null;
          }
          return StringUtils.join(servers, ",");
      }
  
      /**
       * Set zookeeper servers to connect.
       *
       * <p>`zkServers` is deprecating, in favor of using `metadataServiceUri`
       *
       * @param zkServers
       *          ZooKeeper servers to connect
       */
      @Deprecated
      public T setZkServers(String zkServers) {
          setProperty(ZK_SERVERS, zkServers);
          return getThis();
      }
  
      /**
       * Get zookeeper timeout.
       *
       * @return zookeeper server timeout
       */
      public int getZkTimeout() {
          return getInt(ZK_TIMEOUT, 10000);
      }
  
      /**
       * Set zookeeper timeout.
       *
       * @param zkTimeout
       *          ZooKeeper server timeout
       * @return server configuration
       */
      public T setZkTimeout(int zkTimeout) {
          setProperty(ZK_TIMEOUT, Integer.toString(zkTimeout));
          return getThis();
      }
  
      /**
       * Get zookeeper client backoff max retry times.
       *
       * @return zk backoff max retry times.
       */
      public int getZkRetryBackoffMaxRetries() {
          return getInt(ZK_RETRY_BACKOFF_MAX_RETRIES, Integer.MAX_VALUE);
      }
  
      /**
       * Set zookeeper client backoff max retry times.
       *
       * @param maxRetries
       *          backoff max retry times
       * @return server configuration.
       */
      public T setZkRetryBackoffMaxRetries(int maxRetries) {
          setProperty(ZK_RETRY_BACKOFF_MAX_RETRIES, Integer.toString(maxRetries));
          return getThis();
      }
  
      /**
       * Set Ledger Manager Type.
       *
       * @param lmType
       *          Ledger Manager Type
       * @deprecated replaced by {@link #setLedgerManagerFactoryClass}
       */
      @Deprecated
      public void setLedgerManagerType(String lmType) {
          setProperty(LEDGER_MANAGER_TYPE, lmType);
      }
  
      /**
       * Get Ledger Manager Type.
       *
       * @return ledger manager type
       *
       * @deprecated replaced by {@link #getLedgerManagerFactoryClass()}
       */
      @Deprecated
      public String getLedgerManagerType() {
          return getString(LEDGER_MANAGER_TYPE);
      }
  
      /**
       * Set the flag to allow using shaded ledger manager factory class for
       * instantiating a ledger manager factory.
       *
       * @param allowed
       *          the flag to allow/disallow using shaded ledger manager factory class
       * @return configuration instance.
       */
      public T setAllowShadedLedgerManagerFactoryClass(boolean allowed) {
          setProperty(ALLOW_SHADED_LEDGER_MANAGER_FACTORY_CLASS, allowed);
          return getThis();
      }
  
      /**
       * Is shaded ledger manager factory class name allowed to be used for
       * instantiating ledger manager factory.
       *
       * @return ledger manager factory class name.
       */
      public boolean isShadedLedgerManagerFactoryClassAllowed() {
          return getBoolean(ALLOW_SHADED_LEDGER_MANAGER_FACTORY_CLASS, false);
      }
  
      /**
       * Set the class prefix of the shaded ledger manager factory class for
       * instantiating a ledger manager factory.
       *
       * <p>This setting only takes effects when {@link #isShadedLedgerManagerFactoryClassAllowed()}
       * returns true.
       *
       * @param classPrefix
       *          the class prefix of shaded ledger manager factory class
       * @return configuration instance.
       */
      public T setShadedLedgerManagerFactoryClassPrefix(String classPrefix) {
          setProperty(SHADED_LEDGER_MANAGER_FACTORY_CLASS_PREFIX, classPrefix);
          return getThis();
      }
  
      /**
       * Get the class prefix of the shaded ledger manager factory class name allowed to be used for
       * instantiating ledger manager factory.
       *
       * <p>This setting only takes effects when {@link #isShadedLedgerManagerFactoryClassAllowed()}
       * returns true
       *
       * @return ledger manager factory class name.
       * @see #isShadedLedgerManagerFactoryClassAllowed()
       */
      public String getShadedLedgerManagerFactoryClassPrefix() {
          return getString(SHADED_LEDGER_MANAGER_FACTORY_CLASS_PREFIX, "dlshade.");
      }
  
      /**
       * Set Ledger Manager Factory Class Name.
       *
       * @param factoryClassName
       *          Ledger Manager Factory Class Name
       */
      public void setLedgerManagerFactoryClassName(String factoryClassName) {
          setProperty(LEDGER_MANAGER_FACTORY_CLASS, factoryClassName);
      }
  
      /**
       * Get Ledger Manager Factory Class Name.
       *
       * @return ledger manager factory class name.
       */
      public String getLedgerManagerFactoryClassName() {
          return getString(LEDGER_MANAGER_FACTORY_CLASS);
      }
  
      /**
       * Set Ledger metadata format version.
       *
       * @param metadataFormatVersion
       *          Ledger metadata format version. pass -1 to use default version
       */
      public void setLedgerMetadataFormatVersion(int metadataFormatVersion) {
          setProperty(LEDGER_METADATA_FORMAT_VERSION, metadataFormatVersion);
      }
  
      /**
       * Get Ledger metadata format version.
       *
       * @return ledger metadata format version.
       */
      public int getLedgerMetadataFormatVersion() {
          return getInt(LEDGER_METADATA_FORMAT_VERSION, -1);
      }
  
      /**
       * Get layout string ("null" if unconfigured).
       *
       * @return null, hierarchical, longhierarchical, or flat based on LEDGER_MANAGER_FACTORY_CLASS
       */
      @SuppressWarnings("deprecation")
      public String getLedgerManagerLayoutStringFromFactoryClass() throws ConfigurationException {
          String ledgerManagerType;
          Class<? extends LedgerManagerFactory> factoryClass = getLedgerManagerFactoryClass();
          if (factoryClass == null) {
              // set the ledger manager type to "null", so the driver implementation knows that the type is not set.
              ledgerManagerType = "null";
          } else {
              if (!AbstractZkLedgerManagerFactory.class.isAssignableFrom(factoryClass)) {
                  // this is a non-zk implementation
                  throw new ConfigurationException("metadata service uri is not supported for " + factoryClass);
              }
              if (factoryClass == HierarchicalLedgerManagerFactory.class) {
                  ledgerManagerType = HierarchicalLedgerManagerFactory.NAME;
              } else if (factoryClass == org.apache.bookkeeper.meta.FlatLedgerManagerFactory.class) {
                  ledgerManagerType = org.apache.bookkeeper.meta.FlatLedgerManagerFactory.NAME;
              } else if (factoryClass == LongHierarchicalLedgerManagerFactory.class) {
                  ledgerManagerType = LongHierarchicalLedgerManagerFactory.NAME;
              } else if (factoryClass == org.apache.bookkeeper.meta.MSLedgerManagerFactory.class) {
                  ledgerManagerType = org.apache.bookkeeper.meta.MSLedgerManagerFactory.NAME;
              } else {
                  throw new IllegalArgumentException("Unknown zookeeper based ledger manager factory : "
                          + factoryClass);
              }
          }
          return ledgerManagerType;
      }
  
      /**
       * Set Ledger Manager Factory Class.
       *
       * @param factoryClass
       *          Ledger Manager Factory Class
       */
      public void setLedgerManagerFactoryClass(Class<? extends LedgerManagerFactory> factoryClass) {
          setProperty(LEDGER_MANAGER_FACTORY_CLASS, factoryClass.getName());
      }
  
      /**
       * Get ledger manager factory class.
       *
       * @return ledger manager factory class
       */
      public Class<? extends LedgerManagerFactory> getLedgerManagerFactoryClass()
          throws ConfigurationException {
          return ReflectionUtils.getClass(this, LEDGER_MANAGER_FACTORY_CLASS,
                                          null, LedgerManagerFactory.class,
                                          DEFAULT_LOADER);
      }
  
      /**
       * Set Zk Ledgers Root Path.
       *
       * @param zkLedgersPath zk ledgers root path
       */
      @Deprecated
      public void setZkLedgersRootPath(String zkLedgersPath) {
          setProperty(ZK_LEDGERS_ROOT_PATH, zkLedgersPath);
      }
  
      /**
       * Get Zk Ledgers Root Path.
       *
       * @return zk ledgers root path
       */
      @Deprecated
      public String getZkLedgersRootPath() {
          return getString(ZK_LEDGERS_ROOT_PATH, "/ledgers");
      }
  
      /**
       * Get zookeeper access request rate limit.
       *
       * @return zookeeper access request rate limit.
       */
      public double getZkRequestRateLimit() {
          return getDouble(ZK_REQUEST_RATE_LIMIT, 0);
      }
  
      /**
       * Set zookeeper access request rate limit.
       *
       * @param rateLimit
       *          zookeeper access request rate limit.
       */
      public void setZkRequestRateLimit(double rateLimit) {
          setProperty(ZK_REQUEST_RATE_LIMIT, rateLimit);
      }
  
      /**
       * Are z-node created with strict ACLs.
       *
       * @return usage of secure ZooKeeper ACLs
       */
      public boolean isZkEnableSecurity() {
          return getBoolean(ZK_ENABLE_SECURITY, false);
      }
  
      /**
       * Set the usage of ACLs of new z-nodes.
       *
       * @param zkEnableSecurity
       */
      public void setZkEnableSecurity(boolean zkEnableSecurity) {
          setProperty(ZK_ENABLE_SECURITY, zkEnableSecurity);
      }
  
      /**
       * Get the node under which available bookies are stored.
       *
       * @return Node under which available bookies are stored.
       */
      @Deprecated
      public String getZkAvailableBookiesPath() {
          return getZkLedgersRootPath() + "/" + AVAILABLE_NODE;
      }
  
      /**
       * Set the max entries to keep in fragment for re-replication. If fragment
       * has more entries than this count, then the original fragment will be
       * split into multiple small logical fragments by keeping max entries count
       * to rereplicationEntryBatchSize. So, re-replication will happen in batches
       * wise.
       */
      public void setRereplicationEntryBatchSize(long rereplicationEntryBatchSize) {
          setProperty(REREPLICATION_ENTRY_BATCH_SIZE, rereplicationEntryBatchSize);
      }
  
      /**
       * Get the re-replication entry batch size.
       */
      public long getRereplicationEntryBatchSize() {
          return getLong(REREPLICATION_ENTRY_BATCH_SIZE, 10);
      }
  
      /**
       * Get metastore implementation class.
       *
       * @return metastore implementation class name.
       */
      public String getMetastoreImplClass() {
          return getString(METASTORE_IMPL_CLASS);
      }
  
      /**
       * Set metastore implementation class.
       *
       * @param metastoreImplClass
       *          Metastore implementation Class name.
       */
      public void setMetastoreImplClass(String metastoreImplClass) {
          setProperty(METASTORE_IMPL_CLASS, metastoreImplClass);
      }
  
      /**
       * Get max entries per scan in metastore.
       *
       * @return max entries per scan in metastore.
       */
      public int getMetastoreMaxEntriesPerScan() {
          return getInt(METASTORE_MAX_ENTRIES_PER_SCAN, 50);
      }
  
      /**
       * Set max entries per scan in metastore.
       *
       * @param maxEntries
       *          Max entries per scan in metastore.
       */
      public void setMetastoreMaxEntriesPerScan(int maxEntries) {
          setProperty(METASTORE_MAX_ENTRIES_PER_SCAN, maxEntries);
      }
  
      public void setFeature(String configProperty, Feature feature) {
          setProperty(configProperty, feature);
      }
  
      public Feature getFeature(String configProperty, Feature defaultValue) {
          if (null == getProperty(configProperty)) {
              return defaultValue;
          } else {
              return (Feature) getProperty(configProperty);
          }
      }
  
      /**
       * Set Ledger id formatter Class.
       *
       * @param formatterClass
       *          LedgerIdFormatter Class
       */
      public void setLedgerIdFormatterClass(Class<? extends LedgerIdFormatter> formatterClass) {
          setProperty(LEDGERID_FORMATTER_CLASS, formatterClass.getName());
      }
  
      /**
       * Get ledger id formatter class.
       *
       * @return LedgerIdFormatter class
       */
      public Class<? extends LedgerIdFormatter> getLedgerIdFormatterClass()
          throws ConfigurationException {
          return ReflectionUtils.getClass(this, LEDGERID_FORMATTER_CLASS, DEFAULT_LEDGERID_FORMATTER,
                  LedgerIdFormatter.class, DEFAULT_LOADER);
      }
  
      /**
       * Set entry formatter Class.
       *
       * @param formatterClass
       *          EntryFormatter Class
       */
      public void setEntryFormatterClass(Class<? extends EntryFormatter> formatterClass) {
          setProperty(ENTRY_FORMATTER_CLASS, formatterClass.getName());
      }
  
      /**
       * Get entry formatter class.
       *
       * @return EntryFormatter class
       */
      public Class<? extends EntryFormatter> getEntryFormatterClass()
          throws ConfigurationException {
          return ReflectionUtils.getClass(this, ENTRY_FORMATTER_CLASS, DEFAULT_ENTRY_FORMATTER, EntryFormatter.class,
                  DEFAULT_LOADER);
      }
  
      /**
       * Set the client authentication provider factory class name.
       * If this is not set, no authentication will be used
       *
       * @param factoryClass
       *          the client authentication provider factory class name
       * @return client configuration
       */
      public T setClientAuthProviderFactoryClass(
              String factoryClass) {
          setProperty(CLIENT_AUTH_PROVIDER_FACTORY_CLASS, factoryClass);
          return getThis();
      }
  
      /**
       * Get the client authentication provider factory class name.
       * If this returns null, no authentication will take place.
       *
       * @return the client authentication provider factory class name or null.
       */
      public String getClientAuthProviderFactoryClass() {
          return getString(CLIENT_AUTH_PROVIDER_FACTORY_CLASS, null);
      }
  
      /**
       * Get the maximum netty frame size in bytes.  Any message received larger
       * that this will be rejected.
       *
       * @return the maximum netty frame size in bytes.
       */
      public int getNettyMaxFrameSizeBytes() {
          return getInt(NETTY_MAX_FRAME_SIZE, DEFAULT_NETTY_MAX_FRAME_SIZE);
      }
  
      /**
       * Set the max number of bytes a single message can be that is read by the bookie.
       * Any message larger than that size will be rejected.
       *
       * @param maxSize
       *          the max size in bytes
       * @return server configuration
       */
      public T setNettyMaxFrameSizeBytes(int maxSize) {
          setProperty(NETTY_MAX_FRAME_SIZE, String.valueOf(maxSize));
          return getThis();
      }
  
      /**
       * Get the security provider factory class name. If this returns null, no security will be enforced on the channel.
       *
       * @return the security provider factory class name or null.
       */
      public String getTLSProviderFactoryClass() {
          return getString(TLS_PROVIDER_FACTORY_CLASS, null);
      }
  
      /**
       * Set the client security provider factory class name. If this is not set, no security will be used on the channel.
       *
       * @param factoryClass
       *            the client security provider factory class name
       * @return client configuration
       */
      public T setTLSProviderFactoryClass(String factoryClass) {
          setProperty(TLS_PROVIDER_FACTORY_CLASS, factoryClass);
          return getThis();
      }
  
      /**
       * Get TLS Provider (JDK or OpenSSL).
       *
       * @return the TLS provider to use in creating TLS Context
       */
      public String getTLSProvider() {
          return getString(TLS_PROVIDER, "OpenSSL");
      }
  
      /**
       * Set TLS Provider (JDK or OpenSSL).
       *
       * @param provider
       *            TLS Provider type
       * @return Client Configuration
       */
      public T setTLSProvider(String provider) {
          setProperty(TLS_PROVIDER, provider);
          return getThis();
      }
  
      /**
       * Whether the client will send an TLS certificate on TLS-handshake.
       *
       * @see #setTLSClientAuthentication(boolean)
       * @return whether TLS is enabled on the bookie or not.
       */
      public boolean getTLSClientAuthentication() {
          return getBoolean(TLS_CLIENT_AUTHENTICATION, false);
      }
  
      /**
       * Specify whether the client will send an TLS certificate on TLS-handshake.
       *
       * @param enabled
       *            Whether to send a certificate or not
       * @return client configuration
       */
      public T setTLSClientAuthentication(boolean enabled) {
          setProperty(TLS_CLIENT_AUTHENTICATION, enabled);
          return getThis();
      }
  
      /**
       * Set tls certificate files refresh duration in seconds.
       *
       * @param certFilesRefreshSec
       *            tls certificate files refresh duration in seconds (set 0 to
       *            disable auto refresh)
       * @return current configuration
       */
      public T setTLSCertFilesRefreshDurationSeconds(long certFilesRefreshSec) {
          setProperty(TLS_CERT_FILES_REFRESH_DURATION_SECONDS, certFilesRefreshSec);
          return getThis();
      }
  
      /**
       * Get tls certificate files refresh duration in seconds.
       *
       * @return tls certificate files refresh duration in seconds. Default 0
       *         to disable auto refresh.
       *
       */
      public long getTLSCertFilesRefreshDurationSeconds() {
          return getLong(TLS_CERT_FILES_REFRESH_DURATION_SECONDS, 0);
      }
  
      /**
       * Set the list of enabled TLS cipher suites. Leave null not to override default JDK list. This list will be passed
       * to {@link SSLEngine#setEnabledCipherSuites(java.lang.String[]) }. Please refer to official JDK JavaDocs
       *
       * @param list
       *            comma separated list of enabled TLS cipher suites
       * @return current configuration
       */
      public T setTLSEnabledCipherSuites(
              String list) {
          setProperty(TLS_ENABLED_CIPHER_SUITES, list);
          return getThis();
      }
  
      /**
       * Get the list of enabled TLS cipher suites.
       *
       * @return this list of enabled TLS cipher suites
       *
       * @see #setTLSEnabledCipherSuites(java.lang.String)
       */
      public String getTLSEnabledCipherSuites() {
          return getString(TLS_ENABLED_CIPHER_SUITES, null);
      }
  
      /**
       * Set the list of enabled TLS protocols. Leave null not to override default JDK list. This list will be passed to
       * {@link SSLEngine#setEnabledProtocols(java.lang.String[]) }. Please refer to official JDK JavaDocs
       *
       * @param list
       *            comma separated list of enabled TLS cipher suites
       * @return current configuration
       */
      public T setTLSEnabledProtocols(
              String list) {
          setProperty(TLS_ENABLED_PROTOCOLS, list);
          return getThis();
      }
  
      /**
       * Get the list of enabled TLS protocols.
       *
       * @return the list of enabled TLS protocols.
       *
       * @see #setTLSEnabledProtocols(java.lang.String)
       */
      public String getTLSEnabledProtocols() {
          return getString(TLS_ENABLED_PROTOCOLS, null);
      }
  
      /**
       * Set the minimum number of racks per write quorum.
       */
      public void setMinNumRacksPerWriteQuorum(int minNumRacksPerWriteQuorum) {
          setProperty(MIN_NUM_RACKS_PER_WRITE_QUORUM, minNumRacksPerWriteQuorum);
      }
  
      /**
       * Get the minimum number of racks per write quorum.
       */
      public int getMinNumRacksPerWriteQuorum() {
          return getInteger(MIN_NUM_RACKS_PER_WRITE_QUORUM, 2);
      }
  
      /**
       * Set the minimum number of zones per write quorum in
       * ZoneAwarePlacementPolicy.
       */
      public void setMinNumZonesPerWriteQuorum(int minNumZonesPerWriteQuorum) {
          setProperty(MIN_NUM_ZONES_PER_WRITE_QUORUM, minNumZonesPerWriteQuorum);
      }
  
      /**
       * Get the minimum number of zones per write quorum in
       * ZoneAwarePlacementPolicy.
       */
      public int getMinNumZonesPerWriteQuorum() {
          return getInteger(MIN_NUM_ZONES_PER_WRITE_QUORUM, 2);
      }
  
      /**
       * Set the desired number of zones per write quorum in
       * ZoneAwarePlacementPolicy.
       */
      public void setDesiredNumZonesPerWriteQuorum(int desiredNumZonesPerWriteQuorum) {
          setProperty(DESIRED_NUM_ZONES_PER_WRITE_QUORUM, desiredNumZonesPerWriteQuorum);
      }
  
      /**
       * Get the desired number of zones per write quorum in
       * ZoneAwarePlacementPolicy.
       */
      public int getDesiredNumZonesPerWriteQuorum() {
          return getInteger(DESIRED_NUM_ZONES_PER_WRITE_QUORUM, 3);
      }
  
      /**
       * Set the flag to enforce strict zoneaware placement.
       *
       * <p>in ZoneawareEnsemblePlacementPolicy if strict placement is enabled then
       * minZones/desiredZones in writeQuorum would be maintained otherwise it
       * will pick nodes randomly.
       */
      public void setEnforceStrictZoneawarePlacement(boolean enforceStrictZoneawarePlacement) {
          setProperty(ENFORCE_STRICT_ZONEAWARE_PLACEMENT, enforceStrictZoneawarePlacement);
      }
  
      /**
       * Get the flag to enforce strict zoneaware placement.
       *
       * <p>in ZoneawareEnsemblePlacementPolicy if strict placement is enabled then
       * minZones/desiredZones in writeQuorum would be maintained otherwise it
       * will pick nodes randomly.
       */
      public boolean getEnforceStrictZoneawarePlacement() {
          return getBoolean(ENFORCE_STRICT_ZONEAWARE_PLACEMENT, true);
      }
  
      /**
       * Set the flag to enforce minimum number of racks per write quorum.
       */
      public void setEnforceMinNumRacksPerWriteQuorum(boolean enforceMinNumRacksPerWriteQuorum) {
          setProperty(ENFORCE_MIN_NUM_RACKS_PER_WRITE_QUORUM, enforceMinNumRacksPerWriteQuorum);
      }
  
      /**
       * Get the flag which enforces the minimum number of racks per write quorum.
       */
      public boolean getEnforceMinNumRacksPerWriteQuorum() {
          return getBoolean(ENFORCE_MIN_NUM_RACKS_PER_WRITE_QUORUM, false);
      }
  
      /**
       * Set the flag to enforce minimum number of fault domains for write.
       */
      public void setEnforceMinNumFaultDomainsForWrite(boolean enforceMinNumFaultDomainsForWrite) {
          setProperty(ENFORCE_MIN_NUM_FAULT_DOMAINS_FOR_WRITE, enforceMinNumFaultDomainsForWrite);
      }
  
      /**
       * Get the flag to enforce minimum number of fault domains for write.
       */
      public boolean getEnforceMinNumFaultDomainsForWrite() {
          return getBoolean(ENFORCE_MIN_NUM_FAULT_DOMAINS_FOR_WRITE, false);
      }
  
      /**
       * Sets the flag to ignore usage of localnode in placement policy.
       */
      public void setIgnoreLocalNodeInPlacementPolicy(boolean ignoreLocalNodeInPlacementPolicy) {
          setProperty(IGNORE_LOCAL_NODE_IN_PLACEMENT_POLICY, ignoreLocalNodeInPlacementPolicy);
      }
  
      /**
       * Whether to ignore localnode in placementpolicy.
       */
      public boolean getIgnoreLocalNodeInPlacementPolicy() {
          return getBoolean(IGNORE_LOCAL_NODE_IN_PLACEMENT_POLICY, false);
      }
  
      /**
       * Enable the Auditor to use system time as underreplicated ledger mark
       * time.
       *
       * <p>If this is enabled, Auditor will write a ctime field into the
       * underreplicated ledger znode.
       *
       * @param enabled
       *            flag to enable/disable Auditor using system time as
       *            underreplicated ledger mark time.
       */
      public T setStoreSystemTimeAsLedgerUnderreplicatedMarkTime(boolean enabled) {
          setProperty(STORE_SYSTEMTIME_AS_LEDGER_UNDERREPLICATED_MARK_TIME, enabled);
          return getThis();
      }
  
      /**
       * Return the flag that indicates whether auditor is using system time as
       * underreplicated ledger mark time.
       *
       * @return the flag that indicates whether auditor is using system time as
       *         underreplicated ledger mark time.
       */
      public boolean getStoreSystemTimeAsLedgerUnderreplicatedMarkTime() {
          return getBoolean(STORE_SYSTEMTIME_AS_LEDGER_UNDERREPLICATED_MARK_TIME, true);
      }
  
      /**
       * Whether to preserve MDC for tasks in Executor.
       *
       * @return flag to enable/disable MDC preservation in Executor.
       */
      public boolean getPreserveMdcForTaskExecution() {
          return getBoolean(PRESERVE_MDC_FOR_TASK_EXECUTION, false);
      }
  
      /**
       * Whether to preserve MDC for tasks in Executor.
       *
       * @param enabled
       *          flag to enable/disable MDC preservation in Executor.
       * @return configuration.
       */
      public T setPreserveMdcForTaskExecution(boolean enabled) {
          setProperty(PRESERVE_MDC_FOR_TASK_EXECUTION, enabled);
          return getThis();
      }
  
      /**
       * @return the configured pooling policy for the allocator.
       */
      public PoolingPolicy getAllocatorPoolingPolicy() {
          return PoolingPolicy.valueOf(this.getString(ALLOCATOR_POOLING_POLICY, PoolingPolicy.PooledDirect.toString()));
      }
  
      /**
       * Define the memory pooling policy.
       *
       * <p>Default is {@link PoolingPolicy#PooledDirect}
       *
       * @param poolingPolicy
       *            the memory pooling policy
       * @return configuration object.
       */
      public T setAllocatorPoolingPolicy(PoolingPolicy poolingPolicy) {
          this.setProperty(ALLOCATOR_POOLING_POLICY, poolingPolicy.toString());
          return getThis();
      }
  
      /**
       * @return the configured pooling concurrency for the allocator.
       */
      public int getAllocatorPoolingConcurrency() {
          return this.getInteger(ALLOCATOR_POOLING_CONCURRENCY, PooledByteBufAllocator.defaultNumDirectArena());
      }
  
      /**
       * Controls the amount of concurrency for the memory pool.
       *
       * <p>Default is to have a number of allocator arenas equals to 2 * CPUS.
       *
       * <p>Decreasing this number will reduce the amount of memory overhead, at the
       * expense of increased allocation contention.
       *
       * @param concurrency
       *            the concurrency level to use for the allocator pool
       * @return configuration object.
       */
      public T setAllocatorPoolingConcurrency(int concurrency) {
          this.setProperty(ALLOCATOR_POOLING_POLICY, concurrency);
          return getThis();
      }
  
      /**
       * @return the configured ouf of memory policy for the allocator.
       */
      public OutOfMemoryPolicy getAllocatorOutOfMemoryPolicy() {
          return OutOfMemoryPolicy
                  .valueOf(this.getString(ALLOCATOR_OOM_POLICY, OutOfMemoryPolicy.FallbackToHeap.toString()));
      }
  
      /**
       * Define the memory allocator out of memory policy.
       *
       * <p>Default is {@link OutOfMemoryPolicy#FallbackToHeap}
       *
       * @param oomPolicy
       *            the "out-of-memory" policy for the memory allocator
       * @return configuration object.
       */
      public T setAllocatorOutOfMemoryPolicy(OutOfMemoryPolicy oomPolicy) {
          this.setProperty(ALLOCATOR_OOM_POLICY, oomPolicy.toString());
          return getThis();
      }
  
      /**
       * Return the configured leak detection policy for the allocator.
       */
      public LeakDetectionPolicy getAllocatorLeakDetectionPolicy() {
          //see: https://lists.apache.org/thread/d3zw8bxhlg0wxfhocyjglq0nbxrww3sg
          String nettyLevelStr = System.getProperty("io.netty.leakDetectionLevel", LeakDetectionPolicy.Disabled.name());
          nettyLevelStr = System.getProperty("io.netty.leakDetection.level", nettyLevelStr);
          String bkLevelStr = getString(ALLOCATOR_LEAK_DETECTION_POLICY, LeakDetectionPolicy.Disabled.name());
          LeakDetectionPolicy nettyLevel = LeakDetectionPolicy.parseLevel(nettyLevelStr);
          LeakDetectionPolicy bkLevel = LeakDetectionPolicy.parseLevel(bkLevelStr);
          if (nettyLevel.ordinal() >= bkLevel.ordinal()) {
              return nettyLevel;
          } else {
              return bkLevel;
          }
      }
  
      /**
       * Enable the leak detection for the allocator.
       *
       * <p>Default is {@link LeakDetectionPolicy#Disabled}
       *
       * @param leakDetectionPolicy
       *            the leak detection policy for the memory allocator
       * @return configuration object.
       */
      public T setAllocatorLeakDetectionPolicy(LeakDetectionPolicy leakDetectionPolicy) {
          this.setProperty(ALLOCATOR_LEAK_DETECTION_POLICY, leakDetectionPolicy.toString());
          return getThis();
      }
  
      public T setExitOnOutOfMemory(boolean exitOnOutOfMemory) {
          this.setProperty(ALLOCATOR_EXIT_ON_OUT_OF_MEMORY, exitOnOutOfMemory);
          return getThis();
      }
  
      public boolean exitOnOutOfMemory() {
          return getBoolean(ALLOCATOR_EXIT_ON_OUT_OF_MEMORY, false);
      }
  
      /**
       * Return whether the busy-wait is enabled for BookKeeper and Netty IO threads.
       *
       * <p>Default is false
       *
       * @return the value of the option
       */
      public boolean isBusyWaitEnabled() {
          return getBoolean(ENABLE_BUSY_WAIT, false);
      }
  
      /**
       * Option to enable busy-wait settings.
       *
       * <p>Default is false.
       *
       * <p>WARNING: This option will enable spin-waiting on executors and IO threads
       * in order to reduce latency during context switches. The spinning will
       * consume 100% CPU even when bookie is not doing any work. It is
       * recommended to reduce the number of threads in the main workers pool
       * ({@link ClientConfiguration#setNumWorkerThreads(int)}) and Netty event
       * loop {@link ClientConfiguration#setNumIOThreads(int)} to only have few
       * CPU cores busy.
       * </p>
       *
       * @param busyWaitEnabled
       *            if enabled, use spin-waiting strategy to reduce latency in
       *            context switches
       *
       * @see #isBusyWaitEnabled()
       */
      public T setBusyWaitEnabled(boolean busyWaitEnabled) {
          setProperty(ENABLE_BUSY_WAIT, busyWaitEnabled);
          return getThis();
      }
  
      /**
       * Return the flag indicating whether to limit stats logging.
       *
       * @return
       *      the boolean flag indicating whether to limit stats logging
       */
      public boolean getLimitStatsLogging() {
          return getBoolean(LIMIT_STATS_LOGGING, true);
      }
  
      /**
       * Sets flag to limit the stats logging.
       *
       * @param limitStatsLogging
       *          flag to limit the stats logging.
       * @return configuration.
       */
      public T setLimitStatsLogging(boolean limitStatsLogging) {
          setProperty(LIMIT_STATS_LOGGING, limitStatsLogging);
          return getThis();
      }
  
      /**
       * Get the bytes rate of re-replication.
       * Default value is -1 which it means entries will replicated without any throttling activity.
       *
       * @return bytes rate of re-replication.
       */
      public int getReplicationRateByBytes() {
          return getInt(REPLICATION_RATE_BY_BYTES, -1);
      }
  
      /**
       * Set the bytes rate of re-replication.
       *
       * @param rate bytes rate of re-replication.
       *
       * @return ClientConfiguration
       */
      public T setReplicationRateByBytes(int rate) {
          this.setProperty(REPLICATION_RATE_BY_BYTES, rate);
          return getThis();
      }
  
      /**
       * get the max tasks can be acquired per second of re-replication.
       * @return max tasks can be acquired per second of re-replication.
       */
      public double getZkReplicationTaskRateLimit() {
          return getDouble(ZK_REPLICATION_TASK_RATE_LIMIT, 0);
      }
  
      /**
       * set the max tasks can be acquired per second of re-replication, default is 0, which means no limit.
       * Value greater than 0 will enable the rate limiting. Decimal value is allowed.
       * For example, 0.5 means 1 task per 2 seconds, 1 means 1 task per second.
       * @param zkReplicationTaskRateLimit
       * @return ClientConfiguration
       */
      public T setZkReplicationTaskRateLimit(double zkReplicationTaskRateLimit) {
          setProperty(ZK_REPLICATION_TASK_RATE_LIMIT, zkReplicationTaskRateLimit);
          return getThis();
      }
  
      /**
       * Trickery to allow inheritance with fluent style.
       */
      protected abstract T getThis();
  
      /**
       * returns the string representation of json format of this config.
       *
       * @return
       * @throws ParseJsonException
       */
      public String asJson() throws ParseJsonException {
          return JsonUtil.toJson(toMap());
      }
  
      private Map<String, Object> toMap() {
          Map<String, Object> configMap = new HashMap<>();
          Iterator<String> iterator = this.getKeys();
          while (iterator.hasNext()) {
              String key = iterator.next();
              Object property = this.getProperty(key);
              if (property != null) {
                  configMap.put(key, property.toString());
              }
          }
          return configMap;
      }
  }
  