geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jinmeil...@apache.org
Subject [39/50] [abbrv] incubator-geode git commit: GEODE-1111 Connection Pooling needs more tests
Date Tue, 29 Mar 2016 22:26:09 GMT
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/4ed2fd37/geode-core/src/test/java/com/gemstone/gemfire/cache/ConnectionPoolDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/com/gemstone/gemfire/cache/ConnectionPoolDUnitTest.java b/geode-core/src/test/java/com/gemstone/gemfire/cache/ConnectionPoolDUnitTest.java
new file mode 100755
index 0000000..41d48aa
--- /dev/null
+++ b/geode-core/src/test/java/com/gemstone/gemfire/cache/ConnectionPoolDUnitTest.java
@@ -0,0 +1,5871 @@
+/*=========================================================================
+ * Copyright (c) 2010-2014 Pivotal Software, Inc. All Rights Reserved.
+ * This product is protected by U.S. and international copyright
+ * and intellectual property laws. Pivotal products are covered by
+ * one or more patents listed at http://www.pivotal.io/patents.
+ *=========================================================================
+ */
+package com.gemstone.gemfire.cache;
+
+import static org.junit.runners.MethodSorters.NAME_ASCENDING;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.Set;
+
+import junit.framework.AssertionFailedError;
+
+import org.junit.FixMethodOrder;
+
+import com.gemstone.gemfire.CancelException;
+import com.gemstone.gemfire.LogWriter;
+import com.gemstone.gemfire.cache.client.NoAvailableServersException;
+import com.gemstone.gemfire.cache.client.Pool;
+import com.gemstone.gemfire.cache.client.PoolManager;
+import com.gemstone.gemfire.cache.client.internal.Endpoint;
+import com.gemstone.gemfire.cache.client.internal.PoolImpl;
+import com.gemstone.gemfire.cache.server.CacheServer;
+import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
+import com.gemstone.gemfire.cache30.ClientServerTestCase;
+import com.gemstone.gemfire.cache30.CacheSerializableRunnable;
+import com.gemstone.gemfire.cache30.CacheTestCase;
+import com.gemstone.gemfire.cache30.CertifiableTestCacheListener;
+import com.gemstone.gemfire.cache30.TestCacheLoader;
+import com.gemstone.gemfire.cache30.TestCacheWriter;
+import com.gemstone.gemfire.distributed.DistributedMember;
+import com.gemstone.gemfire.distributed.internal.DistributionManager;
+import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
+import com.gemstone.gemfire.internal.Assert;
+import com.gemstone.gemfire.internal.cache.CacheServerImpl;
+import com.gemstone.gemfire.internal.cache.EntryExpiryTask;
+import com.gemstone.gemfire.internal.cache.LocalRegion;
+import com.gemstone.gemfire.internal.cache.PoolStats;
+import com.gemstone.gemfire.internal.cache.tier.sockets.CacheClientNotifier;
+import com.gemstone.gemfire.internal.cache.tier.sockets.CacheClientNotifierStats;
+import com.gemstone.gemfire.internal.cache.tier.sockets.ClientProxyMembershipID;
+import com.gemstone.gemfire.internal.logging.InternalLogWriter;
+import com.gemstone.gemfire.internal.logging.LocalLogWriter;
+import com.gemstone.gemfire.test.dunit.AsyncInvocation;
+import com.gemstone.gemfire.test.dunit.DistributedTestCase;
+import com.gemstone.gemfire.test.dunit.Host;
+import com.gemstone.gemfire.test.dunit.Invoke;
+import com.gemstone.gemfire.test.dunit.NetworkUtils;
+import com.gemstone.gemfire.test.dunit.SerializableRunnable;
+import com.gemstone.gemfire.test.dunit.ThreadUtils;
+import com.gemstone.gemfire.test.dunit.VM;
+import com.gemstone.gemfire.test.dunit.Wait;
+import com.gemstone.gemfire.test.dunit.WaitCriterion;
+
+/**
+ * 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.
+ * Test uses Groboutils TestRunnable objects to achieve multi threading behavior
+ * in the test.
+ *
+ */
+@FixMethodOrder(NAME_ASCENDING)
+public class ConnectionPoolDUnitTest extends CacheTestCase {
+
+  private static final long serialVersionUID = 1L;
+
+  /** The port on which the bridge server was started in this VM */
+  private static int bridgeServerPort;
+
+  protected static int port = 0;
+  protected static int port2 = 0;
+
+  protected static int numberOfAfterInvalidates;
+  protected static int numberOfAfterCreates;
+  protected static int numberOfAfterUpdates;
+
+  protected final static int TYPE_CREATE = 0;
+  protected final static int TYPE_UPDATE = 1;
+  protected final static int TYPE_INVALIDATE = 2;
+  protected final static int TYPE_DESTROY = 3;
+
+  public ConnectionPoolDUnitTest(String name) {
+    super(name);
+  }
+
+  public void setUp() throws Exception {
+    super.setUp();
+    // avoid IllegalStateException from HandShake by connecting all vms to
+    // system before creating pool
+    getSystem();
+    Invoke.invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        getSystem();
+      }
+    });
+  }
+  
+  @Override
+  protected final void postTearDownCacheTestCase() throws Exception {
+    Invoke.invokeInEveryVM(new SerializableRunnable() {
+      public void run() {
+        Map pools = PoolManager.getAll();
+        if (!pools.isEmpty()) {
+          com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter().warning("found pools remaining after teardown: " + pools);
+          assertEquals(0, pools.size());
+        }
+      }
+    });
+    postTearDownConnectionPoolDUnitTest();
+  }
+  
+  protected void postTearDownConnectionPoolDUnitTest() throws Exception {
+  }
+
+  protected/*GemStoneAddition*/ static PoolImpl getPool(Region r) {
+    PoolImpl result = null;
+    String poolName = r.getAttributes().getPoolName();
+    if (poolName != null) {
+      result = (PoolImpl)PoolManager.find(poolName);
+    }
+    return result;
+  }
+  protected static TestCacheWriter getTestWriter(Region r) {
+    return (TestCacheWriter)r.getAttributes().getCacheWriter();
+  }
+  /**
+   * Create a bridge server on the given port without starting it.
+   *
+   * @since 5.0.2
+   */
+  protected void createBridgeServer(int port) throws IOException {
+    CacheServer bridge = getCache().addCacheServer();
+    bridge.setPort(port);
+    bridge.setMaxThreads(getMaxThreads());
+    bridgeServerPort = bridge.getPort();
+  }
+
+  /**
+   * Starts a bridge server on the given port, using the given
+   * deserializeValues and notifyBySubscription to serve up the
+   * given region.
+   *
+   * @since 4.0
+   */
+  protected void startBridgeServer(int port)
+    throws IOException {
+    startBridgeServer(port, -1);
+  }
+  
+  protected void startBridgeServer(int port, int socketBufferSize) throws IOException {
+    startBridgeServer(port, socketBufferSize, CacheServer.DEFAULT_LOAD_POLL_INTERVAL);
+  }
+
+  protected void startBridgeServer(int port, int socketBufferSize, long loadPollInterval)
+    throws IOException {
+
+    Cache cache = getCache();
+    CacheServer bridge = cache.addCacheServer();
+    bridge.setPort(port);
+    if (socketBufferSize != -1) {
+      bridge.setSocketBufferSize(socketBufferSize);
+    }
+    bridge.setMaxThreads(getMaxThreads());
+    bridge.setLoadPollInterval(loadPollInterval);
+    bridge.start();
+    bridgeServerPort = bridge.getPort();
+  }
+
+  /**
+   * By default return 0 which turns off selector and gives thread per cnx.
+   * Test subclasses can override to run with selector.
+   * @since 5.1
+   */
+  protected int getMaxThreads() {
+    return 0;
+  }
+
+  /**
+   * Stops the bridge server that serves up the given cache.
+   *
+   * @since 4.0
+   */
+  void stopBridgeServer(Cache cache) {
+    CacheServer bridge =
+        cache.getCacheServers().iterator().next();
+    bridge.stop();
+    assertFalse(bridge.isRunning());
+  }
+
+  void stopBridgeServers(Cache cache) {
+    CacheServer bridge = null;
+    for (Iterator bsI = cache.getCacheServers().iterator();bsI.hasNext(); ) {
+      bridge = (CacheServer) bsI.next();
+      bridge.stop();
+      assertFalse(bridge.isRunning());
+    }
+  }
+
+  private void restartBridgeServers(Cache cache) throws IOException
+  {
+    CacheServer bridge = null;
+    for (Iterator bsI = cache.getCacheServers().iterator();bsI.hasNext(); ) {
+      bridge = (CacheServer) bsI.next();
+      bridge.start();
+      assertTrue(bridge.isRunning());
+    }
+  }
+
+  protected InternalDistributedSystem createLonerDS() {
+    disconnectFromDS();
+    InternalDistributedSystem ds = getLonerSystem();
+    assertEquals(0, ds.getDistributionManager().getOtherDistributionManagerIds().size());
+    return ds;
+  }
+
+  
+  
+  /**
+   * Returns region attributes for a <code>LOCAL</code> region
+   */
+  protected RegionAttributes getRegionAttributes() {
+    AttributesFactory factory = new AttributesFactory();
+    factory.setScope(Scope.LOCAL);
+    factory.setConcurrencyChecksEnabled(false); // test validation expects this behavior
+    return factory.create();
+  }
+
+  private static String createBridgeClientConnection(String host, int[] ports) {
+    StringBuffer sb = new StringBuffer();
+    for (int i = 0; i < ports.length; i++) {
+      if (i > 0) sb.append(",");
+      sb.append("name" + i + "=");
+      sb.append(host + ":" + ports[i]);
+    }
+    return sb.toString();
+  }
+
+  private class EventWrapper {
+    public final EntryEvent event;
+    public final Object key;
+    public final Object val;
+    public final Object arg;
+    public final  int type;
+    public EventWrapper(EntryEvent ee, int type) {
+      this.event = ee;
+      this.key = ee.getKey();
+      this.val = ee.getNewValue();
+      this.arg = ee.getCallbackArgument();
+      this.type = type;
+    }
+    public String toString() {
+      return "EventWrapper: event=" + event + ", type=" + type;
+    }
+  }
+
+  protected class ControlListener extends CacheListenerAdapter {
+    public final LinkedList events = new LinkedList();
+    public final Object CONTROL_LOCK = new Object();
+
+    public boolean waitWhileNotEnoughEvents(long sleepMs, int eventCount) {
+      long maxMillis = System.currentTimeMillis() + sleepMs;
+      synchronized (this.CONTROL_LOCK) {
+        try {
+          while (this.events.size() < eventCount) {
+            long waitMillis = maxMillis - System.currentTimeMillis();
+            if (waitMillis < 10) {
+              break;
+            }
+            this.CONTROL_LOCK.wait(waitMillis);
+          }
+        } catch (InterruptedException abort) {
+          fail("interrupted");
+        }
+        return !this.events.isEmpty();
+      }
+    }
+
+    public void afterCreate(EntryEvent e) {
+      //System.out.println("afterCreate: " + e);
+      synchronized(this.CONTROL_LOCK) {
+        this.events.add(new EventWrapper(e, TYPE_CREATE));
+        this.CONTROL_LOCK.notifyAll();
+      }
+    }
+
+    public void afterUpdate(EntryEvent e) {
+      //System.out.println("afterUpdate: " + e);
+      synchronized(this.CONTROL_LOCK) {
+        this.events.add(new EventWrapper(e, TYPE_UPDATE));
+        this.CONTROL_LOCK.notifyAll();
+      }
+    }
+
+    public void afterInvalidate(EntryEvent e) {
+      //System.out.println("afterInvalidate: " + e);
+      synchronized(this.CONTROL_LOCK) {
+        this.events.add(new EventWrapper(e, TYPE_INVALIDATE));
+        this.CONTROL_LOCK.notifyAll();
+      }
+    }
+
+    public void afterDestroy(EntryEvent e) {
+      //System.out.println("afterDestroy: " + e);
+      synchronized(this.CONTROL_LOCK) {
+        this.events.add(new EventWrapper(e, TYPE_DESTROY));
+        this.CONTROL_LOCK.notifyAll();
+      }
+    }
+  }
+
+  
+
+
+  /**
+   * Create a fake EntryEvent that returns the provided region for {@link CacheEvent#getRegion()}
+   * and returns {@link com.gemstone.gemfire.cache.Operation#LOCAL_LOAD_CREATE} for {@link CacheEvent#getOperation()}
+   * @param r
+   * @return fake entry event
+   */
+  protected static EntryEvent createFakeyEntryEvent(final Region r) {
+    return new EntryEvent() {
+      public Operation getOperation()
+      {
+        return Operation.LOCAL_LOAD_CREATE; // fake out pool to exit early
+      }
+      public Region getRegion()
+      {
+        return r;
+      }
+      public Object getKey() { return null; }
+      public Object getOldValue() { return null;}
+      public boolean isOldValueAvailable() {return true;}
+      public Object getNewValue() { return null;}
+      public boolean isLocalLoad() { return false;}
+      public boolean isNetLoad() {return false;}
+      public boolean isLoad() {return true; }
+      public boolean isNetSearch() {return false;}
+      public TransactionId getTransactionId() {return null;}
+      public Object getCallbackArgument() {return null;}
+      public boolean isCallbackArgumentAvailable() {return true;}
+      public boolean isOriginRemote() {return false;}
+      public DistributedMember getDistributedMember() {return null;}
+      public boolean isExpiration() { return false;}
+      public boolean isDistributed() { return false;}
+      public boolean isBridgeEvent() {
+        return hasClientOrigin();
+      }
+      public boolean hasClientOrigin() {
+        return false;
+      }
+      public ClientProxyMembershipID getContext() {
+        return null;
+      }
+      public SerializedCacheValue getSerializedOldValue() {return null;}
+      public SerializedCacheValue getSerializedNewValue() {return null;}
+    };
+  }
+
+  public void verifyBalanced(final PoolImpl pool, int expectedServer, 
+      final int expectedConsPerServer) {
+    verifyServerCount(pool, expectedServer);
+    WaitCriterion ev = new WaitCriterion() {
+      public boolean done() {
+        return balanced(pool, expectedConsPerServer);
+      }
+      public String description() {
+        return "expected " + expectedConsPerServer
+            + " but endpoints=" + outOfBalanceReport(pool);
+      }
+    };
+    Wait.waitForCriterion(ev, 2 * 60 * 1000, 200, true);
+    assertEquals("expected " + expectedConsPerServer
+                 + " but endpoints=" + outOfBalanceReport(pool),
+                 true, balanced(pool, expectedConsPerServer));
+  }
+  protected boolean balanced(PoolImpl pool, int expectedConsPerServer) {
+    Iterator it = pool.getEndpointMap().values().iterator();
+    while (it.hasNext()) {
+      Endpoint ep = (Endpoint)it.next();
+      if (ep.getStats().getConnections() != expectedConsPerServer) {
+        return false;
+      }
+    }
+    return true;
+  }
+  
+  protected String outOfBalanceReport(PoolImpl pool) {
+    StringBuffer result = new StringBuffer();
+    Iterator it = pool.getEndpointMap().values().iterator();
+    result.append("<");
+    while (it.hasNext()) {
+      Endpoint ep = (Endpoint)it.next();
+      result.append("ep=" + ep);
+      result.append(" conCount=" + ep.getStats().getConnections());
+      if (it.hasNext()) {
+        result.append(", ");
+      }
+    }
+    result.append(">");
+    return result.toString();
+  }
+  
+  public void waitForBlacklistToClear(final PoolImpl pool) {
+    WaitCriterion ev = new WaitCriterion() {
+      public boolean done() {
+        return pool.getBlacklistedServers().size() == 0;
+      }
+      public String description() {
+        return null;
+      }
+    };
+    Wait.waitForCriterion(ev, 30 * 1000, 200, true);
+    assertEquals("unexpected blacklistedServers=" + pool.getBlacklistedServers(),
+                 0, pool.getBlacklistedServers().size());
+  }
+  
+  public void verifyServerCount(final PoolImpl pool, final int expectedCount) {
+    getCache().getLogger().info("verifyServerCount expects=" + expectedCount);
+    WaitCriterion ev = new WaitCriterion() {
+      String excuse;
+      public boolean done() {
+        int actual = pool.getConnectedServerCount();
+        if (actual == expectedCount) {
+          return true;
+        }
+        excuse = "Found only " + actual + " servers, expected " + expectedCount;
+        return false;
+      }
+      public String description() {
+        return excuse;
+      }
+    };
+    Wait.waitForCriterion(ev, 5 * 60 * 1000, 200, true);
+  }
+
+  /**
+   * Tests that the callback argument is sent to the server
+   */
+  public void test001CallbackArg() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+
+    final Object createCallbackArg = "CREATE CALLBACK ARG";
+    final Object updateCallbackArg = "PUT CALLBACK ARG";
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+
+          CacheWriter cw = new TestCacheWriter() {
+              public final void beforeUpdate2(EntryEvent event)
+                throws CacheWriterException {
+                Object beca = event.getCallbackArgument();
+                assertEquals(updateCallbackArg, beca);
+              }
+
+              public void beforeCreate2(EntryEvent event)
+                throws CacheWriterException {
+                Object beca =  event.getCallbackArgument();
+                assertEquals(createCallbackArg, beca);
+              }
+            };
+            AttributesFactory factory = getBridgeServerRegionAttributes(null, cw);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      });
+    final int port =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+
+    
+    SerializableRunnable create =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+
+            ClientServerTestCase.configureConnectionPool(factory,NetworkUtils.getServerHostName(host),port,-1,true,-1,-1, null);
+            createRegion(name, factory.create());
+          }
+        };
+
+    vm1.invoke(create);
+    vm1.invoke(new CacheSerializableRunnable("Add entries") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            region.create(new Integer(i), "old" + i, createCallbackArg);
+          }
+          for (int i = 0; i < 10; i++) {
+            region.put(new Integer(i), "new" + i, updateCallbackArg);
+        }
+        }
+      });
+
+    vm0.invoke(new CacheSerializableRunnable("Check cache writer") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          TestCacheWriter writer = getTestWriter(region);
+          assertTrue(writer.wasInvoked());
+        }
+      });
+
+  SerializableRunnable close =
+    new CacheSerializableRunnable("Close Pool") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          region.localDestroyRegion();
+        }
+    };
+
+    vm1.invoke(close);
+
+    vm0.invoke(new SerializableRunnable("Stop CacheServer") {
+        public void run() {
+          stopBridgeServer(getCache());
+        }
+    });
+
+  }
+
+  /**
+   * Tests that consecutive puts have the callback assigned
+   * appropriately.
+   */
+  public void test002CallbackArg2() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+
+    final Object createCallbackArg = "CREATE CALLBACK ARG";
+//    final Object updateCallbackArg = "PUT CALLBACK ARG";
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          CacheWriter cw = new TestCacheWriter() {
+              public void beforeCreate2(EntryEvent event)
+                throws CacheWriterException {
+                Integer key = (Integer) event.getKey();
+                if (key.intValue() % 2 == 0) {
+                  Object beca =  event.getCallbackArgument();
+                  assertEquals(createCallbackArg, beca);
+                } else {
+                  Object beca =  event.getCallbackArgument();
+                  assertNull(beca);
+                }
+              }
+            };
+            AttributesFactory factory = getBridgeServerRegionAttributes(null, cw);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      });
+    final int port =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+
+    SerializableRunnable create =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,true,-1,-1, null);
+            createRegion(name, factory.create());
+          }
+        };
+
+    vm1.invoke(create);
+    vm1.invoke(new CacheSerializableRunnable("Add entries") {
+        public void run2() throws CacheException {
+        Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            if (i % 2 == 0) {
+              region.create(new Integer(i), "old" + i, createCallbackArg);
+
+            } else {
+              region.create(new Integer(i), "old" + i);
+            }
+          }
+        }
+      });
+
+  SerializableRunnable close =
+    new CacheSerializableRunnable("Close Pool") {
+      public void run2() throws CacheException {
+        Region region = getRootRegion().getSubregion(name);
+        region.localDestroyRegion();
+      }
+    };
+
+    vm1.invoke(close);
+
+    vm0.invoke(new CacheSerializableRunnable("Check cache writer") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          TestCacheWriter writer = getTestWriter(region);
+          assertTrue(writer.wasInvoked());
+        }
+      });
+
+    vm0.invoke(new SerializableRunnable("Stop CacheServer") {
+        public void run() {
+          stopBridgeServer(getCache());
+        }
+    });
+  }
+
+  
+  /**
+   * Tests for bug 36684 by having two bridge 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. 
+   * @throws InterruptedException 
+   */
+  public void test003Bug36684() throws CacheException, InterruptedException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = host.getVM(2);
+    VM vm3 = host.getVM(3);
+
+    // Create the cache servers with distributed, mirrored region
+    SerializableRunnable createServer =
+      new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          CacheLoader cl = new CacheLoader() {
+            public Object load(LoaderHelper helper) {
+              return helper.getKey();
+            }
+            public void close() {
+
+            }
+          };
+          AttributesFactory factory = getBridgeServerMirroredAckRegionAttributes(cl, null);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      };
+    getSystem().getLogWriter().info("before create server");
+    vm0.invoke(createServer);
+    vm1.invoke(createServer);
+
+    // Create cache server clients
+    final int numberOfKeys = 1000;
+    final String host0 = NetworkUtils.getServerHostName(host);
+    final int vm0Port = vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final int vm1Port = vm1.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    SerializableRunnable createClient =
+      new CacheSerializableRunnable("Create Cache Server Client") {
+        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();
+          AttributesFactory factory = new AttributesFactory();
+          factory.setScope(Scope.LOCAL);
+          factory.setConcurrencyChecksEnabled(false); // test validation expects this behavior
+          // create bridge writer
+          ClientServerTestCase.configureConnectionPool(factory,host0,vm0Port,vm1Port,true,-1,-1, null);
+         createRegion(name, factory.create());
+        }
+      };
+    getSystem().getLogWriter().info("before create client");
+    vm2.invoke(createClient);
+    vm3.invoke(createClient);
+
+    // Initialize each client with entries (so that afterInvalidate is called)
+    SerializableRunnable initializeClient =
+      new CacheSerializableRunnable("Initialize Client") {
+      public void run2() throws CacheException {
+//        StringBuffer errors = new StringBuffer();
+          numberOfAfterInvalidates = 0;
+          numberOfAfterCreates = 0;
+          numberOfAfterUpdates = 0;
+          LocalRegion region = (LocalRegion) getRootRegion().getSubregion(name);
+          for (int i=0; i<numberOfKeys; i++) {
+            String expected = "key-"+i;
+            String actual = (String) region.get("key-"+i);
+            assertEquals(expected, actual);
+          }
+      }
+    };
+    
+    getSystem().getLogWriter().info("before initialize client");
+    AsyncInvocation inv2 = vm2.invokeAsync(initializeClient);
+    AsyncInvocation inv3 = vm3.invokeAsync(initializeClient);
+    
+    ThreadUtils.join(inv2, 30 * 1000);
+    ThreadUtils.join(inv3, 30 * 1000);
+    
+    if (inv2.exceptionOccurred()) { 
+      com.gemstone.gemfire.test.dunit.Assert.fail("Error occured in vm2", inv2.getException());
+    }
+    if(inv3.exceptionOccurred()) {
+      com.gemstone.gemfire.test.dunit.Assert.fail("Error occured in vm3", inv3.getException());
+    }
+  }
+  
+
+  /**
+   * Test for client connection loss with CacheLoader Exception on the server.
+   */
+  public void test004ForCacheLoaderException() throws CacheException, InterruptedException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM server = host.getVM(0);
+    VM client = host.getVM(1);
+
+    // Create the cache servers with distributed, mirrored region
+    SerializableRunnable createServer =
+      new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          CacheLoader cl = new CacheLoader() {
+            public Object load(LoaderHelper helper) {
+              System.out.println("### CALLING CACHE LOADER....");
+              throw new CacheLoaderException("Test for CahceLoaderException causing Client connection to disconnect.");
+            }
+            public void close() {
+            }
+          };
+          AttributesFactory factory = getBridgeServerMirroredAckRegionAttributes(cl, null);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      };
+    getSystem().getLogWriter().info("before create server");
+
+    server.invoke(createServer);
+
+    // Create cache server clients
+    final int numberOfKeys = 10;
+    final String host0 = NetworkUtils.getServerHostName(host);
+    final int[] port = new int[] {server.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort())};
+    final String poolName = "myPool";
+
+    SerializableRunnable createClient =
+      new CacheSerializableRunnable("Create Cache Server Client") {
+        public void run2() throws CacheException {
+          getLonerSystem();
+          AttributesFactory factory = new AttributesFactory();
+          factory.setScope(Scope.LOCAL);
+          factory.setConcurrencyChecksEnabled(false);
+          // create bridge writer
+          ClientServerTestCase.configureConnectionPoolWithName(factory,host0,port,true,-1, -1, null, poolName);
+         createRegion(name, factory.create());
+        }
+      };
+    getSystem().getLogWriter().info("before create client");
+    client.invoke(createClient);
+
+    // Initialize each client with entries (so that afterInvalidate is called)
+    SerializableRunnable invokeServerCacheLaoder =
+      new CacheSerializableRunnable("Initialize Client") {
+      public void run2() throws CacheException {
+          LocalRegion region = (LocalRegion) getRootRegion().getSubregion(name);
+          PoolStats stats = ((PoolImpl)PoolManager.find(poolName)).getStats();
+          int oldConnects = stats.getConnects();
+          int oldDisConnects = stats.getDisConnects();
+          try {
+          for (int i=0; i<numberOfKeys; i++) {
+            String actual = (String) region.get("key-"+i);
+          }
+          } catch (Exception ex){
+            if (!(ex.getCause() instanceof CacheLoaderException)) {
+              fail ("UnExpected Exception, expected to receive CacheLoaderException from server, instead found: " + ex.getCause().getClass());
+            }
+          }
+          int newConnects = stats.getConnects();
+          int newDisConnects = stats.getDisConnects();
+          //System.out.println("#### new connects/disconnects :" + newConnects + ":" + newDisConnects);
+          if (newConnects != oldConnects && newDisConnects != oldDisConnects) {
+            fail ("New connection has created for Server side CacheLoaderException.");
+          }
+      }
+    };
+
+    getSystem().getLogWriter().info("before initialize client");
+    AsyncInvocation inv2 = client.invokeAsync(invokeServerCacheLaoder);
+    
+    ThreadUtils.join(inv2, 30 * 1000);
+    SerializableRunnable stopServer = new SerializableRunnable("stop CacheServer") {
+      public void run() {
+        stopBridgeServer(getCache());
+      }
+    };
+    server.invoke(stopServer);
+    
+  }
+
+
+  protected void validateDS() {
+    List l = InternalDistributedSystem.getExistingSystems();
+    if (l.size() > 1) {
+      getSystem().getLogWriter().info("validateDS: size="
+                                      + l.size()
+                                      + " isDedicatedAdminVM="
+                                      + DistributionManager.isDedicatedAdminVM
+                                      + " l=" + l);
+    }
+    assertFalse(DistributionManager.isDedicatedAdminVM);
+    assertEquals(1, l.size());
+  }
+  
+
+  /**
+   * Tests the basic operations of the {@link Pool}
+   *
+   * @since 3.5
+   */
+  public void test006Pool() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = Host.getHost(0).getVM(2);
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          AttributesFactory factory = new AttributesFactory();
+          factory.setScope(Scope.DISTRIBUTED_ACK);
+          factory.setConcurrencyChecksEnabled(false);
+          factory.setCacheLoader(new CacheLoader() {
+              public Object load(LoaderHelper helper) {
+                //System.err.println("CacheServer data loader called");
+                return helper.getKey().toString();
+              }
+              public void close() {
+
+              }
+            });
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      });
+    final int port =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+    SerializableRunnable create =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            validateDS();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,true,-1,-1, null);
+            createRegion(name, factory.create());
+          }
+        };
+    vm1.invoke(create);
+
+    vm1.invoke(new CacheSerializableRunnable("Get values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get(new Integer(i));
+            assertEquals(String.valueOf(i), value);
+          }
+        }
+      });
+
+    vm1.invoke(new CacheSerializableRunnable("Update values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+
+          for (int i = 0; i < 10; i++) {
+            region.put(new Integer(i), new Integer(i));
+          }
+        }
+      });
+
+    vm2.invoke(create);
+    vm2.invoke(new CacheSerializableRunnable("Validate values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get(new Integer(i));
+            assertNotNull(value);
+            assertTrue(value instanceof Integer);
+            assertEquals(i, ((Integer) value).intValue());
+          }
+        }
+      });
+
+    vm1.invoke(new CacheSerializableRunnable("Close Pool") {
+        // do some special close validation here
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          String pName = region.getAttributes().getPoolName();
+          PoolImpl p = (PoolImpl)PoolManager.find(pName);
+          assertEquals(false, p.isDestroyed());
+          assertEquals(1, p.getAttachCount());
+          try {
+            p.destroy();
+            fail("expected IllegalStateException");
+          } catch (IllegalStateException expected) {
+          }
+          region.localDestroyRegion();
+          assertEquals(false, p.isDestroyed());
+          assertEquals(0, p.getAttachCount());
+        }
+      });
+
+    vm0.invoke(new SerializableRunnable("Stop CacheServer") {
+        public void run() {
+          stopBridgeServer(getCache());
+        }
+    });
+  }
+  
+  
+    
+  
+  /**
+   * Tests the BridgeServer failover (bug 31832).
+   */
+  public void test007BridgeServerFailoverCnx1() throws CacheException {
+    disconnectAllFromDS();
+    basicTestBridgeServerFailover(1);
+  }
+  /**
+   * Test BridgeServer failover with connectionsPerServer set to 0
+   */
+  public void test008BridgeServerFailoverCnx0() throws CacheException {
+    basicTestBridgeServerFailover(0);
+  }
+  private void basicTestBridgeServerFailover(final int cnxCount) throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = host.getVM(2);
+
+    // Create two bridge servers
+    SerializableRunnable createCacheServer =
+      new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          AttributesFactory factory = getBridgeServerRegionAttributes(null, null);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+        }
+
+        }
+      };
+
+    vm0.invoke(createCacheServer);
+    vm1.invoke(createCacheServer);
+
+    final int port0 =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+
+    final int port1 =
+      vm1.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+//    final String host1 = getServerHostName(vm1.getHost());
+
+    // Create one bridge client in this VM
+    SerializableRunnable create =
+      new CacheSerializableRunnable("Create region") {
+        public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port0,port1,true,-1,cnxCount, null, 100);
+
+            Region region = createRegion(name, factory.create());
+
+            // force connections to form
+            region.put("keyInit", new Integer(0));
+            region.put("keyInit2", new Integer(0));
+        }
+        };
+
+    vm2.invoke(create);
+
+    // Launch async thread that puts objects into cache. This thread will execute until
+    // the test has ended (which is why the RegionDestroyedException and CacheClosedException
+    // are caught and ignored. If any other exception occurs, the test will fail. See
+    // the putAI.exceptionOccurred() assertion below.
+    AsyncInvocation putAI = vm2.invokeAsync(new CacheSerializableRunnable("Put objects") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          try {
+            for (int i=0; i<100000; i++) {
+              region.put("keyAI", new Integer(i));
+              try {Thread.sleep(100);} catch (InterruptedException ie) {
+                fail("interrupted");
+              }
+            }
+          } catch (NoAvailableServersException ignore) {
+            /*ignore*/
+          } catch (RegionDestroyedException e) { //will be thrown when the test ends
+            /*ignore*/
+          } 
+          catch (CancelException e) { //will be thrown when the test ends
+            /*ignore*/
+          }
+        }
+      });
+
+
+    SerializableRunnable verify1Server =
+      new CacheSerializableRunnable("verify1Server") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          PoolImpl pool = getPool(region);
+          verifyServerCount(pool, 1);
+        }
+        };
+    SerializableRunnable verify2Servers =
+      new CacheSerializableRunnable("verify2Servers") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          PoolImpl pool = getPool(region);
+          verifyServerCount(pool, 2);
+        }
+      };
+
+    vm2.invoke(verify2Servers);
+
+    SerializableRunnable stopCacheServer =
+      new SerializableRunnable("Stop CacheServer") {
+          public void run() {
+            stopBridgeServer(getCache());
+          }
+      };
+
+    final String expected = "java.io.IOException";
+    final String addExpected =
+      "<ExpectedException action=add>" + expected + "</ExpectedException>";
+    final String removeExpected =
+      "<ExpectedException action=remove>" + expected + "</ExpectedException>";
+
+    vm2.invoke(new SerializableRunnable() {
+      public void run() {
+        LogWriter bgexecLogger =
+              new LocalLogWriter(InternalLogWriter.ALL_LEVEL, 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(stopCacheServer);
+
+    vm2.invoke(verify1Server);
+
+    final int restartPort = port1;
+    vm1.invoke(
+      new SerializableRunnable("Restart CacheServer") {
+      public void run() {
+        try {
+          Region region = getRootRegion().getSubregion(name);
+          assertNotNull(region);
+          startBridgeServer(restartPort);
+        }
+        catch(Exception e) {
+          getSystem().getLogWriter().fine(new Exception(e));
+          com.gemstone.gemfire.test.dunit.Assert.fail("Failed to start CacheServer", e);
+        }
+      }
+    });
+
+    // Pause long enough for the monitor to realize the server has been bounced
+    // and reconnect to it.
+    vm2.invoke(verify2Servers);
+
+    } finally {
+      vm2.invoke(new SerializableRunnable() {
+        public void run() {
+          LogWriter bgexecLogger =
+                new LocalLogWriter(InternalLogWriter.ALL_LEVEL, System.out);
+          bgexecLogger.info(removeExpected);
+        }
+      });
+    }
+
+    // Stop the other cache server
+    vm0.invoke(stopCacheServer);
+
+    // Run awhile
+    vm2.invoke(verify1Server);
+
+    com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter().info("FIXME: this thread does not terminate"); // FIXME
+//    // Verify that no exception has occurred in the putter thread
+//    DistributedTestCase.join(putAI, 5 * 60 * 1000, getLogWriter());
+//    //assertTrue("Exception occurred while invoking " + putAI, !putAI.exceptionOccurred());
+//    if (putAI.exceptionOccurred()) {
+//      fail("While putting entries: ", putAI.getException());
+//    }
+
+    // Close Pool
+    vm2.invoke(new CacheSerializableRunnable("Close Pool") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          region.localDestroyRegion();
+        }
+      });
+
+    // Stop the last cache server
+    vm1.invoke(stopCacheServer);
+  }
+  
+
+  /**
+   * Make sure cnx lifetime expiration working on thread local cnxs.
+   * @author darrel
+   */
+  public void test009LifetimeExpireOnTL() throws CacheException {
+    basicTestLifetimeExpire(true);
+  }
+
+  /**
+   * Make sure cnx lifetime expiration working on thread local cnxs.
+   * @author darrel
+   */
+  public void test010LifetimeExpireOnPoolCnx() throws CacheException {
+    basicTestLifetimeExpire(false);
+  }
+
+  protected static volatile boolean stopTestLifetimeExpire = false;
+
+  protected static volatile int baselineLifetimeCheck;
+  protected static volatile int baselineLifetimeExtensions;
+  protected static volatile int baselineLifetimeConnect;
+  protected static volatile int baselineLifetimeDisconnect;
+
+  private void basicTestLifetimeExpire(final boolean threadLocal) throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = host.getVM(2);
+
+    AsyncInvocation putAI = null;
+    AsyncInvocation putAI2 = null;
+
+    try {
+
+      // Create two bridge servers
+      SerializableRunnable createCacheServer =
+        new CacheSerializableRunnable("Create Cache Server") {
+          public void run2() throws CacheException {
+            AttributesFactory factory = getBridgeServerRegionAttributes(null, null);
+            factory.setCacheListener(new DelayListener(25));
+            createRegion(name, factory.create());
+            try {
+              startBridgeServer(0);
+
+            } catch (Exception ex) {
+              com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+            }
+
+          }
+        };
+
+      vm0.invoke(createCacheServer);
+
+      final int port0 =
+        vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+      final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+      vm1.invoke(createCacheServer);
+      final int port1 =
+        vm1.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+      SerializableRunnable stopCacheServer =
+        new SerializableRunnable("Stop CacheServer") {
+          public void run() {
+            stopBridgeServer(getCache());
+          }
+        };
+      // we only had to stop it to reserve a port
+      vm1.invoke(stopCacheServer);
+
+
+      // Create one bridge client in this VM
+      SerializableRunnable create =
+        new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port0,port1,false/*queue*/,-1,0, null, 100, 500, threadLocal, 500);
+
+            Region region = createRegion(name, factory.create());
+
+            // force connections to form
+            region.put("keyInit", new Integer(0));
+            region.put("keyInit2", new Integer(0));
+          }
+        };
+
+      vm2.invoke(create);
+
+      // Launch async thread that puts objects into cache. This thread will execute until
+      // the test has ended.
+      SerializableRunnable putter1 = 
+        new CacheSerializableRunnable("Put objects") {
+          public void run2() throws CacheException {
+            Region 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", new Integer(count));
+              }
+            } catch (NoAvailableServersException ex) {
+              if (stopTestLifetimeExpire) {
+                return;
+              } else {
+                throw ex;
+              }
+              //           } catch (RegionDestroyedException e) { //will be thrown when the test ends
+              //             /*ignore*/
+              //           } catch (CancelException e) { //will be thrown when the test ends
+              //             /*ignore*/
+            }
+          }
+        };
+      SerializableRunnable putter2 = 
+        new CacheSerializableRunnable("Put objects") {
+          public void run2() throws CacheException {
+            Region region = getRootRegion().getSubregion(name);
+            try {
+              int count = 0;
+              while (!stopTestLifetimeExpire) {
+                count++;
+                region.put("keyAI2", new Integer(count));
+              }
+            } catch (NoAvailableServersException ex) {
+              if (stopTestLifetimeExpire) {
+                return;
+              } else {
+                throw ex;
+              }
+              //           } catch (RegionDestroyedException e) { //will be thrown when the test ends
+              //             /*ignore*/
+              //           } catch (CancelException e) { //will be thrown when the test ends
+              //             /*ignore*/
+            }
+          }
+        };
+      putAI = vm2.invokeAsync(putter1);
+      putAI2 = vm2.invokeAsync(putter2);
+
+      SerializableRunnable verify1Server =
+        new CacheSerializableRunnable("verify1Server") {
+          public void run2() throws CacheException {
+            Region region = getRootRegion().getSubregion(name);
+            PoolImpl pool = getPool(region);
+            final PoolStats stats = pool.getStats();
+            verifyServerCount(pool, 1);
+            WaitCriterion ev = new WaitCriterion() {
+              public boolean done() {
+                return stats.getLoadConditioningCheck() >= (10 + baselineLifetimeCheck);
+              }
+              public String description() {
+                return null;
+              }
+            };
+            Wait.waitForCriterion(ev, 30 * 1000, 200, true);
+            
+            // 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
+            assertTrue("baselineLifetimeCheck=" + baselineLifetimeCheck
+                       + " but stats.getLoadConditioningCheck()=" + stats.getLoadConditioningCheck(),
+                       stats.getLoadConditioningCheck() >= (10+baselineLifetimeCheck));
+            baselineLifetimeCheck = stats.getLoadConditioningCheck();
+            assertTrue(stats.getLoadConditioningExtensions() > baselineLifetimeExtensions);
+            assertTrue(stats.getLoadConditioningConnect() == baselineLifetimeConnect);
+            assertTrue(stats.getLoadConditioningDisconnect() == baselineLifetimeDisconnect);
+          }
+        };
+      SerializableRunnable verify2Servers =
+        new CacheSerializableRunnable("verify2Servers") {
+          public void run2() throws CacheException {
+            Region region = getRootRegion().getSubregion(name);
+            PoolImpl pool = getPool(region);
+            final PoolStats stats = pool.getStats();
+            verifyServerCount(pool, 2);
+            // make sure some replacements are happening.
+            // since we have 2 threads and 2 cnxs and 2 servers
+            // when lifetimes are up we should connect to the other server sometimes.
+//            int retry = 300;
+//            while ((retry-- > 0)
+//                   && (stats.getLoadConditioningCheck() < (10+baselineLifetimeCheck))) {
+//              pause(100);
+//            }
+//            assertTrue("Bug 39209 expected "
+//                       + stats.getLoadConditioningCheck()
+//                       + " to be >= "
+//                       + (10+baselineLifetimeCheck),
+//                       stats.getLoadConditioningCheck() >= (10+baselineLifetimeCheck));
+            
+            // TODO: does this WaitCriterion actually help?
+            WaitCriterion wc = new WaitCriterion() {
+              String excuse;
+              public boolean done() {
+                int actual = stats.getLoadConditioningCheck();
+                int expected = 10 + baselineLifetimeCheck;
+                if (actual >= expected) {
+                  return true;
+                }
+                excuse = "Bug 39209 expected " + actual + " to be >= " + expected;
+                return false;
+              }
+              public String description() {
+                return excuse;
+              }
+            };
+            try {
+              Wait.waitForCriterion(wc, 60 * 1000, 1000, true);
+            } catch (AssertionFailedError e) {
+//              dumpStack();
+              throw e;
+            }
+            
+            assertTrue(stats.getLoadConditioningConnect() > baselineLifetimeConnect);
+            assertTrue(stats.getLoadConditioningDisconnect() > baselineLifetimeDisconnect);
+          }
+        };
+
+      vm2.invoke(verify1Server);
+      assertEquals(true, putAI.isAlive());
+      assertEquals(true, putAI2.isAlive());
+
+      {
+        final int restartPort = port1;
+        vm1.invoke(new SerializableRunnable("Restart CacheServer") {
+            public void run() {
+              try {
+                Region region = getRootRegion().getSubregion(name);
+                assertNotNull(region);
+                startBridgeServer(restartPort);
+              }
+              catch(Exception e) {
+                getSystem().getLogWriter().fine(new Exception(e));
+                com.gemstone.gemfire.test.dunit.Assert.fail("Failed to start CacheServer", e);
+              }
+            }
+          });
+      }
+
+      vm2.invoke(verify2Servers);
+      assertEquals(true, putAI.isAlive());
+      assertEquals(true, putAI2.isAlive());
+    } finally {
+      vm2.invoke(new SerializableRunnable("Stop Putters") {
+          public void run() {
+            stopTestLifetimeExpire = true;
+          }
+        });
+
+      try {
+        if (putAI != null) {
+          // Verify that no exception has occurred in the putter thread
+          ThreadUtils.join(putAI, 30 * 1000);
+          if (putAI.exceptionOccurred()) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While putting entries: ", putAI.getException());
+          }
+        }
+        
+        if (putAI2 != null) {
+          // Verify that no exception has occurred in the putter thread
+          ThreadUtils.join(putAI, 30 * 1000);
+          // FIXME this thread does not terminate
+//          if (putAI2.exceptionOccurred()) {
+//            fail("While putting entries: ", putAI.getException());
+//          }
+        }
+
+      } finally {
+        vm2.invoke(new SerializableRunnable("Stop Putters") {
+            public void run() {
+              stopTestLifetimeExpire = false;
+            }
+          });
+        // Close Pool
+        vm2.invoke(new CacheSerializableRunnable("Close Pool") {
+            public void run2() throws CacheException {
+              Region region = getRootRegion().getSubregion(name);
+              String poolName = region.getAttributes().getPoolName();
+              region.localDestroyRegion();
+              PoolManager.find(poolName).destroy();
+            }
+          });
+
+        SerializableRunnable stopCacheServer =
+          new SerializableRunnable("Stop CacheServer") {
+            public void run() {
+              stopBridgeServer(getCache());
+            }
+          };
+        vm1.invoke(stopCacheServer);
+        vm0.invoke(stopCacheServer);
+      }
+    }
+  }
+
+  /**
+   * Tests the create operation of the {@link Pool}
+   *
+   * @since 3.5
+   */
+  public void test011PoolCreate() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = Host.getHost(0).getVM(2);
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          AttributesFactory factory = getBridgeServerRegionAttributes(null, null);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      });
+    final int port =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+    SerializableRunnable create =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,false,-1,-1, null);
+           createRegion(name, factory.create());
+          }
+        };
+
+    vm1.invoke(create);
+    vm1.invoke(new CacheSerializableRunnable("Create values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            region.create(new Integer(i), new Integer(i));
+          }
+        }
+      });
+
+    vm2.invoke(create);
+    vm2.invoke(new CacheSerializableRunnable("Validate values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get(new Integer(i));
+            assertNotNull(value);
+            assertTrue(value instanceof Integer);
+            assertEquals(i, ((Integer) value).intValue());
+          }
+        }
+      });
+
+  SerializableRunnable close =
+    new CacheSerializableRunnable("Close Pool") {
+      public void run2() throws CacheException {
+        Region region = getRootRegion().getSubregion(name);
+        region.localDestroyRegion();
+      }
+    };
+
+    vm1.invoke(close);
+    vm2.invoke(close);
+
+    vm0.invoke(new SerializableRunnable("Stop CacheServer") {
+        public void run() {
+          stopBridgeServer(getCache());
+        }
+    });
+  }
+
+  /**
+   * Tests the put operation of the {@link Pool}
+   *
+   * @since 3.5
+   */
+  public void test012PoolPut() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = Host.getHost(0).getVM(2);
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          AttributesFactory factory = getBridgeServerRegionAttributes(null, null);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      });
+
+    final int port =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+    SerializableRunnable createPool =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            // create bridge writer
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,false,-1,-1, null);
+           createRegion(name, factory.create());
+          }
+        };
+
+    vm1.invoke(createPool);
+
+    vm1.invoke(new CacheSerializableRunnable("Put values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            // put string values
+            region.put("key-string-"+i, "value-"+i);
+
+            // put object values
+            Order order = new Order();
+            order.init(i);
+            region.put("key-object-"+i, order);
+
+            // put byte[] values
+            region.put("key-bytes-"+i, ("value-"+i).getBytes());
+          }
+        }
+      });
+
+    vm2.invoke(createPool);
+
+    vm2.invoke(new CacheSerializableRunnable("Get / validate string values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get("key-string-"+i);
+            assertNotNull(value);
+            assertTrue(value instanceof String);
+            assertEquals("value-"+i, value);
+          }
+        }
+      });
+
+    vm2.invoke(new CacheSerializableRunnable("Get / validate object values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get("key-object-"+i);
+            assertNotNull(value);
+            assertTrue(value instanceof Order);
+            assertEquals(i, ((Order) value).getIndex());
+          }
+        }
+      });
+
+    vm2.invoke(new CacheSerializableRunnable("Get / validate byte[] values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get("key-bytes-"+i);
+            assertNotNull(value);
+            assertTrue(value instanceof byte[]);
+            assertEquals("value-"+i, new String((byte[]) value));
+          }
+        }
+      });
+
+  SerializableRunnable closePool =
+    new CacheSerializableRunnable("Close Pool") {
+      public void run2() throws CacheException {
+        Region region = getRootRegion().getSubregion(name);
+        region.localDestroyRegion();
+      }
+    };
+
+    vm1.invoke(closePool);
+    vm2.invoke(closePool);
+
+    vm0.invoke(new SerializableRunnable("Stop CacheServer") {
+        public void run() {
+          stopBridgeServer(getCache());
+        }
+    });
+  }
+    /**
+   * Tests the put operation of the {@link Pool}
+   *
+   * @since 3.5
+   */
+  public void test013PoolPutNoDeserialize() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = Host.getHost(0).getVM(2);
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          AttributesFactory factory = getBridgeServerRegionAttributes(null,null);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      });
+
+    final int port =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+
+    SerializableRunnable createPool =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,false,-1,-1, null);
+           createRegion(name, factory.create());
+          }
+        };
+
+    vm1.invoke(createPool);
+
+    vm1.invoke(new CacheSerializableRunnable("Put values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            // put string values
+            region.put("key-string-"+i, "value-"+i);
+
+            // put object values
+            Order order = new Order();
+            order.init(i);
+            region.put("key-object-"+i, order);
+
+            // put byte[] values
+            region.put("key-bytes-"+i, ("value-"+i).getBytes());
+          }
+        }
+      });
+
+    vm2.invoke(createPool);
+
+    vm2.invoke(new CacheSerializableRunnable("Get / validate string values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get("key-string-"+i);
+            assertNotNull(value);
+            assertTrue(value instanceof String);
+            assertEquals("value-"+i, value);
+          }
+        }
+      });
+
+    vm2.invoke(new CacheSerializableRunnable("Get / validate object values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get("key-object-"+i);
+            assertNotNull(value);
+            assertTrue(value instanceof Order);
+            assertEquals(i, ((Order) value).getIndex());
+          }
+        }
+      });
+
+    vm2.invoke(new CacheSerializableRunnable("Get / validate byte[] values") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object value = region.get("key-bytes-"+i);
+            assertNotNull(value);
+            assertTrue(value instanceof byte[]);
+            assertEquals("value-"+i, new String((byte[]) value));
+          }
+        }
+      });
+
+  SerializableRunnable closePool =
+    new CacheSerializableRunnable("Close Pool") {
+      public void run2() throws CacheException {
+        Region region = getRootRegion().getSubregion(name);
+        region.localDestroyRegion();
+      }
+    };
+
+    vm1.invoke(closePool);
+    vm2.invoke(closePool);
+
+    vm0.invoke(new SerializableRunnable("Stop CacheServer") {
+        public void run() {
+          stopBridgeServer(getCache());
+        }
+    });
+    Wait.pause(5 * 1000);
+  }
+
+  /**
+   * Tests that invalidates and destroys are propagated to {@link Pool}s.
+   *
+   * @since 3.5
+   */
+  public void test014InvalidateAndDestroyPropagation() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = host.getVM(2);
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          AttributesFactory factory = getBridgeServerRegionAttributes(null, null);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+
+      });
+    final int port =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+
+    SerializableRunnable create =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,true,-1,-1, null);
+            CertifiableTestCacheListener l = new CertifiableTestCacheListener(com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter());
+            factory.setCacheListener(l);
+            Region rgn = createRegion(name, factory.create());
+            rgn.registerInterestRegex(".*", false, false);
+          }
+        };
+
+    vm1.invoke(create);
+    vm1.invoke(new CacheSerializableRunnable("Populate region") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            region.put(new Integer(i), "old" + i);
+          }
+        }
+      });
+    vm2.invoke(create);
+    Wait.pause(5 * 1000);
+    
+    vm1.invoke(new CacheSerializableRunnable("Turn on history") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          ctl.enableEventHistory();
+        }
+      });
+    vm2.invoke(new CacheSerializableRunnable("Update region") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            region.put(new Integer(i), "new" + i, "callbackArg" + i);
+          }
+        }
+      });
+    Wait.pause(5 * 1000);
+
+    vm1.invoke(new CacheSerializableRunnable("Verify invalidates") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            ctl.waitForInvalidated(key);
+            Region.Entry entry = region.getEntry(key);
+            assertNotNull(entry);
+            assertNull(entry.getValue());
+          }
+          {
+            List l = ctl.getEventHistory();
+            assertEquals(10, l.size());
+            for (int i = 0; i < 10; i++) {
+              Object key = new Integer(i);
+              EntryEvent ee = (EntryEvent)l.get(i);
+              assertEquals(key, ee.getKey());
+              assertEquals("old" + i, ee.getOldValue());
+              assertEquals(Operation.INVALIDATE, ee.getOperation());
+              assertEquals("callbackArg" + i, ee.getCallbackArgument());
+              assertEquals(true, ee.isOriginRemote());
+            }
+          }
+        }
+      });
+
+
+    vm2.invoke(new CacheSerializableRunnable("Validate original and destroy") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            assertEquals("new" + i, region.getEntry(key).getValue());
+            region.destroy(key, "destroyCB"+i);
+          }
+        }
+      });
+    Wait.pause(5 * 1000);
+
+    vm1.invoke(new CacheSerializableRunnable("Verify destroys") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            ctl.waitForDestroyed(key);
+            Region.Entry entry = region.getEntry(key);
+            assertNull(entry);
+          }
+          {
+            List l = ctl.getEventHistory();
+            assertEquals(10, l.size());
+            for (int i = 0; i < 10; i++) {
+              Object key = new Integer(i);
+              EntryEvent ee = (EntryEvent)l.get(i);
+              assertEquals(key, ee.getKey());
+              assertEquals(null, ee.getOldValue());
+              assertEquals(Operation.DESTROY, ee.getOperation());
+              assertEquals("destroyCB"+i, ee.getCallbackArgument());
+              assertEquals(true, ee.isOriginRemote());
+            }
+          }
+        }
+      });
+    vm2.invoke(new CacheSerializableRunnable("recreate") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            region.create(key, "create" + i);
+          }
+        }
+      });
+    Wait.pause(5 * 1000);
+    
+    vm1.invoke(new CacheSerializableRunnable("Verify creates") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          List l = ctl.getEventHistory();
+          com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter().info("history (should be empty): " + l);
+          assertEquals(0, l.size());
+          // now see if we can get it from the server
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            assertEquals("create"+i, region.get(key, "loadCB"+i));
+          }
+          l = ctl.getEventHistory();
+          assertEquals(10, l.size());
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            EntryEvent ee = (EntryEvent)l.get(i);
+            com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter().info("processing " + ee);
+            assertEquals(key, ee.getKey());
+            assertEquals(null, ee.getOldValue());
+            assertEquals("create"+i, ee.getNewValue());
+            assertEquals(Operation.LOCAL_LOAD_CREATE, ee.getOperation());
+            assertEquals("loadCB"+i, ee.getCallbackArgument());
+            assertEquals(false, ee.isOriginRemote());
+          }
+        }
+      });
+
+  SerializableRunnable close =
+    new CacheSerializableRunnable("Close Pool") {
+      public void run2() throws CacheException {
+        Region region = getRootRegion().getSubregion(name);
+        region.localDestroyRegion();
+      }
+    };
+
+    vm1.invoke(close);
+    vm2.invoke(close);
+
+    vm0.invoke(new SerializableRunnable("Stop CacheServer") {
+        public void run() {
+          stopBridgeServer(getCache());
+        }
+    });
+  }
+  /**
+   * Tests that invalidates and destroys are propagated to {@link Pool}s
+   * correctly to DataPolicy.EMPTY + InterestPolicy.ALL
+   *
+   * @since 5.0
+   */
+  public void test015InvalidateAndDestroyToEmptyAllPropagation() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = host.getVM(2);
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          AttributesFactory factory = getBridgeServerRegionAttributes(null, null);
+          createRegion(name, factory.create());
+          //pause(1000);
+          try {
+            startBridgeServer(0);
+
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      });
+    final int port =
+       vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+
+    SerializableRunnable createEmpty =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,true,-1,-1, null);
+            CertifiableTestCacheListener l = new CertifiableTestCacheListener(com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter());
+            factory.setCacheListener(l);
+            factory.setDataPolicy(DataPolicy.EMPTY);
+            factory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.ALL));
+            Region rgn = createRegion(name, factory.create());
+            rgn.registerInterestRegex(".*", false, false);
+          }
+        };
+    SerializableRunnable createNormal =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,true,-1,-1, null);
+            CertifiableTestCacheListener l = new CertifiableTestCacheListener(com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter());
+            factory.setCacheListener(l);
+            Region rgn = createRegion(name, factory.create());
+            rgn.registerInterestRegex(".*", false, false);
+          }
+        };
+
+    vm1.invoke(createEmpty);
+    vm1.invoke(new CacheSerializableRunnable("Populate region") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            region.put(new Integer(i), "old" + i);
+          }
+        }
+      });
+
+    vm2.invoke(createNormal);
+    vm1.invoke(new CacheSerializableRunnable("Turn on history") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          ctl.enableEventHistory();
+        }
+      });
+    vm2.invoke(new CacheSerializableRunnable("Update region") {
+      public void run2() throws CacheException {
+        Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            region.put(new Integer(i), "new" + i, "callbackArg" + i);
+      }
+        }
+    });
+    Wait.pause(5 * 1000);
+
+    vm1.invoke(new CacheSerializableRunnable("Verify invalidates") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            ctl.waitForInvalidated(key);
+            Region.Entry entry = region.getEntry(key);
+            assertNull(entry); // we are empty!
+                }
+          {
+            List l = ctl.getEventHistory();
+            assertEquals(10, l.size());
+            for (int i = 0; i < 10; i++) {
+              Object key = new Integer(i);
+              EntryEvent ee = (EntryEvent)l.get(i);
+              assertEquals(key, ee.getKey());
+              assertEquals(null, ee.getOldValue());
+              assertEquals(false, ee.isOldValueAvailable()); // failure
+              assertEquals(Operation.INVALIDATE, ee.getOperation());
+              assertEquals("callbackArg" + i, ee.getCallbackArgument());
+              assertEquals(true, ee.isOriginRemote());
+              }
+          }
+
+        }
+      });
+
+
+    vm2.invoke(new CacheSerializableRunnable("Validate original and destroy") {
+          public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            assertEquals("new" + i, region.getEntry(key).getValue());
+            region.destroy(key, "destroyCB"+i);
+          }
+        }
+      });
+    Wait.pause(5 * 1000);
+
+    vm1.invoke(new CacheSerializableRunnable("Verify destroys") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            ctl.waitForDestroyed(key);
+            Region.Entry entry = region.getEntry(key);
+            assertNull(entry);
+          }
+          {
+            List l = ctl.getEventHistory();
+            assertEquals(10, l.size());
+            for (int i = 0; i < 10; i++) {
+              Object key = new Integer(i);
+              EntryEvent ee = (EntryEvent)l.get(i);
+              assertEquals(key, ee.getKey());
+              assertEquals(null, ee.getOldValue());
+              assertEquals(false, ee.isOldValueAvailable());
+              assertEquals(Operation.DESTROY, ee.getOperation());
+              assertEquals("destroyCB"+i, ee.getCallbackArgument());
+              assertEquals(true, ee.isOriginRemote());
+            }
+          }
+        }
+      });
+    vm2.invoke(new CacheSerializableRunnable("recreate") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            region.create(key, "create" + i, "createCB"+i);
+          }
+        }
+      });
+    Wait.pause(5 * 1000);
+    
+    vm1.invoke(new CacheSerializableRunnable("Verify creates") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            ctl.waitForInvalidated(key);
+            Region.Entry entry = region.getEntry(key);
+            assertNull(entry);
+          }
+          List l = ctl.getEventHistory();
+          assertEquals(10, l.size());
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            EntryEvent ee = (EntryEvent)l.get(i);
+            assertEquals(key, ee.getKey());
+            assertEquals(null, ee.getOldValue());
+            assertEquals(false, ee.isOldValueAvailable());
+            assertEquals(Operation.INVALIDATE, ee.getOperation());
+            assertEquals("createCB"+i, ee.getCallbackArgument());
+            assertEquals(true, ee.isOriginRemote());
+          }
+          // now see if we can get it from the server
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            assertEquals("create"+i, region.get(key, "loadCB"+i));
+          }
+          l = ctl.getEventHistory();
+          assertEquals(10, l.size());
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            EntryEvent ee = (EntryEvent)l.get(i);
+            assertEquals(key, ee.getKey());
+            assertEquals(null, ee.getOldValue());
+            assertEquals("create"+i, ee.getNewValue());
+            assertEquals(Operation.LOCAL_LOAD_CREATE, ee.getOperation());
+            assertEquals("loadCB"+i, ee.getCallbackArgument());
+            assertEquals(false, ee.isOriginRemote());
+          }
+        }
+      });
+
+  SerializableRunnable close =
+    new CacheSerializableRunnable("Close Pool") {
+      public void run2() throws CacheException {
+        Region region = getRootRegion().getSubregion(name);
+        region.localDestroyRegion();
+      }
+    };
+
+    vm1.invoke(close);
+    vm2.invoke(close);
+
+    vm0.invoke(new SerializableRunnable("Stop CacheServer") {
+        public void run() {
+          stopBridgeServer(getCache());
+        }
+    });
+  }
+
+  /**
+   * Tests that invalidates and destroys are propagated to {@link Pool}s
+   * correctly to DataPolicy.EMPTY + InterestPolicy.CACHE_CONTENT
+   *
+   * @since 5.0
+   */
+  public void test016InvalidateAndDestroyToEmptyCCPropagation() throws CacheException {
+    final String name = this.getName();
+    final Host host = Host.getHost(0);
+    VM vm0 = host.getVM(0);
+    VM vm1 = host.getVM(1);
+    VM vm2 = host.getVM(2);
+
+    vm0.invoke(new CacheSerializableRunnable("Create Cache Server") {
+        public void run2() throws CacheException {
+          AttributesFactory factory = getBridgeServerRegionAttributes(null, null);
+          createRegion(name, factory.create());
+          try {
+            startBridgeServer(0);
+
+          } catch (Exception ex) {
+            com.gemstone.gemfire.test.dunit.Assert.fail("While starting CacheServer", ex);
+          }
+
+        }
+      });
+    final int port =
+      vm0.invoke(() -> ConnectionPoolDUnitTest.getCacheServerPort());
+    final String host0 = NetworkUtils.getServerHostName(vm0.getHost());
+
+    SerializableRunnable createEmpty =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,true,-1,-1, null);
+            CertifiableTestCacheListener l = new CertifiableTestCacheListener(com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter());
+            factory.setCacheListener(l);
+            factory.setDataPolicy(DataPolicy.EMPTY);
+            factory.setSubscriptionAttributes(new SubscriptionAttributes(InterestPolicy.CACHE_CONTENT));
+            Region rgn = createRegion(name, factory.create());
+            rgn.registerInterestRegex(".*", false, false);
+         }
+        };
+    SerializableRunnable createNormal =
+      new CacheSerializableRunnable("Create region") {
+          public void run2() throws CacheException {
+            getLonerSystem();
+            getCache();
+            AttributesFactory factory = new AttributesFactory();
+            factory.setScope(Scope.LOCAL);
+            factory.setConcurrencyChecksEnabled(false);
+            ClientServerTestCase.configureConnectionPool(factory,host0,port,-1,true,-1,-1, null);
+            CertifiableTestCacheListener l = new CertifiableTestCacheListener(com.gemstone.gemfire.test.dunit.LogWriterUtils.getLogWriter());
+            factory.setCacheListener(l);
+            Region rgn = createRegion(name, factory.create());
+            rgn.registerInterestRegex(".*", false, false);
+          }
+        };
+
+    vm1.invoke(createEmpty);
+    vm1.invoke(new CacheSerializableRunnable("Populate region") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            region.put(new Integer(i), "old" + i);
+          }
+        }
+      });
+
+    vm2.invoke(createNormal);
+    vm1.invoke(new CacheSerializableRunnable("Turn on history") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          ctl.enableEventHistory();
+        }
+      });
+    vm2.invoke(new CacheSerializableRunnable("Update region") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            region.put(new Integer(i), "new" + i, "callbackArg" + i);
+          }
+        }
+      });
+    Wait.pause(5 * 1000);
+
+    vm1.invoke(new CacheSerializableRunnable("Verify invalidates") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          List l = ctl.getEventHistory();
+          assertEquals(0, l.size());
+        }
+      });
+
+
+    vm2.invoke(new CacheSerializableRunnable("Validate original and destroy") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            assertEquals("new" + i, region.getEntry(key).getValue());
+            region.destroy(key, "destroyCB"+i);
+          }
+        }
+      });
+
+    vm1.invoke(new CacheSerializableRunnable("Verify destroys") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          CertifiableTestCacheListener ctl = (CertifiableTestCacheListener) region.getAttributes().getCacheListener();
+          List l = ctl.getEventHistory();
+          assertEquals(0, l.size());
+        }
+      });
+    vm2.invoke(new CacheSerializableRunnable("recreate") {
+        public void run2() throws CacheException {
+          Region region = getRootRegion().getSubregion(name);
+          for (int i = 0; i < 10; i++) {
+            Object key = new Integer(i);
+            region.create(key, "create" + i, "createCB"+i);
+          }
+        }
+

<TRUNCATED>


Mime
View raw message