/*
 * 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.geode.distributed.internal;

import static org.apache.geode.distributed.ConfigurationProperties.ACK_SEVERE_ALERT_THRESHOLD;
import static org.apache.geode.distributed.ConfigurationProperties.ACK_WAIT_THRESHOLD;
import static org.apache.geode.distributed.ConfigurationProperties.ARCHIVE_DISK_SPACE_LIMIT;
import static org.apache.geode.distributed.ConfigurationProperties.ARCHIVE_FILE_SIZE_LIMIT;
import static org.apache.geode.distributed.ConfigurationProperties.ASYNC_DISTRIBUTION_TIMEOUT;
import static org.apache.geode.distributed.ConfigurationProperties.ASYNC_MAX_QUEUE_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.ASYNC_QUEUE_TIMEOUT;
import static org.apache.geode.distributed.ConfigurationProperties.BIND_ADDRESS;
import static org.apache.geode.distributed.ConfigurationProperties.CACHE_XML_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_CONFIGURATION_DIR;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_CIPHERS;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_KEYSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_KEYSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_KEYSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.CLUSTER_SSL_TRUSTSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.CONFLATE_EVENTS;
import static org.apache.geode.distributed.ConfigurationProperties.CONSERVE_SOCKETS;
import static org.apache.geode.distributed.ConfigurationProperties.DELTA_PROPAGATION;
import static org.apache.geode.distributed.ConfigurationProperties.DEPLOY_WORKING_DIR;
import static org.apache.geode.distributed.ConfigurationProperties.DISABLE_AUTO_RECONNECT;
import static org.apache.geode.distributed.ConfigurationProperties.DISABLE_JMX;
import static org.apache.geode.distributed.ConfigurationProperties.DISABLE_TCP;
import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID;
import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_TRANSACTIONS;
import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_ID;
import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_TIMEOUT;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_CLUSTER_CONFIGURATION;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_MANAGEMENT_REST_SERVICE;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_NETWORK_PARTITION_DETECTION;
import static org.apache.geode.distributed.ConfigurationProperties.ENABLE_TIME_STATISTICS;
import static org.apache.geode.distributed.ConfigurationProperties.ENFORCE_UNIQUE_HOST;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_CIPHERS;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_KEYSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.GATEWAY_SSL_TRUSTSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.GROUPS;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_BIND_ADDRESS;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_CIPHERS;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_KEYSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_KEYSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_KEYSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.HTTP_SERVICE_SSL_TRUSTSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_ACCESS_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_BIND_ADDRESS;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_HOSTNAME_FOR_CLIENTS;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_HTTP_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_PASSWORD_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_CIPHERS;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_KEYSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_KEYSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_KEYSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_SSL_TRUSTSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_START;
import static org.apache.geode.distributed.ConfigurationProperties.JMX_MANAGER_UPDATE_RATE;
import static org.apache.geode.distributed.ConfigurationProperties.LOAD_CLUSTER_CONFIGURATION_FROM_DIR;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATOR_WAIT_TIME;
import static org.apache.geode.distributed.ConfigurationProperties.LOCK_MEMORY;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_DISK_SPACE_LIMIT;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_FILE_SIZE_LIMIT;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
import static org.apache.geode.distributed.ConfigurationProperties.MAX_NUM_RECONNECT_TRIES;
import static org.apache.geode.distributed.ConfigurationProperties.MAX_WAIT_TIME_RECONNECT;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_ADDRESS;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_FLOW_CONTROL;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_RECV_BUFFER_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_SEND_BUFFER_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.MCAST_TTL;
import static org.apache.geode.distributed.ConfigurationProperties.MEMBERSHIP_PORT_RANGE;
import static org.apache.geode.distributed.ConfigurationProperties.MEMBER_TIMEOUT;
import static org.apache.geode.distributed.ConfigurationProperties.MEMCACHED_BIND_ADDRESS;
import static org.apache.geode.distributed.ConfigurationProperties.MEMCACHED_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.MEMCACHED_PROTOCOL;
import static org.apache.geode.distributed.ConfigurationProperties.NAME;
import static org.apache.geode.distributed.ConfigurationProperties.OFF_HEAP_MEMORY_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.REDUNDANCY_ZONE;
import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS;
import static org.apache.geode.distributed.ConfigurationProperties.REMOVE_UNRESPONSIVE_CLIENT;
import static org.apache.geode.distributed.ConfigurationProperties.ROLES;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_AUTH_TOKEN_ENABLED_COMPONENTS;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_ACCESSOR;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_ACCESSOR_PP;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTHENTICATOR;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_AUTH_INIT;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_DHALGO;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_LOG_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_LOG_LEVEL;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_MANAGER;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_PEER_AUTHENTICATOR;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_PEER_AUTH_INIT;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_PEER_VERIFY_MEMBER_TIMEOUT;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_POST_PROCESSOR;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_PREFIX;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_SHIRO_INIT;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_UDP_DHALGO;
import static org.apache.geode.distributed.ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_BIND_ADDRESS;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_CIPHERS;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_KEYSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_KEYSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_KEYSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_SSL_TRUSTSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.SOCKET_BUFFER_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.SOCKET_LEASE_TIME;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_CIPHERS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_CLIENT_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_CLUSTER_ALIAS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_DEFAULT_ALIAS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_ENABLED_COMPONENTS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_ENDPOINT_IDENTIFICATION_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_GATEWAY_ALIAS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_JMX_ALIAS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_KEYSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_LOCATOR_ALIAS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_PARAMETER_EXTENSION;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_SERVER_ALIAS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_SERVER_PROTOCOLS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE_PASSWORD;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_TRUSTSTORE_TYPE;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_USE_DEFAULT_CONTEXT;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_WEB_ALIAS;
import static org.apache.geode.distributed.ConfigurationProperties.SSL_WEB_SERVICE_REQUIRE_AUTHENTICATION;
import static org.apache.geode.distributed.ConfigurationProperties.START_DEV_REST_API;
import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR;
import static org.apache.geode.distributed.ConfigurationProperties.STATISTIC_ARCHIVE_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.STATISTIC_SAMPLE_RATE;
import static org.apache.geode.distributed.ConfigurationProperties.STATISTIC_SAMPLING_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.TCP_PORT;
import static org.apache.geode.distributed.ConfigurationProperties.THREAD_MONITOR_ENABLED;
import static org.apache.geode.distributed.ConfigurationProperties.THREAD_MONITOR_INTERVAL;
import static org.apache.geode.distributed.ConfigurationProperties.THREAD_MONITOR_TIME_LIMIT;
import static org.apache.geode.distributed.ConfigurationProperties.UDP_FRAGMENT_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.UDP_RECV_BUFFER_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.UDP_SEND_BUFFER_SIZE;
import static org.apache.geode.distributed.ConfigurationProperties.USER_COMMAND_PACKAGES;
import static org.apache.geode.distributed.ConfigurationProperties.USE_CLUSTER_CONFIGURATION;
import static org.apache.geode.distributed.ConfigurationProperties.VALIDATE_SERIALIZABLE_OBJECTS;

import java.lang.reflect.Method;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Arrays;
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 java.util.StringTokenizer;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;

import org.apache.geode.InternalGemFireException;
import org.apache.geode.InvalidValueException;
import org.apache.geode.UnmodifiableException;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.internal.AbstractConfig;
import org.apache.geode.internal.ConfigSource;
import org.apache.geode.internal.admin.remote.DistributionLocatorId;
import org.apache.geode.internal.inet.LocalHostUtil;
import org.apache.geode.internal.logging.LogWriterImpl;
import org.apache.geode.internal.security.SecurableCommunicationChannel;
import org.apache.geode.logging.internal.log4j.LogLevel;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Provides an implementation of <code>DistributionConfig</code> that knows how to read the
 * configuration file.
 * <p>
 * Note that if you add a property to this interface, you should update the
 * {@link AbstractConfig#sameAs} method and the
 * {@link DistributionConfigImpl#DistributionConfigImpl(DistributionConfig) copy constructor}.
 */
