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

import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.commons.lang3.StringUtils.isEmpty;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.StatisticsFactory;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.NoSubscriptionServersAvailableException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionService;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolFactory;
import org.apache.geode.cache.client.ServerConnectivityException;
import org.apache.geode.cache.client.SocketFactory;
import org.apache.geode.cache.client.SubscriptionNotEnabledException;
import org.apache.geode.cache.client.internal.pooling.ConnectionManager;
import org.apache.geode.cache.client.internal.pooling.ConnectionManagerImpl;
import org.apache.geode.cache.query.QueryService;
import org.apache.geode.cache.query.internal.DefaultQueryService;
import org.apache.geode.cache.wan.GatewaySender;
import org.apache.geode.distributed.PoolCancelledException;
import org.apache.geode.distributed.internal.DistributionConfig;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ServerLocation;
import org.apache.geode.distributed.internal.ServerLocationAndMemberId;
import org.apache.geode.distributed.internal.tcpserver.HostAndPort;
import org.apache.geode.internal.admin.ClientStatsManager;
import org.apache.geode.internal.cache.EventID;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PoolFactoryImpl;
import org.apache.geode.internal.cache.PoolManagerImpl;
import org.apache.geode.internal.cache.PoolStats;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.logging.CoreLoggingExecutors;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
import org.apache.geode.internal.statistics.DummyStatisticsFactory;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Manages the client side of client to server connections and client queues.
 *
 * @since GemFire 5.7
 */
public class PoolImpl implements InternalPool {

  public static final String ON_DISCONNECT_CLEAR_PDXTYPEIDS =
      GeodeGlossary.GEMFIRE_PREFIX + "ON_DISCONNECT_CLEAR_PDXTYPEIDS";

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

