geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From upthewatersp...@apache.org
Subject [41/70] [partial] incubator-geode git commit: WAN and CQ code drop under the Pivotal SGA
Date Thu, 28 Jan 2016 18:13:43 GMT
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqQueryUsingPoolOptimizedExecuteDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqQueryUsingPoolOptimizedExecuteDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqQueryUsingPoolOptimizedExecuteDUnitTest.java
new file mode 100644
index 0000000..c58e02c
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqQueryUsingPoolOptimizedExecuteDUnitTest.java
@@ -0,0 +1,42 @@
+/*=========================================================================
+ * 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.query.cq.dunit;
+
+import com.gemstone.gemfire.cache.query.internal.cq.CqServiceImpl;
+
+import dunit.SerializableRunnable;
+
+/**
+ * Test class for testing {@link CqServiceImpl#EXECUTE_QUERY_DURING_INIT} flag
+ *
+ */
+public class CqQueryUsingPoolOptimizedExecuteDUnitTest extends CqQueryUsingPoolDUnitTest{
+
+  public CqQueryUsingPoolOptimizedExecuteDUnitTest(String name) {
+    super(name);
+   }
+
+  public void setUp() throws Exception {
+    super.setUp();
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        CqServiceImpl.EXECUTE_QUERY_DURING_INIT = false;
+      }
+    });
+  }
+  
+  @Override
+  public void tearDown2() throws Exception {
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        CqServiceImpl.EXECUTE_QUERY_DURING_INIT = true;
+      }
+    });
+    super.tearDown2();
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqResultSetUsingPoolDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqResultSetUsingPoolDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqResultSetUsingPoolDUnitTest.java
new file mode 100644
index 0000000..e7f3e0e
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqResultSetUsingPoolDUnitTest.java
@@ -0,0 +1,1131 @@
+/*=========================================================================
+ * 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.query.cq.dunit;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+import com.gemstone.gemfire.cache.CacheException;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.data.Portfolio;
+import com.gemstone.gemfire.cache.query.internal.DefaultQueryService;
+import com.gemstone.gemfire.cache.query.internal.cq.InternalCqQuery;
+import com.gemstone.gemfire.cache.query.internal.cq.ServerCQImpl;
+import com.gemstone.gemfire.cache.query.internal.cq.CqService;
+import com.gemstone.gemfire.cache.query.internal.index.IndexManager;
+import com.gemstone.gemfire.cache30.CacheSerializableRunnable;
+import com.gemstone.gemfire.cache30.CacheTestCase;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+
+import dunit.Host;
+import dunit.SerializableRunnable;
+import dunit.VM;
+
+/**
+ * This class tests the ContiunousQuery mechanism in GemFire.
+ *
+ * @author anil
+ */
+public class CqResultSetUsingPoolDUnitTest extends CacheTestCase {
+
+  protected CqQueryUsingPoolDUnitTest cqDUnitTest = new CqQueryUsingPoolDUnitTest("CqResultSetUsingPoolDUnitTest");
+  
+  private final String selStr = "SELECT * FROM /root/regionA";
+  
+  /** Supported queries */
+  public final String[] condition = new String [] {
+    /* 0  */ " p WHERE p.ID > 3",
+    /* 1  */ " p WHERE p.ID < 3",
+    /* 2  */ "   WHERE   ID = 3",
+    /* 3  */ " p WHERE p.ID >= 3",
+    /* 4  */ " p WHERE p.ID <= 3",
+    /* 5  */ " p WHERE p.ID > 3 AND p.status = 'active'",
+    /* 6  */ "   WHERE   status = 'active' AND ID < 3",
+    /* 7  */ " p WHERE p.names[0] = 'aaa'",
+    /* 8  */ " p WHERE p.status LIKE 'active'",
+    /* 9  */ " p WHERE p.collectionHolderMap.get('1').arr[0] = '0'",
+    /* 10 */ " p WHERE p.position1.portfolioId > 3",
+    /* 11 */ " p where p.position3[1].portfolioId = 2",
+    /* 12 */ " p where NOT(SELECT DISTINCT * FROM positions.values pos " +
+             " WHERE pos.secId in SET('YHOO', 'SUN', 'IBM', 'YHOO', 'GOOG', " +
+             " 'MSFT', 'AOL', 'APPL', 'ORCL', 'SAP', 'DELL', 'RHAT', 'NOVL', 'HP')).isEmpty",
+    /* 13  */ " p WHERE p.ID != 3 AND p.ID !=4",
+    /* 14  */ " p WHERE (p.ID = 2 OR p.ID = 4) AND p.status = 'active'",
+
+  };
+
+  /** For intial size of 5 */
+  public final int[] resultSize = new int [] {
+    /* 0  */ 2,
+    /* 1  */ 2,
+    /* 2  */ 1,
+    /* 3  */ 3,
+    /* 4  */ 3,
+    /* 5  */ 1,
+    /* 6  */ 1,
+    /* 7  */ 5,
+    /* 8  */ 2,
+    /* 9  */ 5,
+    /* 10 */ 2,
+    /* 11 */ 5,
+    /* 12 */ 5,
+    /* 13 */ 3,
+    /* 14 */ 2,
+  };
+
+  /** For intial size of 5 */
+  public final String[][] expectedKeys = new String [][] {
+    /* 0  */ {"key-4", "key-5"},
+    /* 1  */ {"key-1", "key-2"},
+    /* 2  */ {"key-3"},
+    /* 3  */ {"key-3", "key-4", "key-5"},
+    /* 4  */ {"key-1", "key-2", "key-3"},
+    /* 5  */ {"key-4"},
+    /* 6  */ {"key-2"},
+    /* 7  */ {"key-1", "key-2", "key-3","key-4", "key-5"},
+    /* 8  */ {"key-2", "key-4"},
+    /* 9  */ {"key-1", "key-2", "key-3","key-4", "key-5"},
+    /* 10 */ {"key-4", "key-5"},
+    /* 11 */ {"key-1", "key-2", "key-3","key-4", "key-5"},
+    /* 12 */ {"key-1", "key-2", "key-3","key-4", "key-5"},
+    /* 13 */ {"key-1", "key-2", "key-5"},
+    /* 14  */ {"key-2", "key-4"},
+  };
+
+
+  
+  /** Unsupported queries */
+  public final String[] condition2 = new String [] {
+    /* 0  */ " p1, /root/regionB p2 WHERE p1.status = p2.status",
+    /* 1  */ " p, p.positions.values p1 WHERE p1.secId = 'IBM'",
+    /* 2  */ " p, p.positions.values AS pos WHERE pos.secId != '1'",
+    /* 3  */ " p WHERE p.ID in (SELECT p1.ID FROM /root/regionA p1 WHERE p1.ID > 3)",
+    /* 4  */ ".entries entry WHERE entry.key = '1'",
+    /* 5  */ ".entries entry WHERE entry.value.ID > '3'",
+    /* 6  */ ".values p WHERE p.ID > '3' and p.status = 'active'",
+    /* 7  */ " p, p.position3 pos where pos.portfolioId  = 1",
+  };
+  
+
+  public CqResultSetUsingPoolDUnitTest(String name) {
+    super(name);
+  }
+  
+  public void setUp() throws Exception {
+    super.setUp();
+    
+    // avoid IllegalStateException from HandShake by connecting all vms tor
+    // system before creating ConnectionPools
+    getSystem();
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        getSystem();
+      }
+    });
+    
+  }
+  
+  
+  /**
+   * Tests CQ Result Set.
+   * 
+   * @throws Exception
+   */
+  public void testCqResults() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server = host.getVM(0);
+    VM client = host.getVM(1);
+
+    cqDUnitTest.createServer(server);
+
+    final int port = server.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server.getHost());
+
+    String poolName = "testCqResults";
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // Put 5 entries into the region.
+    cqDUnitTest.createValues(server, "regionA", 5);
+    
+    // Test for supported queries.
+    String cqQuery = "";
+    for (int queryCnt=0; queryCnt < condition.length; queryCnt++) {
+      cqQuery = selStr + condition[queryCnt];
+      cqDUnitTest.createCQ(client, poolName, "testCqResultsP_" + queryCnt, cqQuery);
+      cqDUnitTest.executeCQ(client, "testCqResultsP_" + queryCnt, true, resultSize[queryCnt], expectedKeys[queryCnt], null);
+    }
+        
+    // Test unsupported queries.
+    for (int queryCnt=0; queryCnt < condition2.length; queryCnt++) {
+      cqQuery = selStr + condition2[queryCnt];
+      
+      try {
+        cqDUnitTest.createCQ(client, poolName, "testCqResultsF_" + queryCnt, cqQuery);
+        //cqDUnitTest.executeCQ(client, "testCqResultsF_" + queryCnt, true, cqDUnitTest.noTest, null);
+        fail("UnSupported CQ Query, Expected to fail. " +
+            " CQ :" + "testCqResultsF_" + queryCnt +
+            " Query : " + cqQuery);
+      } catch (Exception ex) {
+        // Expected.
+      }
+    }
+    
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server); 
+  }
+
+
+  /**
+   * Tests CQ Result Set with Compact Range Index.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsWithCompactRangeIndex() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server = host.getVM(0);
+    VM client = host.getVM(1);
+
+    cqDUnitTest.createServer(server);
+
+    final int port = server.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server.getHost());
+
+    String poolName = "testCqResults";
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // Create index.
+    cqDUnitTest.createFunctionalIndex(server, "IdIndex", "p.ID", "/root/regionA p");
+    cqDUnitTest.createFunctionalIndex(server, "statusIndex", "p.status", "/root/regionA p");
+    cqDUnitTest.createFunctionalIndex(server, "portfolioIdIndex", "p.position1.portfolioId", "/root/regionA p");
+    
+    // Put 5 entries into the region.
+    cqDUnitTest.createValues(server, "regionA", 5);
+    
+    // Test for supported queries.
+    String cqQuery = "";
+    for (int queryCnt=0; queryCnt < condition.length; queryCnt++) {
+      cqQuery = selStr + condition[queryCnt];
+      cqDUnitTest.createCQ(client, poolName, "testCqResultsP_" + queryCnt, cqQuery);
+      cqDUnitTest.executeCQ(client, "testCqResultsP_" + queryCnt, true, resultSize[queryCnt], expectedKeys[queryCnt], null);
+    }
+        
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server); 
+  }
+
+  /**
+   * Tests CQ Result Set with Range Index.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsWithRangeIndex() throws Exception
+  {
+    
+    final Host host = Host.getHost(0);
+    VM server = host.getVM(0);
+    VM client = host.getVM(1);
+
+    cqDUnitTest.createServer(server);
+
+    final int port = server.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server.getHost());
+
+    String poolName = "testCqResults";
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // Create index.
+    server.invoke(new CacheSerializableRunnable("Set RangeIndex Falg") {
+      public void run2() throws CacheException {
+        IndexManager.TEST_RANGEINDEX_ONLY = true;
+      }
+    });
+
+    cqDUnitTest.createFunctionalIndex(server, "IdIndex", "p.ID", "/root/regionA p");
+    cqDUnitTest.createFunctionalIndex(server, "statusIndex", "p.status", "/root/regionA p");
+    cqDUnitTest.createFunctionalIndex(server, "portfolioIdIndex", "p.position1.portfolioId", "/root/regionA p");
+    
+    // Put 5 entries into the region.
+    cqDUnitTest.createValues(server, "regionA", 5);
+    
+    // Test for supported queries.
+    String cqQuery = "";
+    for (int queryCnt=0; queryCnt < condition.length; queryCnt++) {
+      cqQuery = selStr + condition[queryCnt];
+      cqDUnitTest.createCQ(client, poolName, "testCqResultsP_" + queryCnt, cqQuery);
+      cqDUnitTest.executeCQ(client, "testCqResultsP_" + queryCnt, true, resultSize[queryCnt], expectedKeys[queryCnt], null);
+    }
+
+    // Unset the flag.
+    server.invoke(new CacheSerializableRunnable("Set RangeIndex Falg") {
+      public void run2() throws CacheException {
+        IndexManager.TEST_RANGEINDEX_ONLY = false;
+      }
+    });
+        
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server); 
+  }
+
+  /**
+   * Tests CQ Result Set.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsOnPR() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server1 = host.getVM(0);
+    VM server2 = host.getVM(1);
+    VM client = host.getVM(2);
+
+    cqDUnitTest.createServerWithPR(server1, 0, false, 0);
+    cqDUnitTest.createServerWithPR(server2, 0, false, 0);
+    
+    final int port = server1.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server1.getHost());
+
+    String poolName = "testCqResults";
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // Put 5 entries into the region.
+    cqDUnitTest.createValues(server1, "regionA", 5);
+    
+    // Test for supported queries.
+    String cqQuery = "";
+    for (int queryCnt=0; queryCnt < condition.length; queryCnt++) {
+      cqQuery = selStr + condition[queryCnt];
+      cqDUnitTest.createCQ(client, poolName, "testCqResultsP_" + queryCnt, cqQuery);
+      cqDUnitTest.executeCQ(client, "testCqResultsP_" + queryCnt, true, resultSize[queryCnt], expectedKeys[queryCnt], null);
+    }
+        
+    // Test unsupported queries.
+    for (int queryCnt=0; queryCnt < condition2.length; queryCnt++) {
+      cqQuery = selStr + condition2[queryCnt];
+      
+      try {
+        cqDUnitTest.createCQ(client, poolName, "testCqResultsF_" + queryCnt, cqQuery);
+        //cqDUnitTest.executeCQ(client, "testCqResultsF_" + queryCnt, true, cqDUnitTest.noTest, null);
+        fail("UnSupported CQ Query, Expected to fail. " +
+            " CQ :" + "testCqResultsF_" + queryCnt +
+            " Query : " + cqQuery);
+      } catch (Exception ex) {
+        // Expected.
+      }
+    }
+    
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server1);
+    cqDUnitTest.closeServer(server2);
+  }
+
+  /**
+   * Tests CQ Result Set with Compact Range Index.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsWithCompactRangeIndexOnPR() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server1 = host.getVM(0);
+    VM server2 = host.getVM(1);
+    VM client = host.getVM(2);
+
+    cqDUnitTest.createServerWithPR(server1, 0, false, 0);
+    cqDUnitTest.createServerWithPR(server2, 0, false, 0);
+
+    final int port = server1.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server1.getHost());
+
+    String poolName = "testCqResults";
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // Create index.
+    cqDUnitTest.createFunctionalIndex(server1, "IdIndex", "p.ID", "/root/regionA p");
+    cqDUnitTest.createFunctionalIndex(server1, "statusIndex", "p.status", "/root/regionA p");
+    cqDUnitTest.createFunctionalIndex(server1, "portfolioIdIndex", "p.position1.portfolioId", "/root/regionA p");
+    
+    // Put 5 entries into the region.
+    cqDUnitTest.createValues(server1, "regionA", 5);
+    
+    // Test for supported queries.
+    String cqQuery = "";
+    for (int queryCnt=0; queryCnt < condition.length; queryCnt++) {
+      cqQuery = selStr + condition[queryCnt];
+      cqDUnitTest.createCQ(client, poolName, "testCqResultsP_" + queryCnt, cqQuery);
+      cqDUnitTest.executeCQ(client, "testCqResultsP_" + queryCnt, true, resultSize[queryCnt], expectedKeys[queryCnt], null);
+    }
+        
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server1); 
+    cqDUnitTest.closeServer(server2); 
+
+  }
+
+  /**
+   * Tests CQ Result Set with Range Index.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsWithRangeIndexOnPR() throws Exception
+  {
+    
+    final Host host = Host.getHost(0);
+    VM server1 = host.getVM(0);
+    VM server2 = host.getVM(1);
+    VM client = host.getVM(2);
+
+    cqDUnitTest.createServerWithPR(server1, 0, false, 0);
+    cqDUnitTest.createServerWithPR(server2, 0, false, 0);
+
+    final int port = server1.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server1.getHost());
+
+    String poolName = "testCqResults";
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // Create index.
+    server1.invoke(new CacheSerializableRunnable("Set RangeIndex Falg") {
+      public void run2() throws CacheException {
+        IndexManager.TEST_RANGEINDEX_ONLY = true;
+      }
+    });
+
+    server2.invoke(new CacheSerializableRunnable("Set RangeIndex Falg") {
+      public void run2() throws CacheException {
+        IndexManager.TEST_RANGEINDEX_ONLY = true;
+      }
+    });
+
+    cqDUnitTest.createFunctionalIndex(server1, "IdIndex", "p.ID", "/root/regionA p");
+    cqDUnitTest.createFunctionalIndex(server1, "statusIndex", "p.status", "/root/regionA p");
+    cqDUnitTest.createFunctionalIndex(server1, "portfolioIdIndex", "p.position1.portfolioId", "/root/regionA p");
+    
+    // Put 5 entries into the region.
+    cqDUnitTest.createValues(server1, "regionA", 5);
+    
+    // Test for supported queries.
+    String cqQuery = "";
+    for (int queryCnt=0; queryCnt < condition.length; queryCnt++) {
+      cqQuery = selStr + condition[queryCnt];
+      cqDUnitTest.createCQ(client, poolName, "testCqResultsP_" + queryCnt, cqQuery);
+      cqDUnitTest.executeCQ(client, "testCqResultsP_" + queryCnt, true, resultSize[queryCnt], expectedKeys[queryCnt], null);
+    }
+        
+    // Create index.
+    server1.invoke(new CacheSerializableRunnable("Set RangeIndex Falg") {
+      public void run2() throws CacheException {
+        IndexManager.TEST_RANGEINDEX_ONLY = false;
+      }
+    });
+
+    server2.invoke(new CacheSerializableRunnable("Set RangeIndex Falg") {
+      public void run2() throws CacheException {
+        IndexManager.TEST_RANGEINDEX_ONLY = false;
+      }
+    });
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server1); 
+    cqDUnitTest.closeServer(server2); 
+  }
+
+  /**
+   * Tests CQ Result Set.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsCaching() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server = host.getVM(0);
+    VM client = host.getVM(1);
+
+    cqDUnitTest.createServer(server);
+
+    final int port = server.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server.getHost());
+
+    String poolName = "testCqResults";
+    final String cqName = "testCqResultsP_0";
+    
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // create CQ.
+    cqDUnitTest.createCQ(client, poolName, cqName, cqDUnitTest.cqs[0]);
+    
+    final int numObjects = 300;
+    final int totalObjects = 500;
+    
+    // initialize Region.
+    server.invoke(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        for (int i = 1; i <= numObjects; i++) {
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+      }
+    });
+    
+    // Keep updating region (async invocation).
+    server.invokeAsync(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        // Update (totalObjects - 1) entries.
+        for (int i = 1; i < totalObjects; i++) {
+          // Destroy entries.
+          if (i > 25 && i < 201) {
+            region.destroy(""+i);
+            continue;
+          }
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+        // recreate destroyed entries.
+        for (int j = 26; j < 201; j++) {
+          Portfolio p = new Portfolio(j);
+          region.put(""+j, p);
+        }
+        // Add the last key.
+        Portfolio p = new Portfolio(totalObjects);
+        region.put(""+totalObjects, p);
+      }
+    });
+
+    // Execute CQ.
+    // While region operation is in progress execute CQ.
+    cqDUnitTest.executeCQ(client, cqName, true, null);
+    
+    // Verify CQ Cache results.
+    server.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqService cqService = null;
+        try {
+          cqService = ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqService.", ex);
+          fail ("Failed to get the internal CqService.", ex);
+        }
+        
+        // Wait till all the region update is performed.
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        while(true){
+          if (region.get(""+ totalObjects) == null){
+            try {
+              Thread.sleep(50);
+            } catch (Exception ex){
+              //ignore.
+            }
+            continue;
+          }
+          break;
+        }
+        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          if (cqQuery.getName().equals(cqName)) {
+            int size = cqQuery.getCqResultKeysSize();
+            if (size != totalObjects) {
+              getLogWriter().info("The number of Cached events " + size + 
+                  " is not equal to the expected size " + totalObjects);
+              HashSet expectedKeys = new HashSet();
+              for (int i = 1; i < totalObjects; i++) {
+                expectedKeys.add("" + i);
+              }
+              Set cachedKeys = cqQuery.getCqResultKeyCache();
+              expectedKeys.removeAll(cachedKeys);
+              getLogWriter().info("Missing keys from the Cache : " + expectedKeys);
+            }
+            assertEquals("The number of keys cached for cq " + cqName + " is wrong.", 
+                totalObjects, cqQuery.getCqResultKeysSize());              
+          }
+        }
+      }
+    });
+    
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server); 
+  }
+
+  /**
+   * Tests CQ Result Set.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsCachingForMultipleCQs() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server = host.getVM(0);
+    VM client1 = host.getVM(1);
+    VM client2 = host.getVM(2);
+
+    cqDUnitTest.createServer(server);
+
+    final int port = server.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server.getHost());
+
+    String poolName = "testCqResults";
+    final String cqName1 = "testCqResultsP_0";
+    final String cqName2 = "testCqResultsP_1";
+    
+    cqDUnitTest.createPool(client1, poolName, host0, port);
+    cqDUnitTest.createPool(client2, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client1, port, host0);
+    cqDUnitTest.createClient(client2, port, host0);
+    
+    // create CQ.
+    cqDUnitTest.createCQ(client1, poolName, cqName1, cqDUnitTest.cqs[0]);
+    cqDUnitTest.createCQ(client2, poolName, cqName2, cqDUnitTest.cqs[0]);
+    
+    final int numObjects = 300;
+    final int totalObjects = 500;
+    
+    // initialize Region.
+    server.invoke(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        for (int i = 1; i <= numObjects; i++) {
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+      }
+    });
+    
+    // Keep updating region (async invocation).
+    server.invokeAsync(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        // Update (totalObjects - 1) entries.
+        for (int i = 1; i < totalObjects; i++) {
+          // Destroy entries.
+          if (i > 25 && i < 201) {
+            region.destroy(""+i);
+            continue;
+          }
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+        // recreate destroyed entries.
+        for (int j = 26; j < 201; j++) {
+          Portfolio p = new Portfolio(j);
+          region.put(""+j, p);
+        }
+        // Add the last key.
+        Portfolio p = new Portfolio(totalObjects);
+        region.put(""+totalObjects, p);
+      }
+    });
+
+    // Execute CQ.
+    // While region operation is in progress execute CQ.
+    cqDUnitTest.executeCQ(client1, cqName1, true, null);
+    cqDUnitTest.executeCQ(client2, cqName2, true, null);
+    
+    // Verify CQ Cache results.
+    server.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqService cqService = null;
+        try {
+          cqService = ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqService.", ex);
+          fail ("Failed to get the internal CqService.", ex);
+        }
+        
+        // Wait till all the region update is performed.
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        while(true){
+          if (region.get(""+ totalObjects) == null){
+            try {
+              Thread.sleep(50);
+            } catch (Exception ex){
+              //ignore.
+            }
+            continue;
+          }
+          break;
+        }
+        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          int size = cqQuery.getCqResultKeysSize();
+          if (size != totalObjects) {
+            getLogWriter().info("The number of Cached events " + size + 
+                " is not equal to the expected size " + totalObjects);
+            HashSet expectedKeys = new HashSet();
+            for (int i = 1; i < totalObjects; i++) {
+              expectedKeys.add("" + i);
+            }
+            Set cachedKeys = cqQuery.getCqResultKeyCache();
+            expectedKeys.removeAll(cachedKeys);
+            getLogWriter().info("Missing keys from the Cache : " + expectedKeys);
+          }
+          assertEquals("The number of keys cached for cq " + cqQuery.getName() + " is wrong.", 
+              totalObjects, cqQuery.getCqResultKeysSize());              
+        }
+      }
+    });
+    
+    // Close.
+    cqDUnitTest.closeClient(client1);
+    cqDUnitTest.closeClient(client2);
+    cqDUnitTest.closeServer(server); 
+  }
+
+  /**
+   * Tests CQ Result Set.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsCachingForPR() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server1 = host.getVM(0);
+    VM server2 = host.getVM(1);
+    VM client = host.getVM(2);
+
+    cqDUnitTest.createServerWithPR(server1, 0, false, 0);
+    cqDUnitTest.createServerWithPR(server2, 0, false, 0);
+    
+    final int port = server1.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server1.getHost());
+
+    String poolName = "testCqResults";
+    final String cqName = "testCqResultsP_0";
+    
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // create CQ.
+    cqDUnitTest.createCQ(client, poolName, cqName, cqDUnitTest.cqs[0]);
+    
+    final int numObjects = 300;
+    final int totalObjects = 500;
+    
+    // initialize Region.
+    server1.invoke(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        for (int i = 1; i <= numObjects; i++) {
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+      }
+    });
+    
+    // Keep updating region (async invocation).
+    server2.invokeAsync(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        // Update (totalObjects - 1) entries.
+        for (int i = 1; i < totalObjects; i++) {
+          // Destroy entries.
+          if (i > 25 && i < 201) {
+            region.destroy(""+i);
+            continue;
+          }
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+        // recreate destroyed entries.
+        for (int j = 26; j < 201; j++) {
+          Portfolio p = new Portfolio(j);
+          region.put(""+j, p);
+        }
+        // Add the last key.
+        Portfolio p = new Portfolio(totalObjects);
+        region.put(""+totalObjects, p);
+      }
+    });
+
+    // Execute CQ.
+    // While region operation is in progress execute CQ.
+    cqDUnitTest.executeCQ(client, cqName, true, null);
+    
+    // Verify CQ Cache results.
+    server1.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqService cqService = null;
+        try {
+          cqService = ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqService.", ex);
+          fail ("Failed to get the internal CqService.", ex);
+        }
+        
+        // Wait till all the region update is performed.
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        while(true){
+          if (region.get(""+ totalObjects) == null){
+            try {
+              Thread.sleep(50);
+            } catch (Exception ex){
+              //ignore.
+            }
+            continue;
+          }
+          break;
+        }
+        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          if (cqQuery.getCqResultKeysSize() <= 0) {
+            fail("The Result Cache for CQ on PR is not working. CQ :" + cqName);              
+          }
+        }
+      }
+    });
+    
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server1);
+    cqDUnitTest.closeServer(server2);
+  }
+
+  /**
+   * Tests CQ Result Set caching for destroy events.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsCachingForDestroyEventsOnPR() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server1 = host.getVM(0);
+    VM server2 = host.getVM(1);
+    VM client = host.getVM(2);
+
+    cqDUnitTest.createServerWithPR(server1, 0, false, 0);
+    cqDUnitTest.createServerWithPR(server2, 0, false, 0);
+    
+    final int port = server1.invokeInt(CqQueryUsingPoolDUnitTest.class,
+        "getCacheServerPort");
+    final String host0 = getServerHostName(server1.getHost());
+
+    String poolName = "testCqResults";
+    final String cqName = "testCqResultsCachingForDestroyEventsOnPR_0";
+    
+    cqDUnitTest.createPool(client, poolName, host0, port);
+    
+    // Create client.
+    cqDUnitTest.createClient(client, port, host0);
+
+    // create CQ.
+    cqDUnitTest.createCQ(client, poolName, cqName, cqDUnitTest.cqs[0]);
+
+    // Execute CQ.
+    cqDUnitTest.executeCQ(client, cqName, true, null);
+
+    final int numObjects = 50;
+    
+    // initialize Region.
+    server1.invoke(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        for (int i = 1; i <= numObjects; i++) {
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+      }
+    });
+
+    // Update from server2.
+    server2.invoke(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        for (int i = 1; i <= numObjects; i++) {
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+      }
+    });
+
+    // Destroy entries.
+    server2.invoke(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        for (int i = 1; i <= numObjects; i++) {
+          Portfolio p = new Portfolio(i);
+          region.destroy(""+i);
+        }
+      }
+    });
+    
+    // Wait for events to be sent.
+    cqDUnitTest.waitForDestroyed(client, cqName, "" + numObjects);
+    
+    // Verify CQ Cache results.
+    server1.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqService cqService = null;
+        try {
+          cqService = ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqService.", ex);
+          fail ("Failed to get the internal CqService.", ex);
+        }
+        
+        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          if (cqQuery.getCqResultKeysSize() > 0) {
+            fail("The CQ Result Cache on PR should have been empty for CQ :" + cqName + " keys=" + cqQuery.getCqResultKeyCache());              
+          }
+        }
+      }
+    });
+
+    server2.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqService cqService = null;
+        try {
+          cqService = ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqService.", ex);
+          fail ("Failed to get the internal CqService.", ex);
+        }
+        
+        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          if (cqQuery.getCqResultKeysSize() > 0) {
+            fail("The CQ Result Cache on PR should have been empty for CQ :" + cqName);              
+          }
+        }
+      }
+    });
+
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server1);
+    cqDUnitTest.closeServer(server2);
+  }
+
+
+  /**
+   * Tests CQ Result Caching with CQ Failover.
+   * 
+   * @throws Exception
+   */
+  public void testCqResultsCachingWithFailOver() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server1 = host.getVM(0);
+    VM server2 = host.getVM(1);
+    VM client = host.getVM(2);
+    
+    cqDUnitTest.createServer(server1);
+    
+    final int port1 = server1.invokeInt(CqQueryUsingPoolDUnitTest.class, "getCacheServerPort");
+    final String host0 = getServerHostName(server1.getHost());
+    final int[] ports = AvailablePortHelper.getRandomAvailableTCPPorts(1);
+    
+    String poolName = "testCQFailOver";
+    final String cqName = "testCQFailOver_0";
+    
+    cqDUnitTest.createPool(client, poolName, new String[] {host0, host0}, new int[] {port1, ports[0]});
+
+    // create CQ.
+    cqDUnitTest.createCQ(client, poolName, cqName, cqDUnitTest.cqs[0]);
+    
+    final int numObjects = 300;
+    final int totalObjects = 500;
+    
+    // initialize Region.
+    server1.invoke(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        for (int i = 1; i <= numObjects; i++) {
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+      }
+    });
+    
+    // Keep updating region (async invocation).
+    server1.invokeAsync(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        // Update (totalObjects - 1) entries.
+        for (int i = 1; i < totalObjects; i++) {
+          // Destroy entries.
+          if (i > 25 && i < 201) {
+            region.destroy(""+i);
+            continue;
+          }
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+        // recreate destroyed entries.
+        for (int j = 26; j < 201; j++) {
+          Portfolio p = new Portfolio(j);
+          region.put(""+j, p);
+        }
+        // Add the last key.
+        Portfolio p = new Portfolio(totalObjects);
+        region.put(""+totalObjects, p);
+      }
+    });
+
+    // Execute CQ.
+    // While region operation is in progress execute CQ.
+    cqDUnitTest.executeCQ(client, cqName, true, null);
+    
+    // Verify CQ Cache results.
+    server1.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqService cqService = null;
+        try {
+          cqService = ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqService.", ex);
+          fail ("Failed to get the internal CqService.", ex);
+        }
+        
+        // Wait till all the region update is performed.
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        while(true){
+          if (region.get(""+ totalObjects) == null){
+            try {
+              Thread.sleep(50);
+            } catch (Exception ex){
+              //ignore.
+            }
+            continue;
+          }
+          break;
+        }
+        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          if (cqQuery.getName().equals(cqName)) {
+            int size = cqQuery.getCqResultKeysSize();
+            if (size != totalObjects) {
+              getLogWriter().info("The number of Cached events " + size + 
+                  " is not equal to the expected size " + totalObjects);
+              HashSet expectedKeys = new HashSet();
+              for (int i = 1; i < totalObjects; i++) {
+                expectedKeys.add("" + i);
+              }
+              Set cachedKeys = cqQuery.getCqResultKeyCache();
+              expectedKeys.removeAll(cachedKeys);
+              getLogWriter().info("Missing keys from the Cache : " + expectedKeys);
+            }
+            assertEquals("The number of keys cached for cq " + cqName + " is wrong.", 
+                totalObjects, cqQuery.getCqResultKeysSize());              
+          }
+        }
+      }
+    });
+    
+    cqDUnitTest.createServer(server2, ports[0]);
+    final int thePort2 = server2.invokeInt(CqQueryUsingPoolDUnitTest.class, "getCacheServerPort");
+    System.out.println("### Port on which server1 running : " + port1 + 
+        " Server2 running : " + thePort2);
+    pause(3 * 1000);
+    
+    // Close server1 for CQ fail over to server2.
+    cqDUnitTest.closeServer(server1); 
+    pause(3 * 1000);
+    
+    // Verify CQ Cache results.
+    server2.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqService cqService = null;
+        try {
+          cqService = ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqService.", ex);
+          fail ("Failed to get the internal CqService.", ex);
+        }
+        
+        // Wait till all the region update is performed.
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        while(true){
+          if (region.get(""+ totalObjects) == null){
+            try {
+              Thread.sleep(50);
+            } catch (Exception ex){
+              //ignore.
+            }
+            continue;
+          }
+          break;
+        }
+        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          if (cqQuery.getName().equals(cqName)) {
+            int size = cqQuery.getCqResultKeysSize();
+            if (size != totalObjects) {
+              getLogWriter().info("The number of Cached events " + size + 
+                  " is not equal to the expected size " + totalObjects);
+              HashSet expectedKeys = new HashSet();
+              for (int i = 1; i < totalObjects; i++) {
+                expectedKeys.add("" + i);
+              }
+              Set cachedKeys = cqQuery.getCqResultKeyCache();
+              expectedKeys.removeAll(cachedKeys);
+              getLogWriter().info("Missing keys from the Cache : " + expectedKeys);
+            }
+            assertEquals("The number of keys cached for cq " + cqName + " is wrong.", 
+                totalObjects, cqQuery.getCqResultKeysSize());              
+          }
+        }
+      }
+    });    
+        
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server2); 
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqResultSetUsingPoolOptimizedExecuteDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqResultSetUsingPoolOptimizedExecuteDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqResultSetUsingPoolOptimizedExecuteDUnitTest.java
new file mode 100644
index 0000000..6d465f0
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqResultSetUsingPoolOptimizedExecuteDUnitTest.java
@@ -0,0 +1,223 @@
+/*=========================================================================
+ * 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.query.cq.dunit;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+import com.gemstone.gemfire.cache.CacheException;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.data.Portfolio;
+import com.gemstone.gemfire.cache.query.internal.DefaultQueryService;
+import com.gemstone.gemfire.cache.query.internal.cq.InternalCqQuery;
+import com.gemstone.gemfire.cache.query.internal.cq.ServerCQImpl;
+import com.gemstone.gemfire.cache.query.internal.cq.CqServiceImpl;
+import com.gemstone.gemfire.cache30.CacheSerializableRunnable;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+
+import dunit.Host;
+import dunit.SerializableRunnable;
+import dunit.VM;
+
+public class CqResultSetUsingPoolOptimizedExecuteDUnitTest extends CqResultSetUsingPoolDUnitTest{
+
+  public CqResultSetUsingPoolOptimizedExecuteDUnitTest(String name) {
+    super(name);
+  }
+
+  public void setUp() throws Exception {
+    super.setUp();
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        CqServiceImpl.EXECUTE_QUERY_DURING_INIT = false;
+      }
+    });
+  }
+  
+  @Override
+  public void tearDown2() throws Exception {
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        CqServiceImpl.EXECUTE_QUERY_DURING_INIT = true;
+      }
+    });
+    super.tearDown2();
+  }
+  
+  /**
+   * Tests CQ Result Caching with CQ Failover.
+   * When EXECUTE_QUERY_DURING_INIT is false and new server calls 
+   * execute during HA the results cache is not initialized.
+   * @throws Exception
+   */
+  @Override
+  public void testCqResultsCachingWithFailOver() throws Exception
+  {
+    final Host host = Host.getHost(0);
+    VM server1 = host.getVM(0);
+    VM server2 = host.getVM(1);
+    VM client = host.getVM(2);
+    
+    cqDUnitTest.createServer(server1);
+    
+    final int port1 = server1.invokeInt(CqQueryUsingPoolDUnitTest.class, "getCacheServerPort");
+    final String host0 = getServerHostName(server1.getHost());
+    final int[] ports = AvailablePortHelper.getRandomAvailableTCPPorts(1);
+    
+    String poolName = "testCQFailOver";
+    final String cqName = "testCQFailOver_0";
+    
+    cqDUnitTest.createPool(client, poolName, new String[] {host0, host0}, new int[] {port1, ports[0]});
+
+    // create CQ.
+    cqDUnitTest.createCQ(client, poolName, cqName, cqDUnitTest.cqs[0]);
+    
+    final int numObjects = 300;
+    final int totalObjects = 500;
+    
+    // initialize Region.
+    server1.invoke(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        for (int i = 1; i <= numObjects; i++) {
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+      }
+    });
+    
+    // Keep updating region (async invocation).
+    server1.invokeAsync(new CacheSerializableRunnable("Update Region"){
+      public void run2()throws CacheException {
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        // Update (totalObjects - 1) entries.
+        for (int i = 1; i < totalObjects; i++) {
+          // Destroy entries.
+          if (i > 25 && i < 201) {
+            region.destroy(""+i);
+            continue;
+          }
+          Portfolio p = new Portfolio(i);
+          region.put(""+i, p);
+        }
+        // recreate destroyed entries.
+        for (int j = 26; j < 201; j++) {
+          Portfolio p = new Portfolio(j);
+          region.put(""+j, p);
+        }
+        // Add the last key.
+        Portfolio p = new Portfolio(totalObjects);
+        region.put(""+totalObjects, p);
+      }
+    });
+
+    // Execute CQ.
+    // While region operation is in progress execute CQ.
+    cqDUnitTest.executeCQ(client, cqName, true, null);
+    
+    // Verify CQ Cache results.
+    server1.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqServiceImpl CqServiceImpl = null;
+        try {
+          CqServiceImpl = (com.gemstone.gemfire.cache.query.internal.cq.CqServiceImpl) ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqServiceImpl.", ex);
+          fail ("Failed to get the internal CqServiceImpl.", ex);
+        }
+        
+        // Wait till all the region update is performed.
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        while(true){
+          if (region.get(""+ totalObjects) == null){
+            try {
+              Thread.sleep(50);
+            } catch (Exception ex){
+              //ignore.
+            }
+            continue;
+          }
+          break;
+        }
+        Collection<? extends InternalCqQuery> cqs = CqServiceImpl.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          if (cqQuery.getName().equals(cqName)) {
+            int size = cqQuery.getCqResultKeysSize();
+            if (size != totalObjects) {
+              getLogWriter().info("The number of Cached events " + size + 
+                  " is not equal to the expected size " + totalObjects);
+              HashSet expectedKeys = new HashSet();
+              for (int i = 1; i < totalObjects; i++) {
+                expectedKeys.add("" + i);
+              }
+              Set cachedKeys = cqQuery.getCqResultKeyCache();
+              expectedKeys.removeAll(cachedKeys);
+              getLogWriter().info("Missing keys from the Cache : " + expectedKeys);
+            }
+            assertEquals("The number of keys cached for cq " + cqName + " is wrong.", 
+                totalObjects, cqQuery.getCqResultKeysSize());              
+          }
+        }
+      }
+    });
+    
+    cqDUnitTest.createServer(server2, ports[0]);
+    final int thePort2 = server2.invokeInt(CqQueryUsingPoolDUnitTest.class, "getCacheServerPort");
+    System.out.println("### Port on which server1 running : " + port1 + 
+        " Server2 running : " + thePort2);
+    pause(3 * 1000);
+    
+    // Close server1 for CQ fail over to server2.
+    cqDUnitTest.closeServer(server1); 
+    pause(3 * 1000);
+    
+    // Verify CQ Cache results.
+    server2.invoke(new CacheSerializableRunnable("Verify CQ Cache results"){
+      public void run2()throws CacheException {
+        CqServiceImpl CqServiceImpl = null;
+        try {
+          CqServiceImpl = (CqServiceImpl) ((DefaultQueryService)getCache().getQueryService()).getCqService();
+        } catch (Exception ex) {
+          getLogWriter().info("Failed to get the internal CqServiceImpl.", ex);
+          fail ("Failed to get the internal CqServiceImpl.", ex);
+        }
+        
+        // Wait till all the region update is performed.
+        Region region = getCache().getRegion("/root/" + cqDUnitTest.regions[0]);
+        while(true){
+          if (region.get(""+ totalObjects) == null){
+            try {
+              Thread.sleep(50);
+            } catch (Exception ex){
+              //ignore.
+            }
+            continue;
+          }
+          break;
+        }
+        Collection<? extends InternalCqQuery> cqs = CqServiceImpl.getAllCqs();
+        for (InternalCqQuery cq: cqs){
+          ServerCQImpl cqQuery = (ServerCQImpl)cq;
+          if (cqQuery.getName().equals(cqName)) {
+            int size = cqQuery.getCqResultKeysSize();
+            assertEquals("The number of keys cached for cq " + cqName + " is wrong.", 
+                0, size);              
+          }
+        }
+      }
+    });    
+        
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server2); 
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStateDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStateDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStateDUnitTest.java
new file mode 100644
index 0000000..29ed0a4
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStateDUnitTest.java
@@ -0,0 +1,119 @@
+package com.gemstone.gemfire.cache.query.cq.dunit;
+
+import java.util.Properties;
+
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.dunit.HelperTestCase;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+
+import dunit.AsyncInvocation;
+import dunit.DistributedTestCase;
+import dunit.Host;
+import dunit.SerializableCallable;
+import dunit.VM;
+
+public class CqStateDUnitTest extends HelperTestCase {
+
+  
+  public CqStateDUnitTest(String name) {
+    super(name);
+  }
+  
+  public void testNothingBecauseBug51953() {
+    // remove when bug #51953 is fixed
+  }
+  
+  // this test is disabled due to a 25% failure rate in
+  // CI testing.  See internal ticket #52229
+  public void disabledtestBug51222() throws Exception {
+    //The client can log this when the server shuts down.
+    addExpectedException("Could not find any server");
+    addExpectedException("java.net.ConnectException");
+    final String cqName = "theCqInQuestion";
+    final String regionName = "aattbbss";
+    final Host host = Host.getHost(0);
+    VM serverA = host.getVM(1);
+    VM serverB = host.getVM(2);
+    VM client = host.getVM(3);
+    
+    final int[] ports = AvailablePortHelper.getRandomAvailableTCPPorts(2);
+    startCacheServer(serverA, ports[0], getAuthenticatedServerProperties());
+    createReplicatedRegion(serverA, regionName, null);
+
+    final String host0 = getServerHostName(serverA.getHost());
+    startClient(client, new VM[]{ serverA, serverB }, ports, 1, getClientProperties());
+    createCQ(client, cqName, "select * from /"+ regionName, null);
+    
+    //create the cacheserver but regions must be present first or else cq execute will fail with no region found
+    createCacheServer(serverB, ports[1], getServerProperties(0));
+    createReplicatedRegion(serverB, regionName, null);
+    startCacheServers(serverB);
+    
+    AsyncInvocation async = executeCQ(client, cqName);
+    DistributedTestCase.join(async, 10000, getLogWriter());
+
+    Boolean clientRunning = (Boolean) client.invoke(new SerializableCallable() {
+      @Override
+      public Object call() throws Exception {
+        final CqQuery cq = getCache().getQueryService().getCq(cqName);
+        waitForCriterion(new WaitCriterion() {
+          @Override
+          public boolean done() {
+            return cq.getState().isRunning();
+          }
+          @Override
+          public String description() {
+            return "waiting for Cq to be in a running state: " + cq;
+          }
+        },  30000, 1000, false);
+        return cq.getState().isRunning();
+      }
+    });
+    assertTrue("Client was not running", clientRunning);
+    
+    //hope that server 2 comes up before num retries is exhausted by the execute cq command
+    //hope that the redundancy satisfier sends message and is executed after execute cq has been executed
+    //This is the only way bug 51222 would be noticed
+    //verify that the cq on the server is still in RUNNING state;
+    Boolean isRunning = (Boolean) serverB.invoke(new SerializableCallable() {
+      @Override
+      public Object call() throws Exception {
+        CqQuery cq = getCache().getQueryService().getCqs()[0];
+        return cq.getState().isRunning();
+      }
+      
+    });
+    
+    assertTrue("Cq was not running on server" , isRunning);
+  }
+  
+  public Properties getAuthenticatedServerProperties() {
+    Properties props = new Properties();
+    props.put("mcast-port", "0");
+    props.put("security-client-accessor",
+        "com.gemstone.gemfire.cache.query.dunit.CloseCacheAuthorization.create");
+    props.put("security-client-accessor-pp",
+        "com.gemstone.gemfire.cache.query.dunit.CloseCacheAuthorization.create");
+    props.put("security-client-authenticator",
+        "templates.security.DummyAuthenticator.create");
+    return props;
+  }
+  
+  public Properties getServerProperties() {
+    Properties props = new Properties();
+    props.put("mcast-port", "0");
+    return props;
+  }
+  
+  public Properties getClientProperties() {
+    Properties props = new Properties();
+    props.put("security-client-auth-init",
+        "templates.security.UserPasswordAuthInit.create");
+    
+    props.put("security-username", "root");
+    props.put("security-password", "root");
+    return props;
+  }  
+  
+}
+

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStatsDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStatsDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStatsDUnitTest.java
new file mode 100644
index 0000000..b66228e
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStatsDUnitTest.java
@@ -0,0 +1,433 @@
+/*=========================================================================
+ * 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.query.cq.dunit;
+
+import java.util.Collection;
+
+import com.gemstone.gemfire.cache.CacheException;
+import com.gemstone.gemfire.cache.query.CqException;
+import com.gemstone.gemfire.cache.query.CqQuery;
+import com.gemstone.gemfire.cache.query.CqServiceStatistics;
+import com.gemstone.gemfire.cache.query.CqStatistics;
+import com.gemstone.gemfire.cache.query.QueryService;
+import com.gemstone.gemfire.cache.query.internal.CqQueryVsdStats;
+import com.gemstone.gemfire.cache.query.internal.CqStateImpl;
+import com.gemstone.gemfire.cache.query.internal.DefaultQueryService;
+import com.gemstone.gemfire.cache.query.internal.cq.CqQueryImpl;
+import com.gemstone.gemfire.cache.query.internal.cq.CqService;
+import com.gemstone.gemfire.cache.query.internal.cq.CqServiceImpl;
+import com.gemstone.gemfire.cache.query.internal.cq.CqServiceVsdStats;
+import com.gemstone.gemfire.cache.query.internal.cq.InternalCqQuery;
+import com.gemstone.gemfire.cache30.CacheSerializableRunnable;
+import com.gemstone.gemfire.cache30.CacheTestCase;
+
+import dunit.Host;
+import dunit.SerializableRunnable;
+import dunit.VM;
+
+/**
+ * This class tests the ContiunousQuery mechanism in GemFire.
+ * This includes the test with different data activities.
+ *
+ * @author Rao
+ */
+public class CqStatsDUnitTest extends CacheTestCase {
+
+  private CqQueryDUnitTest cqDUnitTest = new CqQueryDUnitTest("CqStatsDUnitTest");
+  
+  public CqStatsDUnitTest(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();
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        getSystem();
+      }
+    });
+    
+  }
+  
+  public void validateCQStats(VM vm, final String cqName,
+      final int creates,
+      final int updates,
+      final int deletes,
+      final int totalEvents,
+      final int cqListenerInvocations) {
+    vm.invoke(new CacheSerializableRunnable("Validate CQs") {
+      public void run2() throws CacheException {
+        getLogWriter().info("### Validating CQ Stats. ### " + cqName);
+//      Get CQ Service.
+        QueryService qService = null;
+        try {          
+          qService = getCache().getQueryService();
+        } catch (Exception cqe) {
+          cqe.printStackTrace();
+          fail("Failed to get query service.");
+        }
+        
+        CqService cqService = null;
+        try {
+          cqService = ((DefaultQueryService)qService).getCqService();
+        }
+        catch (CqException e) {
+          // TODO Auto-generated catch block
+          e.printStackTrace();
+          fail("Failed to get CqService, CQ : " + cqName);
+        }
+        Collection<? extends InternalCqQuery> cqs = cqService.getAllCqs();
+        if (cqs.size() == 0) {
+          fail("Failed to get CqQuery for CQ : " + cqName);
+        }
+        CqQueryImpl cQuery = (CqQueryImpl) cqs.iterator().next();
+        
+        CqStatistics cqStats = cQuery.getStatistics();
+        CqQueryVsdStats cqVsdStats = cQuery.getVsdStats();
+        if (cqStats == null || cqVsdStats == null) {
+          fail("Failed to get CqQuery Stats for CQ : " + cqName);
+        }
+        
+        getCache().getLogger().info("#### CQ stats for " + cQuery.getName() + ": " + 
+            " Events Total: " + cqStats.numEvents() +
+            " Events Created: " + cqStats.numInserts() +
+            " Events Updated: " + cqStats.numUpdates() +
+            " Events Deleted: " + cqStats.numDeletes() +
+            " CQ Listener invocations: " + cqVsdStats.getNumCqListenerInvocations() +
+            " Initial results time (nano sec): " + cqVsdStats.getCqInitialResultsTime());
+        
+        
+//      Check for totalEvents count.
+        if (totalEvents != CqQueryDUnitTest.noTest) {
+//        Result size validation.
+          assertEquals("Total Event Count mismatch", totalEvents, cqStats.numEvents());
+        }
+        
+//      Check for create count.
+        if (creates != CqQueryDUnitTest.noTest) {
+          assertEquals("Create Event mismatch", creates, cqStats.numInserts());
+        }
+        
+//      Check for update count.
+        if (updates != CqQueryDUnitTest.noTest) {
+          assertEquals("Update Event mismatch", updates, cqStats.numUpdates());
+        }
+        
+//      Check for delete count.
+        if (deletes != CqQueryDUnitTest.noTest) {
+          assertEquals("Delete Event mismatch", deletes, cqStats.numDeletes());
+        }
+        
+//      Check for CQ listener invocations.
+        if (cqListenerInvocations != CqQueryDUnitTest.noTest) {
+          assertEquals("CQ Listener invocations mismatch", cqListenerInvocations, cqVsdStats.getNumCqListenerInvocations());
+        }
+////      Check for initial results time.
+//        if (initialResultsTime != CqQueryDUnitTest.noTest && cqVsdStats.getCqInitialResultsTime() <= 0) {
+//          assertEquals("Initial results time mismatch", initialResultsTime, cqVsdStats.getCqInitialResultsTime());
+//        }
+      }
+    });
+  }
+  
+  private void validateCQServiceStats(VM vm,
+      final int created,
+      final int activated,
+      final int stopped,
+      final int closed,
+      final int cqsOnClient,
+      final int cqsOnRegion,
+      final int clientsWithCqs) {
+    vm.invoke(new CacheSerializableRunnable("Validate CQ Service Stats") {
+      public void run2() throws CacheException {
+        getLogWriter().info("### Validating CQ Service Stats. ### ");
+//      Get CQ Service.
+        QueryService qService = null;
+        try {          
+          qService = getCache().getQueryService();
+        } catch (Exception cqe) {
+          cqe.printStackTrace();
+          fail("Failed to getCQService.");
+        }
+        CqServiceStatistics cqServiceStats = null;        
+        cqServiceStats = qService.getCqStatistics();
+        CqServiceVsdStats cqServiceVsdStats = null;
+        try {
+          cqServiceVsdStats = ((CqServiceImpl) ((DefaultQueryService)qService).getCqService()).stats;
+        }
+        catch (CqException e) {
+          // TODO Auto-generated catch block
+          e.printStackTrace();
+        }
+        if (cqServiceStats == null) {
+          fail("Failed to get CQ Service Stats");
+        }
+        
+        getCache().getLogger().info("#### CQ Service stats: " + 
+            " CQs created: " + cqServiceStats.numCqsCreated() +
+            " CQs active: " + cqServiceStats.numCqsActive() +
+            " CQs stopped: " + cqServiceStats.numCqsStopped() +
+            " CQs closed: " + cqServiceStats.numCqsClosed() +
+            " CQs on Client: " + cqServiceStats.numCqsOnClient() +
+            " CQs on region /root/regionA : " + cqServiceVsdStats.numCqsOnRegion("/root/regionA") +
+            " Clients with CQs: " + cqServiceVsdStats.getNumClientsWithCqs());
+        
+        
+        //        Check for created count.
+        if (created != CqQueryDUnitTest.noTest) {
+          assertEquals("Number of CQs created mismatch", created, cqServiceStats.numCqsCreated());
+        }
+        
+        //        Check for activated count.
+        if (activated != CqQueryDUnitTest.noTest) {
+          assertEquals("Number of CQs activated mismatch", activated, cqServiceStats.numCqsActive());
+        }
+        
+        //        Check for stopped count.
+        if (stopped != CqQueryDUnitTest.noTest) {
+          assertEquals("Number of CQs stopped mismatch", stopped, cqServiceStats.numCqsStopped());
+        }
+        
+        //        Check for closed count.
+        if (closed != CqQueryDUnitTest.noTest) {
+          assertEquals("Number of CQs closed mismatch", closed, cqServiceStats.numCqsClosed());
+        }
+        
+        // Check for CQs on client count.
+        if (cqsOnClient != CqQueryDUnitTest.noTest) {
+          assertEquals("Number of CQs on client mismatch", cqsOnClient, cqServiceStats.numCqsOnClient());
+        }
+        
+        // Check for CQs on region.
+        if (cqsOnRegion != CqQueryDUnitTest.noTest) {
+          assertEquals("Number of CQs on region /root/regionA mismatch", 
+              cqsOnRegion, cqServiceVsdStats.numCqsOnRegion("/root/regionA"));
+        }
+        
+        // Check for clients with CQs count.
+        if (clientsWithCqs != CqQueryDUnitTest.noTest) {
+          assertEquals("Clints with CQs mismatch", 
+              clientsWithCqs, cqServiceVsdStats.getNumClientsWithCqs());
+        }
+      }
+    });
+  }
+  
+  private static final int PAUSE = 8 * 1000; // 5 * 1000
+  
+  /**
+   * Test for CQ and CQ Service Statistics
+   * @throws Exception
+   */
+  public void testCQStatistics() throws Exception {
+    
+    final Host host = Host.getHost(0);
+    VM server = host.getVM(0);
+    VM client = host.getVM(1);
+    
+    /* Init Server and Client */
+    cqDUnitTest.createServer(server);
+    final int port = server.invokeInt(CqQueryDUnitTest.class, "getCacheServerPort");
+    final String host0 = getServerHostName(server.getHost());
+    cqDUnitTest.createClient(client, port, host0);
+    
+    /* Create CQs. */
+    cqDUnitTest.createCQ(client, "testCQStatistics_0", cqDUnitTest.cqs[0]);
+    
+    /* Init values at server. */
+    int size = 100;
+    cqDUnitTest.createValues(server, cqDUnitTest.regions[0], size);
+    
+    cqDUnitTest.executeCQ(client, "testCQStatistics_0", true, null);
+    
+    // Wait for CQ to be executed.
+    cqDUnitTest.waitForCqState(client, "testCQStatistics_0", CqStateImpl.RUNNING);    
+
+    // Test CQ stats
+    validateCQStats(client, "testCQStatistics_0", 0, 0, 0, 0, 0);
+    
+    // The stat would have not updated yet.
+    // Commenting out the following check; the check for resultset initialization 
+    // is anyway done in the next validation. 
+    // validateCQStats(server, "testCQStatistics_0", 0, 0, 0, 0, CqQueryDUnitTest.noTest, 1);
+    
+    /* Init values at server. */
+    cqDUnitTest.createValues(server, cqDUnitTest.regions[0], 200);
+    // Wait for client to Synch.
+    cqDUnitTest.waitForCreated(client, "testCQStatistics_0", CqQueryDUnitTest.KEY+200);
+    pause(PAUSE);
+    size = 200;
+    
+    // validate CQs.
+    cqDUnitTest.validateCQ(client, "testCQStatistics_0",
+        /* resultSize: */ CqQueryDUnitTest.noTest,
+        /* creates: */ 100,
+        /* updates: */ 100,
+        /* deletes; */ 0,
+        /* queryInserts: */ 100,
+        /* queryUpdates: */ 100,
+        /* queryDeletes: */ 0,
+        /* totalEvents: */ 200);
+    
+    // Test CQ stats
+    validateCQStats(client, "testCQStatistics_0", 100, 100, 0, 200, 200);
+    //We don't have serverside CQ name
+    validateCQStats(server, "testCQStatistics_0", 100, 100, 0, 200, CqQueryDUnitTest.noTest);
+    
+    /* Delete values at server. */
+    cqDUnitTest.deleteValues(server, cqDUnitTest.regions[0], 100);
+    // Wait for client to Synch.
+    cqDUnitTest.waitForDestroyed(client, "testCQStatistics_0", CqQueryDUnitTest.KEY+100);
+    size = 10;
+    pause(PAUSE);
+    
+    cqDUnitTest.validateCQ(client, "testCQStatistics_0",
+        /* resultSize: */ CqQueryDUnitTest.noTest,
+        /* creates: */100,
+        /* updates: */ 100,
+        /* deletes; */ 100,
+        /* queryInserts: */ 100,
+        /* queryUpdates: */ 100,
+        /* queryDeletes: */ 100,
+        /* totalEvents: */ 300);
+    
+    // Test CQ stats
+    validateCQStats(client, "testCQStatistics_0", 100, 100, 100, 300, 300);
+    //We don't have serverside CQ name
+    validateCQStats(server, "testCQStatistics_0", 100, 100, 100, 300, CqQueryDUnitTest.noTest);
+    
+    // Test  CQ Close
+    cqDUnitTest.closeCQ(client, "testCQStatistics_0");
+    pause(PAUSE);
+    
+    // Close.
+    cqDUnitTest.closeClient(client);
+    cqDUnitTest.closeServer(server);
+  }
+  /**
+   * Test for CQ Service Statistics
+   * @throws Exception
+   */
+  public void testCQServiceStatistics() throws Exception {
+    
+    final Host host = Host.getHost(0);
+    VM server = host.getVM(0);
+    VM client1 = host.getVM(1);
+    VM client2 = host.getVM(2);
+    
+    
+    /* Init Server and Client */
+    cqDUnitTest.createServer(server);
+    final int port = server.invokeInt(CqQueryDUnitTest.class, "getCacheServerPort");
+    final String host0 = getServerHostName(server.getHost());
+    cqDUnitTest.createClient(client1, port, host0);
+    cqDUnitTest.createClient(client2, port, host0);
+    
+    /* Create CQs. */
+    String cqName = new String("testCQServiceStatistics_0");
+    String cqName10 = new String("testCQServiceStatistics_10");   
+    cqDUnitTest.createCQ(client1, cqName, cqDUnitTest.cqs[0]);
+    cqDUnitTest.createCQ(client2, cqName10, cqDUnitTest.cqs[2]); 
+    pause(PAUSE);
+    // Test CQ Service stats
+    getCache().getLogger().info("Validating CQ Service stats on clients: #1");
+    validateCQServiceStats(client1, 1, 0, 1, 0, 1, 1, CqQueryDUnitTest.noTest);
+    validateCQServiceStats(server, 0, 0, 0, 0, CqQueryDUnitTest.noTest, 0, 0);
+    
+    cqDUnitTest.executeCQ(client1, cqName, false, null);
+    cqDUnitTest.executeCQ(client2, cqName10, false, null);
+    pause(PAUSE);
+    
+    getCache().getLogger().info("Validating CQ Service stats on clients: #2");
+    validateCQServiceStats(client1, 1, 1, 0, 0, 1, 1, CqQueryDUnitTest.noTest);
+    validateCQServiceStats(client2, 1, 1, 0, 0, 1, CqQueryDUnitTest.noTest, CqQueryDUnitTest.noTest);
+    
+    getCache().getLogger().info("Validating CQ Service stats on server: #1");
+    validateCQServiceStats(server, 2, 2, 0, 0, CqQueryDUnitTest.noTest, 1, 2);
+    
+    /* Init values at server. */
+    int size = 10;
+    cqDUnitTest.createValues(server, cqDUnitTest.regions[0], size);
+    // Wait for client to Synch.
+    cqDUnitTest.waitForCreated(client1, "testCQServiceStatistics_0", CqQueryDUnitTest.KEY+size);
+    
+    // validate CQs.
+    cqDUnitTest.validateCQ(client1, cqName,
+        /* resultSize: */ CqQueryDUnitTest.noTest,
+        /* creates: */ size,
+        /* updates: */ 0,
+        /* deletes; */ 0,
+        /* queryInserts: */ size,
+        /* queryUpdates: */ 0,
+        /* queryDeletes: */ 0,
+        /* totalEvents: */ size);
+    pause(PAUSE);
+    
+    // Test CQ Service stats
+    getCache().getLogger().info("Validating CQ Service stats on clients: #3");
+    validateCQServiceStats(client1, 1, 1, 0, 0, 1, 1, CqQueryDUnitTest.noTest);
+    validateCQServiceStats(client2, 1, 1, 0, 0, 1, CqQueryDUnitTest.noTest, CqQueryDUnitTest.noTest);
+    
+    getCache().getLogger().info("Validating CQ Service stats on server: #1");
+    validateCQServiceStats(server, 2, 2, 0, 0, CqQueryDUnitTest.noTest, 1, 2);
+    
+    
+    //Create CQs with no name, execute, and close. 
+    cqDUnitTest.createAndExecCQNoName(client1, cqDUnitTest.cqs[0]); 
+    pause(PAUSE);      
+    
+    // Test CQ Service stats
+    getCache().getLogger().info("Validating CQ Service stats on client: #4");
+    validateCQServiceStats(client1, 21, 1, 0, 20, 1, 1, CqQueryDUnitTest.noTest);
+    
+    getCache().getLogger().info("Validating CQ Service stats on server: #2");
+    validateCQServiceStats(server, 22, 2, 0, 20, CqQueryDUnitTest.noTest, 1, 2);
+    
+    // Test  CQ Close
+    cqDUnitTest.closeCQ(client1, cqName);
+    pause(PAUSE);      
+    
+    // Test CQ Service stats
+    getCache().getLogger().info("Validating CQ Service stats on client: #5");
+    validateCQServiceStats(client1, 21, 0, 0, 21, 0, 0, CqQueryDUnitTest.noTest);
+    
+    getCache().getLogger().info("Validating CQ Service stats on server: #3");
+    validateCQServiceStats(server, 22, 1, 0, 21, CqQueryDUnitTest.noTest, 0, 1);
+    
+    //Test stop CQ
+    cqDUnitTest.stopCQ(client2, cqName10);
+    pause(PAUSE);
+    
+    // Test CQ Service stats
+    getCache().getLogger().info("Validating CQ Service stats on client: #6");
+    validateCQServiceStats(client2, 1, 0, 1, 0, 1, CqQueryDUnitTest.noTest, CqQueryDUnitTest.noTest);
+    getCache().getLogger().info("Validating CQ Service stats on server: #4");
+    validateCQServiceStats(server, 22, 0, 1, 21, CqQueryDUnitTest.noTest, CqQueryDUnitTest.noTest, 1);
+    
+    // Test  CQ Close
+    cqDUnitTest.closeCQ(client2, cqName10);
+    pause(PAUSE);
+    
+    // Test CQ Service stats
+    getCache().getLogger().info("Validating CQ Service stats on client: #7");
+    validateCQServiceStats(client1, 21, 0, 0, 21, 0, 0, CqQueryDUnitTest.noTest);
+    validateCQServiceStats(client2, 1, 0, 0, 1, 0, 0, CqQueryDUnitTest.noTest);
+    getCache().getLogger().info("Validating CQ Service stats on server: #5");
+    validateCQServiceStats(server, 22, 0, 0, 22, CqQueryDUnitTest.noTest, 0, 0);
+    
+    // Close.
+    cqDUnitTest.closeClient(client1);
+    cqDUnitTest.closeClient(client2);
+    cqDUnitTest.closeServer(server);
+  }
+
+} 

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStatsOptimizedExecuteDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStatsOptimizedExecuteDUnitTest.java b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStatsOptimizedExecuteDUnitTest.java
new file mode 100644
index 0000000..b7eaaa2
--- /dev/null
+++ b/gemfire-cq/src/test/java/com/gemstone/gemfire/cache/query/cq/dunit/CqStatsOptimizedExecuteDUnitTest.java
@@ -0,0 +1,43 @@
+/*=========================================================================
+ * 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.query.cq.dunit;
+
+import com.gemstone.gemfire.cache.query.internal.cq.CqService;
+import com.gemstone.gemfire.cache.query.internal.cq.CqServiceImpl;
+
+import dunit.SerializableRunnable;
+
+/**
+ * Test class for testing {@link CqService#EXECUTE_QUERY_DURING_INIT} flag
+ *
+ */
+public class CqStatsOptimizedExecuteDUnitTest extends CqStatsDUnitTest{
+
+  public CqStatsOptimizedExecuteDUnitTest(String name) {
+    super(name);
+  }
+
+  public void setUp() throws Exception {
+    super.setUp();
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        CqServiceImpl.EXECUTE_QUERY_DURING_INIT = false;
+      }
+    });
+  }
+  
+  @Override
+  public void tearDown2() throws Exception {
+    invokeInEveryVM(new SerializableRunnable("getSystem") {
+      public void run() {
+        CqServiceImpl.EXECUTE_QUERY_DURING_INIT = true;
+      }
+    });
+    super.tearDown2();
+  }
+}


Mime
View raw message