/*
 * 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.internal.cache.tier.sockets;

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.geode.distributed.ConfigurationProperties.SECURITY_CLIENT_ACCESSOR_PP;
import static org.apache.geode.internal.cache.tier.CommunicationMode.ClientToServerForQueue;
import static org.apache.geode.internal.cache.tier.sockets.Handshake.REPLY_REFUSED;

import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.io.OutputStream;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;

import io.micrometer.core.instrument.MeterRegistry;
import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.ToDataException;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.wan.GatewayTransportFilter;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.LonerDistributionManager;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.internal.HeapDataOutputStream;
import org.apache.geode.internal.SystemTimer;
import org.apache.geode.internal.cache.BucketAdvisor;
import org.apache.geode.internal.cache.BucketAdvisor.BucketProfile;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.partitioned.AllBucketProfilesUpdateMessage;
import org.apache.geode.internal.cache.tier.Acceptor;
import org.apache.geode.internal.cache.tier.CachedRegionHelper;
import org.apache.geode.internal.cache.tier.CommunicationMode;
import org.apache.geode.internal.cache.tier.OverflowAttributes;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier.CacheClientNotifierProvider;
import org.apache.geode.internal.cache.tier.sockets.ClientHealthMonitor.ClientHealthMonitorProvider;
import org.apache.geode.internal.cache.wan.GatewayReceiverStats;
import org.apache.geode.internal.inet.LocalHostUtil;
import org.apache.geode.internal.logging.CoreLoggingExecutors;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
import org.apache.geode.internal.net.SocketCloser;
import org.apache.geode.internal.net.SocketCreator;
import org.apache.geode.internal.security.SecurityService;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.statistics.StatisticsClock;
import org.apache.geode.internal.statistics.StatisticsClockFactory;
import org.apache.geode.internal.tcp.ConnectionTable;
import org.apache.geode.internal.util.ArrayUtils;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.executors.LoggingThreadFactory.CommandWrapper;
import org.apache.geode.logging.internal.executors.LoggingThreadFactory.ThreadInitializer;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * Implements the acceptor thread on the cache server. Accepts connections from the edge and starts
 * up threads to process requests from these.
 *
 * @since GemFire 2.0.2
 */
@SuppressWarnings("deprecation")
public class AcceptorImpl implements Acceptor, Runnable {
  private static final Logger logger = LogService.getLogger();

  private static final boolean isJRockit = System.getProperty("java.vm.name").contains("JRockit");
  private static final int HANDSHAKER_DEFAULT_POOL_SIZE = 4;
  private static final int CLIENT_QUEUE_INITIALIZATION_POOL_SIZE = 16;

  private final CacheServerStats stats;
  private final int maxConnections;
  private final int maxThreads;
  private final int maximumTimeBetweenPings;

  private final ExecutorService pool;
  /**
   * A pool used to process handshakes.
   */
  private final ExecutorService hsPool;

  /**
   * A pool used to process client-queue-initializations.
   */
  private final ExecutorService clientQueueInitPool;

  /**
   * The port on which this acceptor listens for client connections
   */
  private final int localPort;

  /**
   * The server socket that handles requests for connections
   */
  private ServerSocket serverSock;

  /**
   * The GemFire cache served up by this acceptor
   */
  private final InternalCache cache;

  /**
   * Caches region information
   */
  private final CachedRegionHelper crHelper;

  /**
   * A lock to prevent close from occurring while creating a ServerConnection
   */
  private final Object syncLock = new Object();

  /**
   * THE selector for the cache server; null if no selector.
   */
  private final Selector selector;

  /**
   * Used for managing direct byte buffer for client comms; null if no selector.
   */
  private final LinkedBlockingQueue<ByteBuffer> commBufferQueue;
  /**
   * Used to timeout accepted sockets that we are waiting for the handshake packet
   */
  private final SystemTimer hsTimer;
  /**
   * A queue used to feed register requests to the selector; null if no selector.
   */
  private final LinkedBlockingQueue<ServerConnection> selectorQueue;
  /**
   * All the objects currently registered with selector.
   */
  private final Set<ServerConnection> selectorRegistrations;
  /**
   * tcpNoDelay setting for outgoing sockets
   */
  private final boolean tcpNoDelay;