  public static final long SHUTDOWN_TIMEOUT =
      Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "PoolImpl.SHUTDOWN_TIMEOUT", 30000);

  private static final int BACKGROUND_TASK_POOL_SIZE = Integer
      .getInteger(GeodeGlossary.GEMFIRE_PREFIX + "PoolImpl.BACKGROUND_TASK_POOL_SIZE", 20);

  private static final int BACKGROUND_TASK_POOL_KEEP_ALIVE = Integer.getInteger(
      GeodeGlossary.GEMFIRE_PREFIX + "PoolImpl.BACKGROUND_TASK_POOL_KEEP_ALIVE", 1000);

  /**
   * For durable client tests only. Connection Sources read this flag and return an empty list of
   * servers.
   */
  @MutableForTesting
  static volatile boolean TEST_DURABLE_IS_NET_DOWN = false;

  private final String name;
  private final int socketConnectTimeout;
  private final int freeConnectionTimeout;
  private final int serverConnectionTimeout;
  private final int loadConditioningInterval;
  private final int socketBufferSize;
  @Deprecated
  private final boolean threadLocalConnections;
  private final int readTimeout;
  private final boolean subscriptionEnabled;
  private final boolean prSingleHopEnabled;
  private final int subscriptionRedundancyLevel;
  private final int subscriptionMessageTrackingTimeout;
  private final int subscriptionAckInterval;
  private final int subscriptionTimeoutMultiplier;
  private final String serverGroup;
  private final List<HostAndPort> locatorAddresses;
  private final List<HostAndPort> locators;
  private final List<HostAndPort> servers;
  private final boolean startDisabled;
  private final boolean usedByGateway;
  private final int maxConnections;
  private final int minConnections;
  private final int retryAttempts;
  private final long idleTimeout;
  private final long pingInterval;
  private final int statisticInterval;
  private final boolean multiuserSecureModeEnabled;

  private final ConnectionSource source;
  private final ConnectionManager manager;
  private final QueueManager queueManager;
  protected final EndpointManager endpointManager;
  private final PoolManagerImpl pm;
  protected final InternalLogWriter securityLogWriter;
  protected volatile boolean destroyed;
  private final PoolStats stats;
  private ScheduledExecutorService backgroundProcessor;
  private final OpExecutorImpl executor;
  private final RegisterInterestTracker riTracker = new RegisterInterestTracker();
  private final InternalDistributedSystem distributedSystem;
  private InternalCache cache;

  private final ClientProxyMembershipID proxyId;
  protected final CancelCriterion cancelCriterion;
  private final ConnectionFactoryImpl connectionFactory;

  private final ArrayList<ProxyCache> proxyCacheList;

  private final GatewaySender gatewaySender;

  private boolean keepAlive = false;
  private static final Object simpleLock = new Object();

  public static final int PRIMARY_QUEUE_NOT_AVAILABLE = -2;
  public static final int PRIMARY_QUEUE_TIMED_OUT = -1;
  private final AtomicInteger primaryQueueSize = new AtomicInteger(PRIMARY_QUEUE_NOT_AVAILABLE);

  private final ThreadsMonitoring threadMonitoring;
  private final SocketFactory socketFactory;

  public static PoolImpl create(PoolManagerImpl pm, String name, Pool attributes,
      List<HostAndPort> locatorAddresses, InternalDistributedSystem distributedSystem,
      InternalCache cache, ThreadsMonitoring tMonitoring) {
    PoolImpl pool =
        new PoolImpl(pm, name, attributes, locatorAddresses, distributedSystem, cache, tMonitoring);
    pool.finishCreate(pm);
    return pool;
  }

  public boolean isUsedByGateway() {
    return usedByGateway;
  }

  /**
   * @since GemFire 5.7
   */
  private void finishCreate(PoolManagerImpl pm) {
    pm.register(this);
    try {
      start();
    } catch (RuntimeException e) {
      try {
        destroy(false);
      } catch (RuntimeException ignore) {
        // do nothing
      }
      throw e;
    }
  }

  protected PoolImpl(PoolManagerImpl pm, String name, Pool attributes,
      List<HostAndPort> locatorAddresses, InternalDistributedSystem distributedSystem,
      InternalCache cache, ThreadsMonitoring threadMonitoring) {
    this.pm = pm;
    this.name = name;
    this.locatorAddresses = locatorAddresses;
    if (distributedSystem == null) {
      throw new IllegalStateException(
          "Distributed System must be created before creating pool");
    }
    this.distributedSystem = distributedSystem;
    this.cache = cache;
    this.threadMonitoring = threadMonitoring;

    socketConnectTimeout = attributes.getSocketConnectTimeout();
    freeConnectionTimeout = attributes.getFreeConnectionTimeout();
    serverConnectionTimeout = attributes.getServerConnectionTimeout();
    loadConditioningInterval = attributes.getLoadConditioningInterval();
    socketBufferSize = attributes.getSocketBufferSize();
    threadLocalConnections = attributes.getThreadLocalConnections();
    readTimeout = attributes.getReadTimeout();
    minConnections = attributes.getMinConnections();
    maxConnections = attributes.getMaxConnections();
    retryAttempts = attributes.getRetryAttempts();
    idleTimeout = attributes.getIdleTimeout();
    pingInterval = attributes.getPingInterval();
    statisticInterval = attributes.getStatisticInterval();
    subscriptionEnabled = attributes.getSubscriptionEnabled();
    prSingleHopEnabled = attributes.getPRSingleHopEnabled();
    subscriptionRedundancyLevel = attributes.getSubscriptionRedundancy();
    subscriptionMessageTrackingTimeout = attributes.getSubscriptionMessageTrackingTimeout();
    subscriptionAckInterval = attributes.getSubscriptionAckInterval();
    subscriptionTimeoutMultiplier = attributes.getSubscriptionTimeoutMultiplier();
    socketFactory = attributes.getSocketFactory();
    if (subscriptionTimeoutMultiplier < 0) {
      throw new IllegalArgumentException(
          "The subscription timeout multiplier must not be negative");
    }
    serverGroup = attributes.getServerGroup();
    multiuserSecureModeEnabled = attributes.getMultiuserAuthentication();
    locators = attributes.getLocators().stream()
        .map(x -> new HostAndPort(x.getHostString(), x.getPort())).collect(Collectors.toList());
    servers = attributes.getServers().stream()
        .map(x -> new HostAndPort(x.getHostString(), x.getPort())).collect(
            Collectors.toList());
    startDisabled =
        ((PoolFactoryImpl.PoolAttributes) attributes).startDisabled || !pm.isNormal();
    usedByGateway = ((PoolFactoryImpl.PoolAttributes) attributes).isGateway();
    gatewaySender = ((PoolFactoryImpl.PoolAttributes) attributes).getGatewaySender();
    securityLogWriter = distributedSystem.getSecurityInternalLogWriter();
    if (!distributedSystem.getConfig().getStatisticSamplingEnabled() && statisticInterval > 0) {
      logger.info("statistic-sampling must be enabled for sampling rate of {} to take affect",
          statisticInterval);
    }
    cancelCriterion = new Stopper();
    if (Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "SPECIAL_DURABLE")) {
      ClientProxyMembershipID.setPoolName(name);
      proxyId = ClientProxyMembershipID.getNewProxyMembership(distributedSystem);
      ClientProxyMembershipID.setPoolName(null);
    } else {
      proxyId = ClientProxyMembershipID.getNewProxyMembership(distributedSystem);
    }
    StatisticsFactory statFactory;
    if (gatewaySender != null) {
      statFactory = new DummyStatisticsFactory();
    } else {
      statFactory = distributedSystem;
    }
    stats = startDisabled ? null
        : new PoolStats(statFactory, getName() + "->"
            + (isEmpty(serverGroup) ? "[any servers]" : "[" + getServerGroup() + "]"));

    source =
        getSourceImpl(((PoolFactoryImpl.PoolAttributes) attributes).locatorCallback, socketFactory);
    endpointManager = new EndpointManagerImpl(name, distributedSystem, cancelCriterion,
        stats);
    connectionFactory = new ConnectionFactoryImpl(source, endpointManager, distributedSystem,
        socketBufferSize, socketConnectTimeout, readTimeout, proxyId, cancelCriterion,
        usedByGateway, gatewaySender, pingInterval, multiuserSecureModeEnabled, this,
        distributedSystem.getConfig());
    if (subscriptionEnabled) {
      queueManager = new QueueManagerImpl(this, endpointManager, source, connectionFactory,
          subscriptionRedundancyLevel, pingInterval, securityLogWriter, proxyId);
    } else {
      queueManager = null;
    }

    manager = new ConnectionManagerImpl(name, connectionFactory, endpointManager, maxConnections,
        minConnections, idleTimeout, loadConditioningInterval, securityLogWriter, pingInterval,
        cancelCriterion, getStats());
    // Fix for 43468 - make sure we check the cache cancel criterion if we get
    // an exception, by passing in the poolOrCache stopper
    executor = new OpExecutorImpl(manager, queueManager, endpointManager, riTracker, retryAttempts,
        freeConnectionTimeout, serverConnectionTimeout, new PoolOrCacheStopper(), this);
    if (multiuserSecureModeEnabled) {
      proxyCacheList = new ArrayList<>();
    } else {
      proxyCacheList = null;
    }
  }

  /**
   * Return true if the given Pool is compatible with these attributes. Currently this does what
   * equals would but in the future we might decide to weaken the compatibility contract.
   *
   * @since GemFire 6.5
   */
  public boolean isCompatible(Pool p) {
    if (p == null) {
      return false;
    }
    return getFreeConnectionTimeout() == p.getFreeConnectionTimeout()
        && getServerConnectionTimeout() == p.getServerConnectionTimeout()
        && getSocketConnectTimeout() == p.getSocketConnectTimeout()
        && getLoadConditioningInterval() == p.getLoadConditioningInterval()
        && getSocketBufferSize() == p.getSocketBufferSize()
        && getMinConnections() == p.getMinConnections()
        && getMaxConnections() == p.getMaxConnections() && getIdleTimeout() == p.getIdleTimeout()
        && getPingInterval() == p.getPingInterval()
        && getStatisticInterval() == p.getStatisticInterval()
        && getRetryAttempts() == p.getRetryAttempts()
        && getThreadLocalConnections() == p.getThreadLocalConnections()
        && getReadTimeout() == p.getReadTimeout()
        && getSubscriptionEnabled() == p.getSubscriptionEnabled()
        && getPRSingleHopEnabled() == p.getPRSingleHopEnabled()
        && getSubscriptionRedundancy() == p.getSubscriptionRedundancy()
        && getSubscriptionMessageTrackingTimeout() == p.getSubscriptionMessageTrackingTimeout()
        && getSubscriptionAckInterval() == p.getSubscriptionAckInterval()
        && getServerGroup().equals(p.getServerGroup())
        && getMultiuserAuthentication() == p.getMultiuserAuthentication()
        && getLocators().equals(p.getLocators()) && getServers().equals(p.getServers());
  }

  private void start() {
    if (startDisabled) {
      return;
    }

    final boolean isDebugEnabled = logger.isDebugEnabled();
    if (isDebugEnabled) {
      List locators = getLocators();
      if (!locators.isEmpty()) {
        logger.debug("PoolImpl - starting pool with locators: {}", locators);
      } else {
        logger.debug("PoolImpl -starting pool with servers: {}", getServers());
      }
    }

    final String timerName = "poolTimer-" + getName() + "-";
    backgroundProcessor = CoreLoggingExecutors.newScheduledThreadPool(BACKGROUND_TASK_POOL_SIZE,
        BACKGROUND_TASK_POOL_KEEP_ALIVE, MILLISECONDS, timerName, threadMonitoring);
    source.start(this);
    connectionFactory.start(backgroundProcessor);
    endpointManager.addListener(new InstantiatorRecoveryListener(backgroundProcessor, this));

    if (!multiuserSecureModeEnabled) {
      endpointManager.addListener(new DataSerializerRecoveryListener(backgroundProcessor, this));
    }

    if (Boolean.getBoolean(ON_DISCONNECT_CLEAR_PDXTYPEIDS)) {
      endpointManager.addListener(new PdxRegistryRecoveryListener(this));
    }
    endpointManager.addListener(new LiveServerPinger(this));

    manager.start(backgroundProcessor);
    if (queueManager != null) {
      if (isDebugEnabled) {
        logger.debug("starting queueManager");
      }
      queueManager.start(backgroundProcessor);
    }
    if (isDebugEnabled) {
      logger.debug("scheduling pings every {} milliseconds", pingInterval);
    }


    if (statisticInterval > 0 && distributedSystem.getConfig().getStatisticSamplingEnabled()) {
      backgroundProcessor.scheduleWithFixedDelay(new PublishClientStatsTask(), statisticInterval,
          statisticInterval, MILLISECONDS);
    }
    // LOG: changed from config to info
    logger.info("Pool {} started with multiuser-authentication={}",
        new Object[] {name, multiuserSecureModeEnabled});
  }

  /**
   * Returns the cancellation criterion for this proxy
   *
   * @return the cancellation criterion
   */
  @Override
  public CancelCriterion getCancelCriterion() {
    return cancelCriterion;
  }

  @Override
  public void setupServerAffinity(boolean allowFailover) {
    executor.setupServerAffinity(allowFailover);
  }

  @Override
  public void releaseServerAffinity() {
    executor.releaseServerAffinity();
  }

  @Override
  public String getName() {
    return name;
  }

  @Override
  public int getSocketConnectTimeout() {
    return socketConnectTimeout;
  }

  @Override
  public int getFreeConnectionTimeout() {
    return freeConnectionTimeout;
  }

  @Override
  public int getServerConnectionTimeout() {
    return serverConnectionTimeout;
  }

  @Override
  public int getLoadConditioningInterval() {
    return loadConditioningInterval;
  }

  @Override
  public int getMaxConnections() {
    return maxConnections;
  }

  @Override
  public int getMinConnections() {
    return minConnections;
  }

  @Override
  public int getRetryAttempts() {
    return retryAttempts;
  }

  @Override
  public long getIdleTimeout() {
    return idleTimeout;
  }

  @Override
  public long getPingInterval() {
    return pingInterval;
  }

  @Override
  public int getStatisticInterval() {
    return statisticInterval;
  }

  @Override
  public int getSocketBufferSize() {
    return socketBufferSize;
  }

  @Override
  public boolean getThreadLocalConnections() {
    return threadLocalConnections;
  }

  @Override
  public int getReadTimeout() {
    return readTimeout;
  }

  @Override
  public boolean getSubscriptionEnabled() {
    return subscriptionEnabled;
  }

  @Override
  public boolean getPRSingleHopEnabled() {
    return prSingleHopEnabled;
  }

  @Override
  public int getSubscriptionRedundancy() {
    return subscriptionRedundancyLevel;
  }

  @Override
  public int getSubscriptionMessageTrackingTimeout() {
    return subscriptionMessageTrackingTimeout;
  }

  @Override
  public int getSubscriptionAckInterval() {
    return subscriptionAckInterval;
  }

  @Override
  public String getServerGroup() {
    return serverGroup;
  }

  @Override
  public boolean getMultiuserAuthentication() {
    return multiuserSecureModeEnabled;
  }

  @Override
  public List<InetSocketAddress> getLocators() {
    return locators.stream().map(x -> x.getSocketInetAddress()).collect(Collectors.toList());
  }

  @Override
  public List<InetSocketAddress> getOnlineLocators() {
    return source.getOnlineLocators();
  }

  @Override
  public List<InetSocketAddress> getServers() {
    return servers.stream().map(x -> x.getSocketInetAddress()).collect(Collectors.toList());
  }

  public GatewaySender getGatewaySender() {
    return gatewaySender;
  }

  @Override
  public void destroy() {
    destroy(false);
  }

  @Override
  public String toString() {
    return getClass().getSimpleName() + '@' + System.identityHashCode(this) + " name=" + getName();
  }

  @Override
  public void destroy(boolean keepAlive) {
    int cnt = getAttachCount();
    this.keepAlive = keepAlive;
    boolean SPECIAL_DURABLE =
        Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "SPECIAL_DURABLE");
    if (cnt > 0) {
      // special case to allow closing durable client pool under the keep alive flag
      // closing regions prior to closing pool can cause them to unregister interest
      if (SPECIAL_DURABLE) {
        synchronized (simpleLock) {
          try {
            if (cache == null && distributedSystem != null) {
              cache = distributedSystem.getCache();
              if (cache == null) {
                throw new IllegalStateException("Cache must be created before creating pool");
              }
            }
            if (!cache.isClosed() && getPoolOrCacheCancelInProgress() == null) {
              Set<Region<?, ?>> regions = cache.rootRegions();
              for (Region<?, ?> roots : regions) {
                Set<Region<?, ?>> subregions = roots.subregions(true);
                for (Region<?, ?> subroots : subregions) {
                  if (!subroots.isDestroyed() && subroots.getAttributes().getPoolName() != null
                      && subroots.getAttributes().getPoolName().equals(name)) {
                    if (logger.isDebugEnabled()) {
                      logger.debug(
                          "PoolImpl.destroy[ Region connected count:{} Region subroot closing:{} Pool Name:{} ]",
                          cnt, subroots.getName(), name);
                    }
                    subroots.close();
                  }
                }

                if (!roots.isDestroyed() && roots.getAttributes().getPoolName() != null
                    && roots.getAttributes().getPoolName().equals(name)) {
                  if (logger.isDebugEnabled()) {
                    logger.debug(
                        "PoolImpl.destroy[ Region connected count:{} Region root closing:{} Pool Name:{} ]",
                        cnt, roots.getName(), name);
                  }
                  roots.close();
                }
              }
            }
          } catch (Exception e) {
            if (logger.isDebugEnabled()) {
              logger.debug(e.getMessage(), e);
            }
          }
        }
      }

      cnt = getAttachCount();
      if (cnt > 0) {
        throw new IllegalStateException(
            String.format("Pool could not be destroyed because it is still in use by %s regions",
                cnt));
      }
    }
    if (pm.unregister(this)) {
      basicDestroy(keepAlive);
    }
  }

  /**
   * Destroys this pool but does not unregister it. This is used by the PoolManagerImpl when it
   * wants to close all its pools.
   */
  public synchronized void basicDestroy(boolean keepAlive) {
    if (!isDestroyed()) {
      destroyed = true;
      this.keepAlive = keepAlive;
      // LOG: changed from config to info
      logger.info("Destroying connection pool {}", name);

      try {
        if (backgroundProcessor != null) {
          backgroundProcessor.shutdown();
          if (!backgroundProcessor.awaitTermination(SHUTDOWN_TIMEOUT, MILLISECONDS)) {
            logger.warn("Timeout waiting for background tasks to complete.");
          }
        }
      } catch (RuntimeException e) {
        logger.error("Error encountered while stopping backgroundProcessor.", e);
      } catch (InterruptedException e) {
        logger.error("Interrupted while stopping backgroundProcessor", e);
      }

      try {
        if (source != null) {
          source.stop();
        }
      } catch (RuntimeException e) {
        logger.error("Error encountered while stopping connection source.", e);
      }

      try {
        if (queueManager != null) {
          queueManager.close(keepAlive);
        }
      } catch (RuntimeException e) {
        logger.error(
            "Error encountered while stopping subscription manager",
            e);
      }

      try {
        if (manager != null) {
          manager.close(keepAlive);
        }
      } catch (RuntimeException e) {
        logger.error("Error encountered while stopping connection manager.", e);
      }

      try {
        endpointManager.close();
      } catch (RuntimeException e) {
        logger.error("Error encountered while stopping endpoint manager", e);
      }

      try {
        if (stats != null) {
          stats.close();
        }
      } catch (RuntimeException e) {
        logger.error("Error while closing statistics", e);
      }
    }
  }

  @Override
  public boolean isDestroyed() {
    return destroyed;
  }


  private ConnectionSource getSourceImpl(LocatorDiscoveryCallback locatorDiscoveryCallback,
      SocketFactory socketFactory) {
    List<InetSocketAddress> locators = getLocators();
    if (locators.isEmpty()) {
      return new ExplicitConnectionSourceImpl(getServers());
    } else {
      AutoConnectionSourceImpl source = new AutoConnectionSourceImpl(locatorAddresses,
          getServerGroup(), socketConnectTimeout, socketFactory);
      if (locatorDiscoveryCallback != null) {
        source.setLocatorDiscoveryCallback(locatorDiscoveryCallback);
      }
      return source;
    }
  }

  /**
   * Used internally by xml parsing code.
   */
  public void sameAs(Object obj) {
    if (!(obj instanceof PoolImpl)) {
      throw new RuntimeException(String
          .format("%s is not the same as %s because it should have been a PoolImpl", this, obj));
    }
    PoolImpl other = (PoolImpl) obj;
    if (!getName().equals(other.getName())) {
      throw new RuntimeException(
          String.format("Pool %s are different", "names"));
    }
    if (getSocketConnectTimeout() != other.getSocketConnectTimeout()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "socketConnectimeout"));
    }
    if (getFreeConnectionTimeout() != other.getFreeConnectionTimeout()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "connectionTimeout"));
    }
    if (getServerConnectionTimeout() != other.getServerConnectionTimeout()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "serverConnectionTimeout"));
    }
    if (getLoadConditioningInterval() != other.getLoadConditioningInterval()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "connectionLifetime"));
    }
    if (getSocketBufferSize() != other.getSocketBufferSize()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "socketBufferSize"));
    }
    if (getThreadLocalConnections() != other.getThreadLocalConnections()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "threadLocalConnections"));
    }
    if (getReadTimeout() != other.getReadTimeout()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "readTimeout"));
    }
    if (getMinConnections() != other.getMinConnections()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "MinConnections"));
    }
    if (getMaxConnections() != other.getMaxConnections()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "MaxConnections"));
    }
    if (getRetryAttempts() != other.getRetryAttempts()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "RetryAttempts"));
    }
    if (getIdleTimeout() != other.getIdleTimeout()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "IdleTimeout"));
    }
    if (getPingInterval() != other.getPingInterval()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "PingInterval"));
    }
    if (getStatisticInterval() != other.getStatisticInterval()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "StatisticInterval"));
    }
    if (getSubscriptionAckInterval() != other.getSubscriptionAckInterval()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "subscriptionAckInterval"));
    }
    if (getSubscriptionEnabled() != other.getSubscriptionEnabled()) {
      throw new RuntimeException(
          String.format("Pool %s is different", "subscriptionEnabled"));
    }
    if (getSubscriptionMessageTrackingTimeout() != other.getSubscriptionMessageTrackingTimeout()) {
      throw new RuntimeException(String.format("Pool %s is different",
          "subscriptionMessageTrackingTimeout"));
    }
    if (getSubscriptionRedundancy() != other.getSubscriptionRedundancy()) {
      throw new RuntimeException(String.format("Pool %s is different",
          "subscriptionRedundancyLevel"));
    }
    if (!getServerGroup().equals(other.getServerGroup())) {
      throw new RuntimeException(
          String.format("Pool %s is different", "serverGroup"));
    }
    if (!getLocators().equals(other.getLocators())) {
      throw new RuntimeException(
          String.format("Pool %s are different", "locators"));
    }
    if (!getServers().equals(other.getServers())) {
      throw new RuntimeException(
          String.format("Pool %s are different", "servers"));
    }
    // ignore startDisabled
  }

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


  /**
   * Execute the given op on the servers that this pool connects to. This method is responsible for
   * retrying the op if an attempt fails. It will only execute it once and on one server.
   *
   * @param op the operation to execute
   * @return the result of execution if any; null if not
   * @since GemFire 5.7
   */
  @Override
  public Object execute(Op op) {
    // if(multiuser)
    // get a server from threadlocal cache else throw cacheWriterException
    // executeOn(ServerLocation server, Op op, boolean accessed,boolean onlyUseExistingCnx)

    // Retries are ignored here. FIX IT - FIXED.
    // But this may lead to a user getting authenticated on all servers, even if
    // a single server could have serviced all its requests.
    authenticateIfRequired(null, op);
    return executor.execute(op);
  }

  /**
   * Execute the given op on the servers that this pool connects to. This method is responsible for
   * retrying the op if an attempt fails. It will only execute it once and on one server.
   *
   * @param op the operation to execute
   * @param retries how many times to retry the operation
   * @return the result of execution if any; null if not
   * @since GemFire 5.7
   */
  @Override
  public Object execute(Op op, int retries) {
    authenticateIfRequired(null, op);
    return executor.execute(op, retries);
  }

  /**
   * Execute the given op on the given server.
   *
   * @param server the server to do the execution on
   * @param op the operation to execute
   * @return the result of execution if any; null if not
   */
  @Override
  public Object executeOn(ServerLocation server, Op op) {
    authenticateIfRequired(server, op);
    return executor.executeOn(server, op);
  }

  /**
   * Execute the given op on the given server.
   *
   * @param server the server to do the execution on
   * @param op the operation to execute
   * @param accessed true if the connection is accessed by this execute
   * @return the result of execution if any; null if not
   */
  @Override
  public Object executeOn(ServerLocation server, Op op, boolean accessed,
      boolean onlyUseExistingCnx) {
    authenticateIfRequired(server, op);
    return executor.executeOn(server, op, accessed, onlyUseExistingCnx);
  }

  /**
   * Execute the given op on the given connection.
   *
   * @param con the connection to do the execution on
   * @param op the operation to execute
   * @return the result of execution if any; null if not
   */
  @Override
  public Object executeOn(Connection con, Op op) {
    authenticateIfRequired(con.getServer(), op);
    return executor.executeOn(con, op);
  }

  @Override
  public Object executeOn(Connection con, Op op, boolean timeoutFatal) {
    return executor.executeOn(con, op, timeoutFatal);
  }

  /**
   * Execute the given op on all the servers that have server-to-client queues for this pool
   *
   * @param op the operation to execute
   * @return the result of execution if any; null if not
   * @since GemFire 5.7
   */
  @Override
  public Object executeOnQueuesAndReturnPrimaryResult(Op op) {
    authenticateOnAllServers(op);
    return executor.executeOnQueuesAndReturnPrimaryResult(op);
  }

  @Override
  public void executeOnAllQueueServers(Op op)
      throws NoSubscriptionServersAvailableException, SubscriptionNotEnabledException {
    authenticateOnAllServers(op);
    executor.executeOnAllQueueServers(op);
  }

  /**
   * Execute the given op on the current primary server.
   *
   * @param op the operation to execute
   * @return the result of execution if any; null if not
   */
  @Override
  public Object executeOnPrimary(Op op) {
    return executor.executeOnPrimary(op);
  }

  @Override