@SuppressWarnings("deprecation")
public abstract class AbstractDistributionConfig extends AbstractConfig
    implements DistributionConfig {

  private static final Logger logger = LogService.getLogger();

  Object checkAttribute(String attName, Object value) {
    // first check to see if this attribute is modifiable, this also checks if the attribute is a
    // valid one.
    if (!isAttributeModifiable(attName)) {
      throw new UnmodifiableException(_getUnmodifiableMsg(attName));
    }

    ConfigAttribute attribute = attributes.get(attName);
    if (attribute == null) {
      // isAttributeModifiable already checks the validity of the attName, if reached here, then
      // they
      // must be those special attributes that starts with ssl_system_props or sys_props, no further
      // checking needed
      return value;
    }
    // for integer attribute, do the range check.
    if (attribute.type().equals(Integer.class)) {
      Integer intValue = (Integer) value;
      minMaxCheck(attName, intValue, attribute.min(), attribute.max());
    }

    Method checker = checkers.get(attName);
    if (checker == null) {
      return value;
    }

    // if specific checker exists for this attribute, call that with the value
    try {
      return checker.invoke(this, value);
    } catch (Exception e) {
      if (e instanceof RuntimeException) {
        throw (RuntimeException) e;
      }
      if (e.getCause() instanceof RuntimeException) {
        throw (RuntimeException) e.getCause();
      } else {
        throw new InternalGemFireException(
            "error invoking " + checker.getName() + " with value " + value);
      }
    }
  }

  private void minMaxCheck(String propName, int value, int minValue, int maxValue) {
    if (value < minValue) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set \"%s\" to \"%s\" because its value can not be less than \"%s\".",

              propName, value, minValue));
    } else if (value > maxValue) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set \"%s\" to \"%s\" because its value can not be greater than \"%s\".",
              propName, value, maxValue));
    }
  }

  @ConfigAttributeChecker(name = START_LOCATOR)
  protected String checkStartLocator(String value) {
    if (value != null && value.trim().length() > 0) {
      // throws IllegalArgumentException if string is malformed
      DistributionLocatorId.unmarshal(value);
    }
    return value;
  }

  @ConfigAttributeChecker(name = TCP_PORT)
  protected int checkTcpPort(int value) {
    if (getClusterSSLEnabled() && value != 0) {
      throw new IllegalArgumentException(
          String.format("Could not set %s to %s because its value must be 0 when %s is true.",
              TCP_PORT, value, CLUSTER_SSL_ENABLED));
    }
    return value;
  }

  @ConfigAttributeChecker(name = MCAST_PORT)
  protected int checkMcastPort(int value) {
    if (getClusterSSLEnabled() && value != 0) {
      throw new IllegalArgumentException(
          String.format("Could not set %s to %s because its value must be 0 when %s is true.",
              MCAST_PORT, value, CLUSTER_SSL_ENABLED));
    }
    return value;
  }

  @ConfigAttributeChecker(name = MCAST_ADDRESS)
  protected InetAddress checkMcastAddress(InetAddress value) {
    if (!value.isMulticastAddress()) {
      throw new IllegalArgumentException(
          String.format("Could not set %s to %s because it was not a multicast address.",
              MCAST_ADDRESS, value));
    }
    return value;
  }

  @ConfigAttributeChecker(name = BIND_ADDRESS)
  protected String checkBindAddress(String value) {
    if (value != null && value.length() > 0 && !LocalHostUtil.isLocalHost(value)) {
      throw new IllegalArgumentException(
          String.format(
              "The bind-address %s is not a valid address for this machine.  These are the valid addresses for this machine: %s",
              value, LocalHostUtil.getMyAddresses()));
    }
    return value;
  }

  @ConfigAttributeChecker(name = SERVER_BIND_ADDRESS)
  protected String checkServerBindAddress(String value) {
    if (value != null && value.length() > 0 && !LocalHostUtil.isLocalHost(value)) {
      throw new IllegalArgumentException(
          String.format(
              "The bind-address %s is not a valid address for this machine.  These are the valid addresses for this machine: %s",
              value, LocalHostUtil.getMyAddresses()));
    }
    return value;
  }

  @ConfigAttributeChecker(name = CLUSTER_SSL_ENABLED)
  protected boolean checkClusterSSLEnabled(boolean value) {
    if (value && getMcastPort() != 0) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set %s to true because its value must be false when %s is not 0.",
              CLUSTER_SSL_ENABLED, MCAST_PORT));
    }
    return value;
  }

  @ConfigAttributeChecker(name = HTTP_SERVICE_BIND_ADDRESS)
  protected String checkHttpServiceBindAddress(String value) {
    if (value != null && value.length() > 0 && !LocalHostUtil.isLocalHost(value)) {
      throw new IllegalArgumentException(
          String.format(
              "The bind-address %s is not a valid address for this machine.  These are the valid addresses for this machine: %s",
              value, LocalHostUtil.getMyAddresses()));
    }
    return value;
  }

  @ConfigAttributeChecker(name = DISTRIBUTED_SYSTEM_ID)
  protected int checkDistributedSystemId(int value) {
    String distributedSystemListener =
        System.getProperty(GeodeGlossary.GEMFIRE_PREFIX + "DistributedSystemListener");
    // this check is specific for Jayesh's use case of WAN BootStrapping
    if (distributedSystemListener == null) {
      if (value < MIN_DISTRIBUTED_SYSTEM_ID) {
        throw new IllegalArgumentException(
            String.format(
                "Could not set \"%s\" to \"%s\" because its value can not be less than \"%s\".",
                DISTRIBUTED_SYSTEM_ID, value,
                MIN_DISTRIBUTED_SYSTEM_ID));
      }
    }
    if (value > MAX_DISTRIBUTED_SYSTEM_ID) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set \"%s\" to \"%s\" because its value can not be greater than \"%s\".",
              DISTRIBUTED_SYSTEM_ID, value,
              MAX_DISTRIBUTED_SYSTEM_ID));
    }
    return value;
  }

  /**
   * Makes sure that the locator string used to configure discovery is valid.
   * <p>
   * Starting in 4.0, we accept locators in the format "host:port" in addition to the traditional
   * "host:bind-address[port]" format. See bug 32306.
   * <p>
   * Starting in 5.1.0.4, we accept locators in the format "host@bind-address[port]" to allow use of
   * numeric IPv6 addresses
   *
   * @return The locator string in the traditional "host:bind-address[port]" format.
   *
   * @throws IllegalArgumentException If <code>value</code> is not a valid locator configuration
   */
  @ConfigAttributeChecker(name = LOCATORS)
  protected String checkLocators(String value) {
    // validate locators value
    StringBuilder sb = new StringBuilder();

    Set<InetSocketAddress> locs = new HashSet<>();

    StringTokenizer st = new StringTokenizer(value, ",");
    boolean firstUniqueLocator = true;
    while (st.hasMoreTokens()) {
      String locator = st.nextToken();
      // string for this locator is accumulated in this buffer
      StringBuilder locatorsb = new StringBuilder();

      int portIndex = locator.indexOf('[');
      if (portIndex < 1) {
        portIndex = locator.lastIndexOf(':');
      }
      if (portIndex < 1) {
        throw new IllegalArgumentException(
            String.format("Invalid locator %s. Host name was empty.",
                value));
      }

      // starting in 5.1.0.4 we allow '@' as the bind-addr separator
      // to let people use IPv6 numeric addresses (which contain colons)
      int bindAddrIdx = locator.lastIndexOf('@', portIndex - 1);

      if (bindAddrIdx < 0) {
        bindAddrIdx = locator.lastIndexOf(':', portIndex - 1);
      }

      String host = locator.substring(0, bindAddrIdx > -1 ? bindAddrIdx : portIndex);

      if (host.indexOf(':') >= 0) {
        bindAddrIdx = locator.lastIndexOf('@');
        host = locator.substring(0, bindAddrIdx > -1 ? bindAddrIdx : portIndex);
      }

      InetAddress hostAddress = null;

      try {
        hostAddress = InetAddress.getByName(host);

      } catch (UnknownHostException ex) {
        logger.warn("Unknown locator host: " + host);
      }

      locatorsb.append(host);

      if (bindAddrIdx > -1) {
        // validate the bindAddress... (console needs this)
        String bindAddr = locator.substring(bindAddrIdx + 1, portIndex);
        try {
          hostAddress = InetAddress.getByName(bindAddr);

        } catch (UnknownHostException ex) {
          throw new IllegalArgumentException(
              String.format("Unknown locator bind address: %s",
                  bindAddr));
        }

        if (bindAddr.indexOf(':') >= 0) {
          locatorsb.append('@');
        } else {
          locatorsb.append(':');
        }
        locatorsb.append(bindAddr);
      }

      int lastIndex = locator.lastIndexOf(']');
      if (lastIndex == -1) {
        if (locator.indexOf('[') >= 0) {
          throw new IllegalArgumentException(
              String.format("Invalid locator: %s",
                  value));

        } else {
          // Using host:port syntax
          lastIndex = locator.length();
        }
      }

      String port = locator.substring(portIndex + 1, lastIndex);
      int portVal;
      try {
        portVal = Integer.parseInt(port);
        if (0 == portVal) {
          return "";
        } else if (portVal < 1 || portVal > 65535) {
          throw new IllegalArgumentException(
              String.format(
                  "Invalid locator %s. The port %s was not greater than zero and less than 65,536.",
                  value, portVal));
        }
      } catch (NumberFormatException ex) {
        throw new IllegalArgumentException(
            String.format("Invalid locator: %s", value));
      }

      locatorsb.append('[');
      locatorsb.append(port);
      locatorsb.append(']');

      // if this wasn't a duplicate, add it to the locators string
      InetSocketAddress sockAddr = new InetSocketAddress(hostAddress, portVal);
      if (!locs.contains(sockAddr)) {
        if (!firstUniqueLocator) {
          sb.append(',');
        } else {
          firstUniqueLocator = false;
        }
        locs.add(new InetSocketAddress(hostAddress, portVal));
        sb.append(locatorsb);
      }
    }

    return sb.toString();
  }

  /**
   * check a new mcast flow-control setting
   */
  @ConfigAttributeChecker(name = MCAST_FLOW_CONTROL)
  protected FlowControlParams checkMcastFlowControl(FlowControlParams params) {
    int value = params.getByteAllowance();
    if (value < MIN_FC_BYTE_ALLOWANCE) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set %s.byteAllowance to %s because its value can not be less than %s",
              MCAST_FLOW_CONTROL, value,
              MIN_FC_BYTE_ALLOWANCE));
    }
    float fvalue = params.getRechargeThreshold();
    if (fvalue < MIN_FC_RECHARGE_THRESHOLD) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set %s.rechargeThreshold to %s because its value can not be less than %s",
              MCAST_FLOW_CONTROL, fvalue,
              MIN_FC_RECHARGE_THRESHOLD));
    } else if (fvalue > MAX_FC_RECHARGE_THRESHOLD) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set %s.rechargeThreshold to %s because its value can not be greater than %s",
              MCAST_FLOW_CONTROL, fvalue,
              MAX_FC_RECHARGE_THRESHOLD));
    }
    value = params.getRechargeBlockMs();
    if (value < MIN_FC_RECHARGE_BLOCK_MS) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set %s.rechargeBlockMs to %s because its value can not be less than %s",
              MCAST_FLOW_CONTROL, value,
              MIN_FC_RECHARGE_BLOCK_MS));
    } else if (value > MAX_FC_RECHARGE_BLOCK_MS) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set %s.rechargeBlockMs to %s because its value can not be greater than %s",
              MCAST_FLOW_CONTROL, value,
              MAX_FC_RECHARGE_BLOCK_MS));
    }
    return params;
  }

  @ConfigAttributeChecker(name = MEMBERSHIP_PORT_RANGE)
  protected int[] checkMembershipPortRange(int[] value) {
    minMaxCheck(MEMBERSHIP_PORT_RANGE, value[0], 1024, value[1]);
    minMaxCheck(MEMBERSHIP_PORT_RANGE, value[1], value[0], 65535);

    // Minimum 3 ports are required to start a Gemfire data node,
    // One for each, UDP, FD_SOCk protocols and Cache Server.
    if (value[1] - value[0] < 2) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set \"%s\" to \"%s\" because its value can not be less than \"%s\".",
              MEMBERSHIP_PORT_RANGE, value[0] + "-" + value[1],
              3));
    }
    return value;
  }

  /**
   * @since GemFire 5.7
   */
  @ConfigAttributeChecker(name = CLIENT_CONFLATION_PROP_NAME)
  protected String checkClientConflation(String value) {
    if (!(value.equals(CLIENT_CONFLATION_PROP_VALUE_DEFAULT)
        || value.equals(CLIENT_CONFLATION_PROP_VALUE_ON)
        || value.equals(CLIENT_CONFLATION_PROP_VALUE_OFF))) {
      throw new IllegalArgumentException("Could not set \"" + CONFLATE_EVENTS + "\" to \"" + value
          + "\" because its value is not recognized");
    }
    return value;
  }

  @ConfigAttributeChecker(name = SECURITY_PEER_AUTH_INIT)
  protected String checkSecurityPeerAuthInit(String value) {
    if (value != null && value.length() > 0 && getMcastPort() != 0) {
      String mcastInfo = MCAST_PORT + "[" + getMcastPort() + "]";
      throw new IllegalArgumentException(
          String.format("Could not set %s to %s because %s must be 0 when security is enabled.",
              SECURITY_PEER_AUTH_INIT, value, mcastInfo));
    }
    return value;
  }

  @ConfigAttributeChecker(name = SECURITY_PEER_AUTHENTICATOR)
  protected String checkSecurityPeerAuthenticator(String value) {
    if (value != null && value.length() > 0 && getMcastPort() != 0) {
      String mcastInfo = MCAST_PORT + "[" + getMcastPort() + "]";
      throw new IllegalArgumentException(
          String.format("Could not set %s to %s because %s must be 0 when security is enabled.",
              SECURITY_PEER_AUTHENTICATOR, value, mcastInfo));
    }
    return value;
  }

  @ConfigAttributeChecker(name = SECURITY_LOG_LEVEL)
  protected int checkSecurityLogLevel(int value) {
    if (value < MIN_LOG_LEVEL) {
      throw new IllegalArgumentException(
          String.format(
              "Could not set \"%s\" to \"%s\" because its value can not be less than \"%s\".",
              SECURITY_LOG_LEVEL,
              LogWriterImpl.levelToString(value), LogWriterImpl.levelToString(MIN_LOG_LEVEL)));
    }