  /**
   * The name of a system property that sets the hand shake timeout (in milliseconds). This is how
   * long a client will wait to hear back from a server.
   */
  private static final String HANDSHAKE_TIMEOUT_PROPERTY_NAME = "BridgeServer.handShakeTimeout";

  /**
   * The default value of the {@link #HANDSHAKE_TIMEOUT_PROPERTY_NAME} system property.
   */
  private static final int DEFAULT_HANDSHAKE_TIMEOUT_MS = 59000;

  /**
   * Test value for handshake timeout
   */
  private static final int handshakeTimeout =
      Integer.getInteger(HANDSHAKE_TIMEOUT_PROPERTY_NAME, DEFAULT_HANDSHAKE_TIMEOUT_MS);

  /**
   * The name of a system property that sets the accept timeout (in milliseconds). This is how long
   * a server will wait to get its first byte from a client it has just accepted.
   */
  public static final String ACCEPT_TIMEOUT_PROPERTY_NAME = "BridgeServer.acceptTimeout";

  /**
   * The default value of the {@link #ACCEPT_TIMEOUT_PROPERTY_NAME} system property.
   */
  private static final int DEFAULT_ACCEPT_TIMEOUT_MS = 9900;

  /**
   * Test value for accept timeout
   */
  private final int acceptTimeout =
      Integer.getInteger(ACCEPT_TIMEOUT_PROPERTY_NAME, DEFAULT_ACCEPT_TIMEOUT_MS);

  /**
   * The minimum value of max-connections
   */
  static final int MINIMUM_MAX_CONNECTIONS = 16;

  /**
   * The buffer size for server-side sockets.
   */
  private final int socketBufferSize;

  /**
   * Notifies clients of updates
   */
  private final CacheClientNotifier clientNotifier;

  /**
   * The default value of the {@link ServerSocket} {@link #BACKLOG_PROPERTY_NAME}system property
   */
  private static final int DEFAULT_BACKLOG = 1280;

  /**
   * The system property name for setting the {@link ServerSocket}backlog
   */
  private static final String BACKLOG_PROPERTY_NAME = "BridgeServer.backlog";

  /**
   * The name of a system property that Defines the time interval (in nano-seconds) with which
   * checkRegisteredKeys function can be called.
   */
  private static final String CHECK_REGISTERED_KEYS_INTERVAL_NAME =
      "check-registered-keys-interval-ns";

  /**
   * The default value of {@link #CHECK_REGISTERED_KEYS_INTERVAL_NAME} system property.
   */
  private static final int DEFAULT_CHECK_REGISTERED_KEYS_INTERVAL_NS = 0;

  /**
   * Set value of check registered keys interval
   */
  private final long checkRegisteredKeysInterval = Long
      .getLong(CHECK_REGISTERED_KEYS_INTERVAL_NAME, DEFAULT_CHECK_REGISTERED_KEYS_INTERVAL_NS);

  /**
   * Current number of ServerConnection instances that are CLIENT_TO_SERVER cons.
   */
  private final AtomicInteger clientServerCnxCount = new AtomicInteger();

  /**
   * Has this acceptor been shut down
   */
  private volatile boolean shutdownStarted;

  /**
   * The thread that runs the acceptor
   */
  private Thread thread;

  /**
   * The thread that runs the selector loop if any
   */
  private Thread selectorThread;

  /**
   * Controls updates to {@link #allSCs}
   */
  private final Object allSCsLock = new Object();

  /**
   * List of ServerConnection.
   *
   * Instances added when constructed; removed when terminated.
   *
   * guarded.By {@link #allSCsLock}
   */
  private final Set<ServerConnection> allSCs = new HashSet<>();

  /**
   * List of ServerConnections, for {@link #emergencyClose()}
   *
   * guarded.By {@link #allSCsLock}
   */
  private volatile ServerConnection[] allSCList = new ServerConnection[0];

  /**
   * The ip address or host name this acceptor is to bind to; {@code null} or "" indicates it
   * will listen on all local addresses.
   *
   * @since GemFire 5.7
   */
  private final String bindHostName;

