/*
 * 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;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.cache30.ClientServerTestCase.TEST_POOL_NAME;
import static org.apache.geode.cache30.ClientServerTestCase.configureConnectionPool;
import static org.apache.geode.cache30.ClientServerTestCase.configureConnectionPoolWithNameAndFactory;
import static org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier.getInstance;
import static org.apache.geode.logging.internal.spi.LogWriterLevel.ALL;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;
import static org.assertj.core.api.Assertions.fail;
import static org.junit.runners.MethodSorters.NAME_ASCENDING;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.io.NotSerializableException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

import org.junit.After;
import org.junit.Before;
import org.junit.FixMethodOrder;
import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.DataSerializable;
import org.apache.geode.LogWriter;
import org.apache.geode.cache.client.NoAvailableServersException;
import org.apache.geode.cache.client.Pool;
import org.apache.geode.cache.client.PoolManager;
import org.apache.geode.cache.client.internal.PoolImpl;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.cache30.CacheSerializableRunnable;
import org.apache.geode.cache30.CertifiableTestCacheListener;
import org.apache.geode.cache30.TestCacheWriter;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.cache.EntryExpiryTask;
import org.apache.geode.internal.cache.InternalCacheServer;
import org.apache.geode.internal.cache.PoolStats;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifierStats;
import org.apache.geode.internal.logging.LocalLogWriter;
import org.apache.geode.test.dunit.AsyncInvocation;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.NetworkUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.ThreadUtils;
import org.apache.geode.test.dunit.VM;
import org.apache.geode.test.dunit.cache.internal.JUnit4CacheTestCase;
import org.apache.geode.test.dunit.rules.DistributedRule;
import org.apache.geode.test.junit.categories.ClientServerTest;

/**
 * This class tests the client connection pool in GemFire. It does so by creating a cache server
 * with a cache and a pre-defined region and a data loader. The client creates the same region with
 * a pool (this happens in the controller VM). the client then spins up 10 different threads and
 * issues gets on keys. The server data loader returns the data to the client.
 * <p>
 * Test uses Groboutils TestRunnable objects to achieve multi threading behavior in the test.
 */
@Category({ClientServerTest.class})
@FixMethodOrder(NAME_ASCENDING)
public class ConnectionPoolDUnitTest extends JUnit4CacheTestCase {

  /**
   * The port on which the cache server was started in this VM
   */
  private static int bridgeServerPort;

  private static int numberOfAfterInvalidates;
  private static int numberOfAfterCreates;
  private static int numberOfAfterUpdates;

  private static final int TYPE_CREATE = 0;
  private static final int TYPE_UPDATE = 1;
  private static final int TYPE_INVALIDATE = 2;
  private static final int TYPE_DESTROY = 3;

  @Rule
  public DistributedRule distributedRule = new DistributedRule();

  private VM vm0;
  private VM vm1;
  private VM vm2;
  private VM vm3;

  @Before
  public void setup() {
    vm0 = VM.getVM(0);
    vm1 = VM.getVM(1);
    vm2 = VM.getVM(2);
    vm3 = VM.getVM(3);
  }

  @After
  public void tearDown() {
    Invoke.invokeInEveryVM(() -> {
      if (basicGetCache() != null) {
        basicGetCache().close();
      }
      PoolManager.getAll().forEach((key, value) -> value.destroy());
    });
  }

  private static PoolImpl getPool(Region r) {
    PoolImpl result = null;
    String poolName = r.getAttributes().getPoolName();
    if (poolName != null) {
      result = (PoolImpl) PoolManager.find(poolName);
    }
    return result;
  }

  private static TestCacheWriter getTestWriter(Region r) {
    return (TestCacheWriter) r.getAttributes().getCacheWriter();
  }

  private void startBridgeServer(int port)
      throws IOException {

    Cache cache = getCache();
    CacheServer bridge = cache.addCacheServer();
    bridge.setPort(port);
    bridge.setMaxThreads(0);
    bridge.setLoadPollInterval(CacheServer.DEFAULT_LOAD_POLL_INTERVAL);
    bridge.start();
    bridgeServerPort = bridge.getPort();
  }

  /**
   * Stops the cache server that serves up the given cache.
   *
   * @since GemFire 4.0
   */
  private void stopBridgeServer(Cache cache) {
    CacheServer bridge = cache.getCacheServers().iterator().next();
    bridge.stop();
    assertThat(bridge.isRunning()).isFalse();
  }

  private void createLonerDS() {
    disconnectFromDS();
    InternalDistributedSystem ds = getLonerSystem();
    assertThat(ds.getDistributionManager().getOtherDistributionManagerIds()).isEmpty();
  }

  /**
   * Returns region attributes for a <code>LOCAL</code> region
   */
  protected <K, V> RegionAttributes<K, V> getRegionAttributes() {
    AttributesFactory<K, V> factory = new AttributesFactory<>();
    factory.setScope(Scope.LOCAL);
    factory.setConcurrencyChecksEnabled(false); // test validation expects this behavior
    return factory.create();
  }

  private static class EventWrapper {
    final EntryEvent event;
    final Object key;
    final Object val;
    final Object arg;
    final int type;

    EventWrapper(EntryEvent ee, int type) {
      event = ee;
      key = ee.getKey();
      val = ee.getNewValue();
      arg = ee.getCallbackArgument();
      this.type = type;
    }

    public String toString() {
      return "EventWrapper: event=" + event + ", type=" + type;
    }
  }

  static class ControlListener extends CacheListenerAdapter<Object, Object> {
    final LinkedList<EventWrapper> events = new LinkedList<>();
    final Object CONTROL_LOCK = new Object();

    void waitWhileNotEnoughEvents(int eventCount) {
      long maxMillis = System.currentTimeMillis() + (long) 60000;
      synchronized (CONTROL_LOCK) {
        try {
          while (events.size() < eventCount) {
            long waitMillis = maxMillis - System.currentTimeMillis();
            if (waitMillis < 10) {
              break;
            }
            CONTROL_LOCK.wait(waitMillis);
          }
        } catch (InterruptedException abort) {
          fail("interrupted");
        }
      }
    }

    @Override
    public void afterCreate(EntryEvent e) {
      synchronized (CONTROL_LOCK) {
        events.add(new EventWrapper(e, TYPE_CREATE));
        CONTROL_LOCK.notifyAll();
      }
    }

    @Override
    public void afterUpdate(EntryEvent e) {
      synchronized (CONTROL_LOCK) {
        events.add(new EventWrapper(e, TYPE_UPDATE));
        CONTROL_LOCK.notifyAll();
      }
    }

    @Override
    public void afterInvalidate(EntryEvent e) {
      synchronized (CONTROL_LOCK) {
        events.add(new EventWrapper(e, TYPE_INVALIDATE));
        CONTROL_LOCK.notifyAll();
      }
    }

    @Override
    public void afterDestroy(EntryEvent e) {
      synchronized (CONTROL_LOCK) {
        events.add(new EventWrapper(e, TYPE_DESTROY));
        CONTROL_LOCK.notifyAll();
      }
    }
  }


  private void verifyServerCount(final PoolImpl pool, final int expectedCount) {
    getCache().getLogger().info("verifyServerCount expects=" + expectedCount);
    await().alias("Expecting found server count to match expected count")
        .until(() -> pool.getConnectedServerCount() == expectedCount);
  }

  /**
   * Tests that the callback argument is sent to the server
   */
  @Test
  public void test001CallbackArg() throws CacheException {
    final String name = getName();

    final Object createCallbackArg = "CREATE CALLBACK ARG";
    final Object updateCallbackArg = "PUT CALLBACK ARG";

    vm0.invoke("Create Cache Server", () -> {
      CacheWriter<Object, Object> cw = new TestCacheWriter<Object, Object>() {
        @Override
        public final void beforeUpdate2(EntryEvent event) throws CacheWriterException {
          Object beca = event.getCallbackArgument();
          assertThat(updateCallbackArg).isEqualTo(beca);
        }

        @Override
        public void beforeCreate2(EntryEvent event) throws CacheWriterException {
          Object beca = event.getCallbackArgument();
          assertThat(createCallbackArg).isEqualTo(beca);
        }
      };
      RegionFactory<Object, Object> factory = getBridgeServerRegionAttributes(null, cw);
      createRegion(name, factory);
      startBridgeServer(0);
    });
    final int port = vm0.invoke(ConnectionPoolDUnitTest::getCacheServerPort);

    vm1.invoke("Create region", () -> {
      getLonerSystem();
      getCache();
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.LOCAL);
      factory.setConcurrencyChecksEnabled(false);

      configureConnectionPool(factory, NetworkUtils.getServerHostName(),
          new int[] {port}, true, -1, -1, null);
      createRegion(name, factory);
    });