  /**
   * A listener for connect/disconnect events
   */
  private final ConnectionListener connectionListener;

  /**
   * The client health monitor tracking connections for this acceptor
   */
  private final ClientHealthMonitor healthMonitor;

  /**
   * bridge's setting of notifyBySubscription
   */
  private final boolean notifyBySubscription;

  /**
   * The AcceptorImpl identifier, used to identify the clients connected to this Acceptor.
   */
  private long acceptorId;

  @MakeNotStatic
  private static boolean isAuthenticationRequired;

  @MakeNotStatic
  private static boolean isPostAuthzCallbackPresent;

  private final boolean isGatewayReceiver;

  private final List<GatewayTransportFilter> gatewayTransportFilters;
  private final StatisticsClock statisticsClock;

  private final SocketCreator socketCreator;

  private final SecurityService securityService;

  private final SocketCloser socketCloser = new SocketCloser();

  /**
   * Constructs an AcceptorImpl for use within a CacheServer.
   *
   * <p>
   * Initializes this acceptor thread to listen for connections on the given port.
   *
   * @param port The port on which this acceptor listens for connections. If {@code 0}, a
   *        random port will be chosen.
   * @param bindHostName The ip address or host name this acceptor listens on for connections. If
   *        {@code null} or "" then all local addresses are used
   * @param socketBufferSize The buffer size for server-side sockets
   * @param maximumTimeBetweenPings The maximum time between client pings. This value is used by the
   *        {@code ClientHealthMonitor} to monitor the health of this server's clients.
   * @param internalCache The GemFire cache whose contents is served to clients
   * @param maxConnections the maximum number of connections allowed in the server pool
   * @param maxThreads the maximum number of threads allowed in the server pool
   * @param securityService the SecurityService to use for authentication and authorization
   *
   * @see ClientHealthMonitor
   *
   * @since GemFire 5.7
   */
  AcceptorImpl(final int port, final String bindHostName, final boolean notifyBySubscription,
      final int socketBufferSize, final int maximumTimeBetweenPings,
      final InternalCache internalCache, final int maxConnections, final int maxThreads,
      final int maximumMessageCount, final int messageTimeToLive,
      final ConnectionListener connectionListener, final OverflowAttributes overflowAttributes,
      final boolean tcpNoDelay,
      final long timeLimitMillis, final SecurityService securityService,
      final Supplier<SocketCreator> socketCreatorSupplier,
      final CacheClientNotifierProvider cacheClientNotifierProvider,
      final ClientHealthMonitorProvider clientHealthMonitorProvider) throws IOException {
    this(port, bindHostName, notifyBySubscription, socketBufferSize, maximumTimeBetweenPings,
        internalCache, maxConnections, maxThreads, maximumMessageCount, messageTimeToLive,
        connectionListener, overflowAttributes, tcpNoDelay,
        timeLimitMillis, securityService, socketCreatorSupplier, cacheClientNotifierProvider,
        clientHealthMonitorProvider, false, Collections.emptyList(),
        StatisticsClockFactory.disabledClock());
  }