    vm1.invoke("Add entries", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < 10; i++) {
        region.create(i, "old" + i, createCallbackArg);
      }
      for (int i = 0; i < 10; i++) {
        region.put(i, "new" + i, updateCallbackArg);
      }
    });

    vm0.invoke("Check cache writer", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      TestCacheWriter writer = getTestWriter(region);
      assertThat(writer.wasInvoked()).isTrue();
    });

    vm1.invoke("Close Pool", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.localDestroyRegion();
    });
  }

  /**
   * Tests that consecutive puts have the callback assigned appropriately.
   */
  @Test
  public void test002CallbackArg2() throws CacheException {
    final String name = getName();
    final Object createCallbackArg = "CREATE CALLBACK ARG";

    vm0.invoke("Create Cache Server", () -> {
      CacheWriter<Object, Object> cacheWriter = new TestCacheWriter<Object, Object>() {
        @Override
        public void beforeCreate2(EntryEvent event) throws CacheWriterException {
          Integer key = (Integer) event.getKey();
          if (key % 2 == 0) {
            Object callbackArgument = event.getCallbackArgument();
            assertThat(createCallbackArg).isEqualTo(callbackArgument);
          } else {
            Object callbackArgument = event.getCallbackArgument();
            assertThat(callbackArgument).isNull();
          }
        }
      };
      RegionFactory<Object, Object> factory = getBridgeServerRegionAttributes(null, cacheWriter);
      createRegion(name, factory);
      startBridgeServer(0);
    });

    final int port = vm0.invoke(ConnectionPoolDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName();

    vm1.invoke("Create region", () -> {
      getLonerSystem();
      getCache();
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.LOCAL);
      factory.setConcurrencyChecksEnabled(false);
      configureConnectionPool(factory, host0, new int[] {port}, true, -1, -1, null);
      createRegion(name, factory);
    });

    vm1.invoke("Add entries", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < 10; i++) {
        if (i % 2 == 0) {
          region.create(i, "old" + i, createCallbackArg);

        } else {
          region.create(i, "old" + i);
        }
      }
    });

    vm1.invoke("Close Pool", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.localDestroyRegion();
    });

    vm0.invoke("Check cache writer", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      TestCacheWriter writer = getTestWriter(region);
      assertThat(writer.wasInvoked()).isTrue();
    });
  }

  /**
   * Tests for bug 36684 by having two cache servers with cacheloaders that should always return a
   * value and one client connected to each server reading values. If the bug exists, the clients
   * will get null sometimes.
   */
  @Test
  public void test003Bug36684() throws CacheException, InterruptedException {
    final String name = getName();

    // Create the cache servers with distributed, mirrored region
    Stream.of(vm0, vm1).forEach(vm -> {
      vm.invoke("Create Cache Server", () -> {
        CacheLoader<Object, Object> cl = new CacheLoader<Object, Object>() {
          @Override
          public Object load(LoaderHelper helper) {
            return helper.getKey();
          }

          @Override
          public void close() {

          }
        };
        RegionFactory<Object, Object> factory =
            getBridgeServerMirroredAckRegionAttributes(cl);
        createRegion(name, factory);
        startBridgeServer(0);

      });
      logger.info("before create server");
    });

    // Create cache server clients
    final int numberOfKeys = 1000;
    final String host0 = NetworkUtils.getServerHostName();
    final int vm0Port = vm0.invoke(ConnectionPoolDUnitTest::getCacheServerPort);
    final int vm1Port = vm1.invoke(ConnectionPoolDUnitTest::getCacheServerPort);
    SerializableRunnable createClient = new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        // reset all static listener variables in case this is being rerun in a subclass
        numberOfAfterInvalidates = 0;
        numberOfAfterCreates = 0;
        numberOfAfterUpdates = 0;
        // create the region
        getLonerSystem();
        RegionFactory<Object, Object> factory = getCache().createRegionFactory();
        factory.setScope(Scope.LOCAL);
        factory.setConcurrencyChecksEnabled(false); // test validation expects this behavior
        // create bridge writer
        configureConnectionPool(factory, host0, new int[] {vm0Port, vm1Port}, true, -1,
            -1, null);
        createRegion(name, factory);
      }
    };
    logger.info("before create client");
    vm2.invoke("Create Cache Server Client", createClient);
    vm3.invoke("Create Cache Server Client", createClient);

    // Initialize each client with entries (so that afterInvalidate is called)
    SerializableRunnable initializeClient = new CacheSerializableRunnable() {
      @Override
      public void run2() throws CacheException {
        // StringBuilder errors = new StringBuilder();
        numberOfAfterInvalidates = 0;
        numberOfAfterCreates = 0;
        numberOfAfterUpdates = 0;
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        for (int i = 0; i < numberOfKeys; i++) {
          String expected = "key-" + i;
          String actual = (String) region.get("key-" + i);
          assertThat(expected).isEqualTo(actual);
        }
      }
    };

    logger.info("before initialize client");
    AsyncInvocation<Void> inv2 = vm2.invokeAsync("Initialize Client", initializeClient);
    AsyncInvocation<Void> inv3 = vm3.invokeAsync("Initialize Client", initializeClient);

    inv2.await();
    inv3.await();
  }

  /**
   * Test for client connection loss with CacheLoader Exception on the server.
   */
  @Test
  public void test004ForCacheLoaderException() throws CacheException {
    final String name = getName();

    VM server = VM.getVM(0);
    VM client = VM.getVM(1);

    // Create the cache servers with distributed, mirrored region
    logger.info("before create server");

    server.invoke("Create Cache Server", () -> {
      CacheLoader<Object, Object> cl = new CacheLoader<Object, Object>() {
        @Override
        public Object load(LoaderHelper helper) {
          System.out.println("### CALLING CACHE LOADER....");
          throw new CacheLoaderException(
              "Test for CahceLoaderException causing Client connection to disconnect.");
        }

        @Override
        public void close() {}
      };
      RegionFactory<Object, Object> factory =
          getBridgeServerMirroredAckRegionAttributes(cl);
      createRegion(name, factory);
      startBridgeServer(0);
    });

    // Create cache server clients
    final int numberOfKeys = 10;
    final String host0 = NetworkUtils.getServerHostName();
    final int[] port =
        new int[] {server.invoke(ConnectionPoolDUnitTest::getCacheServerPort)};
    final String poolName = "myPool";

    logger.info("before create client");
    client.invoke("Create Cache Server Client", () -> {
      getLonerSystem();
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.LOCAL);
      factory.setConcurrencyChecksEnabled(false);
      // create bridge writer
      configureConnectionPoolWithNameAndFactory(factory, host0, port, true,
          -1, -1, null, poolName, PoolManager.createFactory(), -1, -1, -2, -1);
      createRegion(name, factory);
    });

    // Initialize each client with entries (so that afterInvalidate is called)

    logger.info("before initialize client");
    AsyncInvocation<Void> inv2 = client.invokeAsync("Initialize Client", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      PoolStats stats = ((PoolImpl) PoolManager.find(poolName)).getStats();
      int oldConnects = stats.getConnects();
      int oldDisConnects = stats.getDisConnects();
      for (int i = 0; i < numberOfKeys; i++) {
        region.get("key-" + i);
      }
      int newConnects = stats.getConnects();
      int newDisConnects = stats.getDisConnects();

      // newDisConnects);
      if (newConnects != oldConnects && newDisConnects != oldDisConnects) {
        fail("New connection has created for Server side CacheLoaderException.");
      }
    });

    ThreadUtils.join(inv2, 30 * 1000);
  }

  private void validateDS() {
    List list = InternalDistributedSystem.getExistingSystems();
    if (list.size() > 1) {
      logger.info("validateDS: size=" + list.size() + " isDedicatedAdminVM="
          + ClusterDistributionManager.isDedicatedAdminVM() + " l=" + list);
    }
    assertThat(ClusterDistributionManager.isDedicatedAdminVM()).isFalse();
    assertThat(1).isEqualTo(list.size());
  }

  /**
   * Tests the basic operations of the {@link Pool}
   *
   * @since GemFire 3.5
   */
  @Test
  public void test006Pool() throws CacheException {
    final String name = getName();

    vm0.invoke("Create Cache Server", () -> {
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.DISTRIBUTED_ACK);
      factory.setConcurrencyChecksEnabled(false);
      factory.setCacheLoader(new CacheLoader<Object, Object>() {
        @Override
        public Object load(LoaderHelper helper) {
          return helper.getKey().toString();
        }

        @Override
        public void close() {

        }
      });
      createRegion(name, factory);
      startBridgeServer(0);
    });
    final int port = vm0.invoke(ConnectionPoolDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName();
    vm1.invoke("Create region", () -> {
      getLonerSystem();
      getCache();
      validateDS();
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.LOCAL);
      factory.setConcurrencyChecksEnabled(false);
      configureConnectionPool(factory, host0, new int[] {port}, true, -1, -1, null);
      createRegion(name, factory);
    });

    vm1.invoke("Get values", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < 10; i++) {
        Object value = region.get(i);
        assertThat(String.valueOf(i)).isEqualTo(value);
      }
    });

    vm1.invoke("Update values", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);

      for (int i = 0; i < 10; i++) {
        region.put(i, i);
      }
    });

    vm2.invoke("Create region", () -> {
      getLonerSystem();
      getCache();
      validateDS();
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.LOCAL);
      factory.setConcurrencyChecksEnabled(false);
      configureConnectionPool(factory, host0, new int[] {port}, true, -1, -1, null);
      createRegion(name, factory);
    });
    vm2.invoke("Validate values", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      for (int i = 0; i < 10; i++) {
        Object value = region.get(i);
        assertThat(value).isNotNull();
        assertThat(value).isInstanceOf(Integer.class);
        assertThat(i).isEqualTo(((Integer) value).intValue());
      }
    });

    vm1.invoke("Close Pool", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      String pName = region.getAttributes().getPoolName();
      PoolImpl p = (PoolImpl) PoolManager.find(pName);
      assertThat(p.isDestroyed()).isFalse();
      assertThat(1).isEqualTo(p.getAttachCount());
      try {
        p.destroy();
        fail("expected IllegalStateException");
      } catch (IllegalStateException ignored) {
      }
      region.localDestroyRegion();
      assertThat(p.isDestroyed()).isFalse();
      assertThat(p.getAttachCount()).isEqualTo(0);
    });
  }

  /**
   * Tests the BridgeServer failover (bug 31832).
   */
  @Test
  public void test007BridgeServerFailoverCnx1() throws CacheException {
    disconnectAllFromDS();
    basicTestBridgeServerFailover(1);
  }

  /**
   * Test BridgeServer failover with connectionsPerServer set to 0
   */
  @Test
  public void test008BridgeServerFailoverCnx0() throws CacheException {
    basicTestBridgeServerFailover(0);
  }

  private void basicTestBridgeServerFailover(final int cnxCount) throws CacheException {
    final String name = getName();

    // Create two cache servers
    Stream.of(vm0, vm1).forEach(vm -> vm.invoke("Create Cache Server", () -> {
      RegionFactory<Object, Object> factory = getBridgeServerRegionAttributes(null, null);
      createRegion(name, factory);
      startBridgeServer(0);
    }));

    final int port0 = vm0.invoke(ConnectionPoolDUnitTest::getCacheServerPort);
    final String host0 = NetworkUtils.getServerHostName();

    final int port1 = vm1.invoke(ConnectionPoolDUnitTest::getCacheServerPort);

    // Create one bridge client in this VM

    vm2.invoke("Create region", () -> {
      getLonerSystem();
      getCache();
      RegionFactory<Object, Object> factory = getCache().createRegionFactory();
      factory.setScope(Scope.LOCAL);
      factory.setConcurrencyChecksEnabled(false);
      String ServerGroup = null;
      configureConnectionPoolWithNameAndFactory(factory, host0, new int[] {port0, port1}, true,
          -1, cnxCount, ServerGroup, TEST_POOL_NAME, PoolManager.createFactory(), 100, -1, -2,
          -1);
      Region<Object, Object> region = createRegion(name, factory);

      // force connections to form
      region.put("keyInit", 0);
      region.put("keyInit2", 0);
    });

    vm2.invoke("verify2Servers", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      PoolImpl pool = getPool(region);
      verifyServerCount(pool, 2);
    });

    final String expected = "java.io.IOException";
    final String addExpected =
        "<ExpectedException action=add>" + expected + "</ExpectedException>";
    final String removeExpected =
        "<ExpectedException action=remove>" + expected + "</ExpectedException>";

    vm2.invoke(() -> {
      LogWriter bgexecLogger = new LocalLogWriter(ALL.intLevel(), System.out);
      bgexecLogger.info(addExpected);
    });

    try { // make sure we removeExpected

      // Bounce the non-current server (I know that VM1 contains the non-current server
      // because ...
      vm1.invoke(() -> stopBridgeServer(getCache()));

      vm2.invoke("verify1Server", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        PoolImpl pool = getPool(region);
        verifyServerCount(pool, 1);
      });

      vm1.invoke("Restart CacheServer", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        assertThat(region).isNotNull();
        startBridgeServer(port1);
      });

      // Pause long enough for the monitor to realize the server has been bounced
      // and reconnect to it.
      vm2.invoke("verify2Servers", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        PoolImpl pool = getPool(region);
        verifyServerCount(pool, 2);
      });

    } finally {
      vm2.invoke(() -> {
        LogWriter bgexecLogger = new LocalLogWriter(ALL.intLevel(), System.out);
        bgexecLogger.info(removeExpected);
      });
    }

    // Stop the other cache server
    vm0.invoke(() -> stopBridgeServer(getCache()));

    // Run awhile
    vm2.invoke("verify1Server", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      PoolImpl pool = getPool(region);
      verifyServerCount(pool, 1);
    });

    // Close Pool
    vm2.invoke("Close Pool", () -> {
      Region<Object, Object> region = getRootRegion().getSubregion(name);
      region.localDestroyRegion();
    });
  }


  private static volatile boolean stopTestLifetimeExpire = false;

  private static volatile int baselineLifetimeCheck;
  private static volatile int baselineLifetimeExtensions;
  private static volatile int baselineLifetimeConnect;
  private static volatile int baselineLifetimeDisconnect;

  @Test
  public void basicTestLifetimeExpire()
      throws CacheException, InterruptedException {
    final String name = getName();

    AsyncInvocation<Void> putAI = null;
    AsyncInvocation<Void> putAI2 = null;

    try {

      // Create two cache servers

      vm0.invoke("Create Cache Server", () -> {
        RegionFactory<Object, Object> factory = getBridgeServerRegionAttributes(null, null);
        factory.addCacheListener(new DelayListener());
        createRegion(name, factory);
        startBridgeServer(0);

      });

      final int port0 = vm0.invoke(ConnectionPoolDUnitTest::getCacheServerPort);
      final String host0 = NetworkUtils.getServerHostName();
      vm1.invoke("Create Cache Server", () -> {
        RegionFactory<Object, Object> factory = getBridgeServerRegionAttributes(null, null);
        factory.addCacheListener(new DelayListener());
        createRegion(name, factory);
        startBridgeServer(0);

      });
      final int port1 = vm1.invoke(ConnectionPoolDUnitTest::getCacheServerPort);
      // we only had to stop it to reserve a port
      vm1.invoke("Stop CacheServer", () -> stopBridgeServer(getCache()));

      // Create one bridge client in this VM

      vm2.invoke("Create region", () -> {
        getLonerSystem();
        getCache();
        RegionFactory<Object, Object> factory = getCache().createRegionFactory();
        factory.setScope(Scope.LOCAL);
        factory.setConcurrencyChecksEnabled(false);
        configureConnectionPoolWithNameAndFactory(factory, host0, new int[] {port0, port1},
            false, -1, 0, null, TEST_POOL_NAME, PoolManager.createFactory(), 100, 500, 500, -1);

        Region<Object, Object> region = createRegion(name, factory);

        // force connections to form
        region.put("keyInit", 0);
        region.put("keyInit2", 0);
      });

      // Launch async thread that puts objects into cache. This thread will execute until
      // the test has ended.
      putAI = vm2.invokeAsync("Put objects", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        PoolImpl pool = getPool(region);
        PoolStats stats = pool.getStats();
        baselineLifetimeCheck = stats.getLoadConditioningCheck();
        baselineLifetimeExtensions = stats.getLoadConditioningExtensions();
        baselineLifetimeConnect = stats.getLoadConditioningConnect();
        baselineLifetimeDisconnect = stats.getLoadConditioningDisconnect();
        try {
          int count = 0;
          while (!stopTestLifetimeExpire) {
            count++;
            region.put("keyAI1", count);
          }
        } catch (NoAvailableServersException ex) {
          if (!stopTestLifetimeExpire) {
            throw ex;
          }

        }
      });
      putAI2 = vm2.invokeAsync("Put objects", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        try {
          int count = 0;
          while (!stopTestLifetimeExpire) {
            count++;
            region.put("keyAI2", count);
          }
        } catch (NoAvailableServersException ex) {
          if (!stopTestLifetimeExpire) {
            throw ex;
          }
        }
      });

      vm2.invoke("verify1Server", () -> {
        Region<Object, Object> region = getRootRegion().getSubregion(name);
        PoolImpl pool = getPool(region);
        final PoolStats stats = pool.getStats();
        verifyServerCount(pool, 1);

        await().until(() -> stats.getLoadConditioningCheck() >= (10 + baselineLifetimeCheck));

        // make sure no replacements are happening.
        // since we have 2 threads and 2 cnxs and 1 server
        // when lifetimes are up we should only want to connect back to the
        // server we are already connected to and thus just extend our lifetime
        assertThat(stats.getLoadConditioningCheck() >= (10 + baselineLifetimeCheck))
            .describedAs("baselineLifetimeCheck=" + baselineLifetimeCheck
                + " but stats.getLoadConditioningCheck()=" + stats.getLoadConditioningCheck())
            .isTrue();
        baselineLifetimeCheck = stats.getLoadConditioningCheck();
        assertThat(stats.getLoadConditioningExtensions())
            .isGreaterThan(baselineLifetimeExtensions);
        assertThat(stats.getLoadConditioningConnect()).isEqualTo(baselineLifetimeConnect);
        assertThat(stats.getLoadConditioningDisconnect()).isEqualTo(baselineLifetimeDisconnect);
      });

      await().until(putAI::isAlive);
      await().until(putAI2::isAlive);

    } finally {
      vm2.invoke("Stop Putters", () -> stopTestLifetimeExpire = true);

      try {
        if (putAI != null) {
          // Verify that no exception has occurred in the putter thread
          putAI.await();
        }

        if (putAI2 != null) {
          // Verify that no exception has occurred in the putter thread
          putAI.await();
        }
      } finally {
        vm2.invoke("Stop Putters", () -> stopTestLifetimeExpire = false);
        // Close Pool
        vm2.invoke("Close Pool", () -> {
          Region<Object, Object> region = getRootRegion().getSubregion(name);
          String poolName = region.getAttributes().getPoolName();
          region.localDestroyRegion();
          PoolManager.find(poolName).destroy();
        });
      }
    }
  }

  /**
   * Tests the create operation of the {@link Pool}
   *
   * @since GemFire 3.5
   */
  @Test