  /**
   * Constructs an AcceptorImpl for use within a GatewayReceiver.
   *
   * <p>
   * Initializes this acceptor thread to listen for connections on the given port.
   *
   * @param port The port on which this acceptor listens for connections. If {@code 0}, a
   *        random port will be chosen.
   * @param notifyBySubscription the bridges setting whether if it to be notified by subscription
   * @param maximumMessageCount maximum message count setting in the Cache Client Notifier
   * @param messageTimeToLive message time to live setting in the Cache Client Notifier
   * @param connectionListener listener to detect if connect or disconnect events
   * @param overflowAttributes overflow attributes of Cache Client Notifier
   * @param tcpNoDelay TCP delay for the outgoing sockets
   * @param timeLimitMillis time limit to wait attemping to bind to a server socket
   * @param socketCreatorSupplier socket creator for the server connection
   * @param cacheClientNotifierProvider collection of cache client notifiers
   * @param clientHealthMonitorProvider collection of clinet health monitors
   * @param isGatewayReceiver flag to determine if member is gateway receiver
   * @param statisticsClock maintains the JVM's clock
   * @param bindHostName The ip address or host name this acceptor listens on for connections. If
   *        {@code null} or "" then all local addresses are used
   * @param socketBufferSize The buffer size for server-side sockets
   * @param maximumTimeBetweenPings The maximum time between client pings. This value is used by the
   *        {@code ClientHealthMonitor} to monitor the health of this server's clients.
   * @param internalCache The GemFire cache whose contents is served to clients
   * @param maxConnections the maximum number of connections allowed in the server pool
   * @param maxThreads the maximum number of threads allowed in the server pool
   * @param securityService the SecurityService to use for authentication and authorization
   * @param gatewayTransportFilters List of GatewayTransportFilters
   */
  AcceptorImpl(final int port, final String bindHostName, final boolean notifyBySubscription,
      final int socketBufferSize, final int maximumTimeBetweenPings,
      final InternalCache internalCache, final int maxConnections, final int maxThreads,
      final int maximumMessageCount, final int messageTimeToLive,
      final ConnectionListener connectionListener,
      final OverflowAttributes overflowAttributes,
      final boolean tcpNoDelay,
      final long timeLimitMillis, final SecurityService securityService,
      final Supplier<SocketCreator> socketCreatorSupplier,
      final CacheClientNotifierProvider cacheClientNotifierProvider,
      final ClientHealthMonitorProvider clientHealthMonitorProvider,
      final boolean isGatewayReceiver,
      final List<GatewayTransportFilter> gatewayTransportFilters,
      final StatisticsClock statisticsClock) throws IOException {
    this.securityService = securityService;
    this.statisticsClock = statisticsClock;

    this.isGatewayReceiver = isGatewayReceiver;
    this.gatewayTransportFilters = gatewayTransportFilters;

    this.bindHostName = calcBindHostName(internalCache, bindHostName);
    this.connectionListener =
        connectionListener == null ? new ConnectionListenerAdapter() : connectionListener;
    this.notifyBySubscription = notifyBySubscription;

    {
      int tmp_maxConnections = maxConnections;
      if (tmp_maxConnections < MINIMUM_MAX_CONNECTIONS) {
        tmp_maxConnections = MINIMUM_MAX_CONNECTIONS;
      }
      this.maxConnections = tmp_maxConnections;
    }
    {
      int tmp_maxThreads = maxThreads;
      if (maxThreads == CacheServer.DEFAULT_MAX_THREADS) {
        // consult system properties for 5.0.2 backwards compatibility
        if (DEPRECATED_SELECTOR) {
          tmp_maxThreads = DEPRECATED_SELECTOR_POOL_SIZE;
        }
      }
      if (tmp_maxThreads < 0) {
        tmp_maxThreads = 0;
      } else if (tmp_maxThreads > this.maxConnections) {
        tmp_maxThreads = this.maxConnections;
      }
      boolean isWindows = false;
      String os = System.getProperty("os.name");
      if (os != null) {
        if (os.contains("Windows")) {
          isWindows = true;
        }
      }
      if (tmp_maxThreads > 0 && isWindows) {
        // bug #40472 and JDK bug 6230761 - NIO can't be used with IPv6 on Windows
        if (getBindAddress() instanceof Inet6Address) {
          logger.warn(
              "Ignoring max-threads setting and using zero instead due to JRockit NIO bugs.  See GemFire bug #40198");
          tmp_maxThreads = 0;
        }
        // bug #40198 - Selector.wakeup() hangs if VM starts to exit
        if (isJRockit) {
          logger.warn(
              "Ignoring max-threads setting and using zero instead due to Java bug 6230761: NIO does not work with IPv6 on Windows.  See GemFire bug #40472");
          tmp_maxThreads = 0;
        }
      }
      this.maxThreads = tmp_maxThreads;
    }
    {
      Selector tmp_s = null;
      LinkedBlockingQueue<ServerConnection> tmp_q = null;
      LinkedBlockingQueue<ByteBuffer> tmp_commQ = null;
      Set<ServerConnection> tmp_hs = null;
      SystemTimer tmp_timer = null;
      if (isSelector()) {
        tmp_s = Selector.open(); // no longer catch ex to fix bug 36907
        tmp_q = new LinkedBlockingQueue<>();
        tmp_commQ = new LinkedBlockingQueue<>();
        tmp_hs = new HashSet<>(512);
        tmp_timer = new SystemTimer(internalCache.getDistributedSystem());
      }
      selector = tmp_s;
      selectorQueue = tmp_q;
      commBufferQueue = tmp_commQ;
      selectorRegistrations = tmp_hs;
      hsTimer = tmp_timer;
      this.tcpNoDelay = tcpNoDelay;
    }

    {
      socketCreator = socketCreatorSupplier.get();

      final InternalCache gc;
      if (getCachedRegionHelper() != null) {
        gc = getCachedRegionHelper().getCache();
      } else {
        gc = null;
      }
      final int backLog = Integer.getInteger(BACKLOG_PROPERTY_NAME, DEFAULT_BACKLOG);
      final long tilt = System.currentTimeMillis() + timeLimitMillis;

      if (isSelector()) {
        if (socketCreator.forCluster().useSSL()) {
          throw new IllegalArgumentException(
              "Selector thread pooling can not be used with client/server SSL. The selector can be disabled by setting max-threads=0.");
        }
        ServerSocketChannel channel = ServerSocketChannel.open();
        serverSock = channel.socket();
        serverSock.setReuseAddress(true);

        // Set the receive buffer size before binding the socket so that large
        // buffers will be allocated on accepted sockets (see
        // java.net.ServerSocket.setReceiverBufferSize javadocs)
        serverSock.setReceiveBufferSize(socketBufferSize);

        // fix for bug 36617. If BindException is thrown, retry after
        // sleeping. The server may have been stopped and then
        // immediately restarted, which sometimes results in a bind exception
        for (;;) {
          try {
            serverSock.bind(new InetSocketAddress(getBindAddress(), port), backLog);
            break;
          } catch (SocketException b) {
            if (System.currentTimeMillis() > tilt) {
              throw b;
            }
          }

          boolean interrupted = Thread.interrupted();
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e) {
            interrupted = true;
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
          }
          if (gc != null) {
            gc.getCancelCriterion().checkCancelInProgress(null);
          }
        } // for
      } // isSelector
      else { // !isSelector
        // fix for bug 36617. If BindException is thrown, retry after
        // sleeping. The server may have been stopped and then
        // immediately restarted, which sometimes results in a bind exception
        for (;;) {
          try {
            serverSock = socketCreator.createServerSocket(port, backLog, getBindAddress(),
                this.gatewayTransportFilters, socketBufferSize);
            break;
          } catch (SocketException e) {
            if (System.currentTimeMillis() > tilt) {
              throw e;
            }
          }

          boolean interrupted = Thread.interrupted();
          try {
            Thread.sleep(1000);
          } catch (InterruptedException e) {
            interrupted = true;
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
          }
          if (gc != null) {
            gc.getCancelCriterion().checkCancelInProgress(null);
          }
        } // for
      } // !isSelector

      if (port == 0) {
        localPort = serverSock.getLocalPort();
      } else {
        localPort = port;
      }
      {
        InternalDistributedSystem ds = internalCache.getInternalDistributedSystem();
        if (ds != null) {
          DistributionManager dm = ds.getDistributionManager();
          if (dm != null && dm.getDistributionManagerId().getMembershipPort() == 0
              && dm instanceof LonerDistributionManager) {
            // a server with a loner distribution manager - update it's port number
            ((LonerDistributionManager) dm).updateLonerPort(localPort);
          }
        }
      }
      String sockName = getServerName();
      logger.info("Cache server connection listener bound to address {} with backlog {}.",
          sockName, backLog);
      StatisticsFactory statisticsFactory =
          internalCache.getInternalDistributedSystem().getStatisticsManager();
      if (isGatewayReceiver()) {
        MeterRegistry meterRegistry = internalCache.getMeterRegistry();
        stats = GatewayReceiverStats.createGatewayReceiverStats(statisticsFactory, sockName,
            meterRegistry);
      } else {
        stats = new CacheServerStats(statisticsFactory, sockName);
      }
    }

    cache = internalCache;
    crHelper = new CachedRegionHelper(cache);

    clientNotifier =
        cacheClientNotifierProvider.get(internalCache, new ClientRegistrationEventQueueManager(),
            statisticsClock, stats, maximumMessageCount,
            messageTimeToLive, this.connectionListener, overflowAttributes, isGatewayReceiver());

    this.socketBufferSize = socketBufferSize;

    // Create the singleton ClientHealthMonitor
    this.maximumTimeBetweenPings = maximumTimeBetweenPings;
    healthMonitor = clientHealthMonitorProvider.get(internalCache, maximumTimeBetweenPings,
        clientNotifier.getStats());

    pool = initializeServerConnectionThreadPool();
    hsPool = initializeHandshakerThreadPool();
    clientQueueInitPool = initializeClientQueueInitializerThreadPool();

    isAuthenticationRequired = securityService.isClientSecurityRequired();

    String postAuthzFactoryName =
        cache.getDistributedSystem().getProperties().getProperty(SECURITY_CLIENT_ACCESSOR_PP);

    isPostAuthzCallbackPresent =
        postAuthzFactoryName != null && !postAuthzFactoryName.isEmpty();
  }

  private ExecutorService initializeHandshakerThreadPool() throws IOException {
    String threadName =
        "Handshaker " + serverSock.getInetAddress() + ":" + localPort + " Thread ";
    try {
      logger.warn("Handshaker max Pool size: " + HANDSHAKE_POOL_SIZE);
      return CoreLoggingExecutors.newThreadPoolWithSynchronousFeedThatHandlesRejection(1,
          HANDSHAKE_POOL_SIZE, 60, SECONDS, threadName,
          thread -> getStats().incAcceptThreadsCreated(), null);
    } catch (IllegalArgumentException poolInitException) {
      stats.close();
      serverSock.close();
      pool.shutdown();
      throw poolInitException;
    }
  }

  private ExecutorService initializeClientQueueInitializerThreadPool() {
    return CoreLoggingExecutors.newThreadPoolWithSynchronousFeed(
        CLIENT_QUEUE_INITIALIZATION_POOL_SIZE, 60000, MILLISECONDS,
        "Client Queue Initialization Thread ",
        command -> {
          try {
            command.run();
          } catch (CancelException e) {
            logger.debug("Client Queue Initialization was canceled.", e);
          }
        },
        getStats().getCnxPoolHelper(), getThreadMonitorObj());
  }

  private ExecutorService initializeServerConnectionThreadPool() throws IOException {
    ThreadInitializer threadInitializer = thread -> getStats().incConnectionThreadsCreated();
    CommandWrapper commandWrapper = command -> {
      try {
        command.run();
      } catch (CancelException e) { // bug 39463
        // ignore
      } finally {
        ConnectionTable.releaseThreadsSockets();
      }
    };
    try {
      String threadName = "ServerConnection on port " + localPort + " Thread ";
      if (isSelector()) {
        return CoreLoggingExecutors.newThreadPoolWithUnlimitedFeed(maxThreads, Integer.MAX_VALUE,
            MILLISECONDS, threadName, threadInitializer, commandWrapper,
            getStats().getCnxPoolHelper(), getThreadMonitorObj());
      }
      return CoreLoggingExecutors.newThreadPoolWithSynchronousFeed(MINIMUM_MAX_CONNECTIONS,
          maxConnections, 0L, SECONDS, threadName, threadInitializer, commandWrapper);
    } catch (IllegalArgumentException poolInitException) {
      stats.close();
      serverSock.close();
      throw poolInitException;
    }
  }

  private ThreadsMonitoring getThreadMonitorObj() {
    DistributionManager distributionManager = cache.getDistributionManager();
    if (distributionManager != null) {
      return distributionManager.getThreadMonitoring();
    }
    return null;
  }

  @Override
  public long getAcceptorId() {
    return acceptorId;
  }

  @Override
  public CacheServerStats getStats() {
    return stats;
  }

  /**
   * Returns true if this acceptor is using a selector to detect client events.
   */
  @Override
  public boolean isSelector() {
    return maxThreads > 0;
  }

  /**
   * This system property is only used if max-threads == 0. This is for 5.0.2 backwards
   * compatibility.
   *
   * @deprecated since 5.1 use cache-server max-threads instead
   */
  @Deprecated
  private static final boolean DEPRECATED_SELECTOR = Boolean.getBoolean("BridgeServer.SELECTOR");

  /**
   * This system property is only used if max-threads == 0. This is for 5.0.2 backwards
   * compatibility.
   *
   * @deprecated since 5.1 use cache-server max-threads instead
   */
  @Deprecated
  private final int DEPRECATED_SELECTOR_POOL_SIZE =
      Integer.getInteger("BridgeServer.SELECTOR_POOL_SIZE", 16);
  private final int HANDSHAKE_POOL_SIZE = Integer
      .getInteger("BridgeServer.HANDSHAKE_POOL_SIZE", HANDSHAKER_DEFAULT_POOL_SIZE);

  @Override
  public void start() {
    // This thread should not be a daemon to keep BridgeServers created
    // in code from exiting immediately.
    thread =
        new LoggingThread("Cache Server Acceptor " + serverSock.getInetAddress() + ":"
            + localPort + " local port: " + serverSock.getLocalPort(), false, this);
    acceptorId = thread.getId();
    thread.start();

    if (isSelector()) {
      selectorThread =
          new LoggingThread("Cache Server Selector " + serverSock.getInetAddress() + ":"
              + localPort + " local port: " + serverSock.getLocalPort(),
              false,
              this::runSelectorLoop);
      selectorThread.start();
    }
    Set<PartitionedRegion> prs = cache.getPartitionedRegions();
    for (PartitionedRegion pr : prs) {
      Map<Integer, BucketAdvisor.BucketProfile> profiles =
          new HashMap<>();
      // get all local real bucket advisors
      Map<Integer, BucketAdvisor> advisors = pr.getRegionAdvisor().getAllBucketAdvisors();
      for (Map.Entry<Integer, BucketAdvisor> entry : advisors.entrySet()) {
        BucketAdvisor advisor = entry.getValue();
        // addLocally
        BucketProfile bp = (BucketProfile) advisor.createProfile();
        advisor.updateServerBucketProfile(bp);
        // advisor.basicAddClientProfile(bp);
        profiles.put(entry.getKey(), bp);
      }
      Set recipients = pr.getRegionAdvisor().adviseAllPRNodes();
      // send it to all in one message
      ReplyProcessor21 reply = AllBucketProfilesUpdateMessage.send(recipients,
          pr.getDistributionManager(), pr.getPRId(), profiles);
      if (reply != null) {
        reply.waitForRepliesUninterruptibly();
      }
    }
  }

  @Override
  public void registerServerConnection(ServerConnection serverConnection) {
    synchronized (syncLock) {
      if (!isRunning()) {
        finishCon(serverConnection);
        return;
      }
    }
    getSelectorQueue().offer(serverConnection);
    wakeupSelector();
  }

  /**
   * wake up the selector thread
   */
  private void wakeupSelector() {
    Selector s = getSelector();
    if (s != null && s.isOpen()) {
      selector.wakeup();
    }
  }

  @Override
  public void unregisterServerConnection(ServerConnection serverConnection) {
    // removed syncLock synchronization to fix bug 37104
    synchronized (allSCsLock) {
      allSCs.remove(serverConnection);
      Iterator it = allSCs.iterator();
      ServerConnection[] again = new ServerConnection[allSCs.size()];
      for (int i = 0; i < again.length; i++) {
        again[i] = (ServerConnection) it.next();
      }
      allSCList = again;
    }
    if (!isRunning()) {
      return;
    }
    // just need to wake the selector up so it will notice our socket was closed
    wakeupSelector();
  }

  private void finishCon(ServerConnection sc) {
    if (sc != null) {
      sc.handleTermination();
    }
  }

  private void drainSelectorQueue() {
    ServerConnection sc = selectorQueue.poll();
    CancelException cce = null;
    while (sc != null) {
      try {
        finishCon(sc);
      } catch (CancelException e) {
        if (cce == null) {
          cce = e;
        }
      }
      sc = selectorQueue.poll();
    }
    for (ServerConnection selectorRegistration : selectorRegistrations) {
      try {
        finishCon(selectorRegistration);
