geode-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From upthewatersp...@apache.org
Subject [15/70] [partial] incubator-geode git commit: WAN and CQ code drop under the Pivotal SGA
Date Thu, 28 Jan 2016 18:13:17 GMT
http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/ReplicatedRegion_ParallelWANPropogationDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/ReplicatedRegion_ParallelWANPropogationDUnitTest.java b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/ReplicatedRegion_ParallelWANPropogationDUnitTest.java
new file mode 100644
index 0000000..1c073f9
--- /dev/null
+++ b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/ReplicatedRegion_ParallelWANPropogationDUnitTest.java
@@ -0,0 +1,1124 @@
+/*=========================================================================
+ * 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.internal.cache.wan.misc;
+
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.StringTokenizer;
+
+import com.gemstone.gemfire.cache.AttributesFactory;
+import com.gemstone.gemfire.cache.CacheClosedException;
+import com.gemstone.gemfire.cache.DataPolicy;
+import com.gemstone.gemfire.cache.EntryExistsException;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.RegionDestroyedException;
+import com.gemstone.gemfire.cache.Scope;
+import com.gemstone.gemfire.cache.client.ServerConnectivityException;
+import com.gemstone.gemfire.cache.client.ServerOperationException;
+import com.gemstone.gemfire.cache.wan.GatewaySender;
+import com.gemstone.gemfire.distributed.internal.ReplyException;
+import com.gemstone.gemfire.internal.cache.BucketRegion;
+import com.gemstone.gemfire.internal.cache.ForceReattemptException;
+import com.gemstone.gemfire.internal.cache.PartitionedRegion;
+import com.gemstone.gemfire.internal.cache.PrimaryBucketException;
+import com.gemstone.gemfire.internal.cache.PutAllPartialResultException;
+import com.gemstone.gemfire.internal.cache.RegionQueue;
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase;
+
+import dunit.AsyncInvocation;
+import dunit.DistributedTestCase;
+import dunit.DistributedTestCase.ExpectedException;
+import dunit.DistributedTestCase.WaitCriterion;
+
+public class ReplicatedRegion_ParallelWANPropogationDUnitTest extends WANTestBase{
+
+  public ReplicatedRegion_ParallelWANPropogationDUnitTest(String name) {
+    super(name);
+    // TODO Auto-generated constructor stub
+  }
+
+  final String expectedExceptions = null;
+  
+  
+  /**
+   * 
+   */
+  private static final long serialVersionUID = 1L;
+
+  public void test_DR_PGS_1Nodes_Put_Receiver() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", null, isOffHeap() });
+
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap() });
+
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+          true, 10, 100, false, false, null, true });
+
+      vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln1" });
+      fail("Expected GatewaySenderConfigException where parallel gateway sender can not be used with replicated region");
+    }
+    catch (Exception e) {
+      if (!e.getCause().getMessage()
+          .contains("can not be used with replicated region")) {
+        fail("Expected GatewaySenderConfigException where parallel gateway sender can not be used with replicated region");
+      }
+    }
+  }
+  
+  /*1. Validate that parallelGAtewaySenderId can be added to distributed region
+   *Region distributed ack/noack + PGS
+   *1. Find out the restrictions on totalNumBuckets on shadowPR
+   *2. Find out the restrictions on redundancy on shadowPR
+   *3. Find out the restrictions on localMaxMemory on shadowPR
+   *4. Find out the best way user will specify PR attributes to PGS
+   *5. Find out the restrictions on ordering.
+   *6. put on region populates the queue    
+   *7. put on region reaches to remote site. Dispatcher works as expected
+   *8. m1 and m2 has DR(ack/noack). put on DR from m1 populates queue on both m1 and m2. Validate that remote site got all the events
+   *9. m1 and m2 has DR(ack/noack). create/put/destroy/operations populates the queue. Validate that remote site got correct events
+   *10. m1 and m2 has DR(ack/noack). localDestory is called on m1's DR. This locally destroys M1's shadowPr
+   *11. m1 and m2 has DR(ack/noack). destory is called on m1's DR. This destroys entire shadowPr on m1 and m2
+   *12. m1 and m2 has DR(ack/noack). close Region is called on m1's DR. This locally destroys shadowPr on m1
+   *13. m1 and m2 has DR(ack/noack). cache.close on m1'. This locally destroys shadowPr on m1
+   *14. Validate HA scenario does not cause any event loss
+   *15. PDX events of DR are propagated to remote sites
+   *16. validate stats
+   *17: PR and DR regions with same name.. Can this be created. If yes then how to differentiate these 2 different shadowPR. 
+   *18. test for redundancy. FOR SPR's redundancy will be equal to the number of nodes where DR is present. Max is 3. I know this needs to be figure it out at runtime. 
+   *19. test without prviding diskstorename..I suspect some problem with this code. diskStoreName=null looks like this is not handled very well. need to verify
+   *20. PAralleGatewaySenderQueue#addPR method has multiple check for inPersistenceEnabled. Can's we do it with only one check.  
+  */
+  
+  /**
+   * Test to validate that created parallel gatewaySenderls id can be added to
+   * distributed region
+   * Below test is disabled intentionally
+    1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+    2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+    3> We have to enabled it in next release
+    4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0
+   */
+  public void DISABLED_test_PGS_Started_DR_CREATED_NO_RECEIVER() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+/*      ExpectedException exp1 = addExpectedException(GatewaySenderException.class
+          .getName(), vm4);
+      ExpectedException exp2 = addExpectedException(InterruptedException.class
+          .getName(), vm4);
+      try {
+*/        vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+            true, 10, 100, false, false, null, false });
+        vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+            testName + "_RR", "ln1", isOffHeap()  });
+        vm4.invoke(WANTestBase.class, "doPuts", new Object[] {
+            testName + "_RR", 1000 });
+        vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] {
+            "ln1", 1000 });
+
+/*      }
+      finally {
+        exp1.remove();
+        exp2.remove();
+      }
+*/    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+
+  /**
+   * Test to validate that distributed region with given paralleGAtewaySender id
+   * is created first and then a same parallelGatewaySender is created
+   * a single put in DR is enqueued in parallelQueue
+   * Below test is disabled intentionally
+    1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+    2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+    3> We have to enabled it in next release
+    4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0
+   */
+  public void DISABLED_test_DR_CREATED_PGS_STARTED_NO_RECEIVER() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln1", isOffHeap()  });
+/*      ExpectedException exp1 = addExpectedException(GatewaySenderException.class
+          .getName(), vm4);
+      ExpectedException exp2 = addExpectedException(InterruptedException.class
+          .getName(), vm4);
+      try {*/
+        vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+            true, 10, 100, false, false, null, false });
+        vm4.invoke(WANTestBase.class, "doPuts", new Object[] {
+            testName + "_RR", 1000 });
+        vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] {
+            "ln1", 1000 });
+/*      }
+      finally {
+        exp1.remove();
+        exp2.remove();
+      }
+*/    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_1Node_Put_ValidateQueue_No_Receiver() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap()  });
+      
+/*      ExpectedException exp1 = addExpectedException(GatewaySenderException.class
+          .getName(), vm4);
+      ExpectedException exp2 = addExpectedException(InterruptedException.class
+          .getName(), vm4);
+      try {*/
+        vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+            true, 10, 100, false, false, null, true });
+        vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln1" });
+
+        vm4.invoke(WANTestBase.class, "doPuts", new Object[] {
+            testName + "_RR", 10000 });
+
+        vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_RR", 10000 });
+        vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] {
+            "ln1", 10000 });
+/*      }
+    finally {
+      exp1.remove();
+      exp2.remove();
+    }
+     */ 
+    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_2Nodes_Put_ValidateQueue_No_Receiver() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap()  });
+      vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln1", isOffHeap()  });
+      
+/*      ExpectedException exp1 = addExpectedException(
+          GatewaySenderException.class.getName());
+      ExpectedException exp2 = addExpectedException(
+          InterruptedException.class.getName());
+      ExpectedException exp3 = addExpectedException(
+          CacheClosedException.class.getName());
+      try {
+*/        vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+            true, 10, 100, false, false, null, true });
+        vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+            true, 10, 100, false, false, null, true });
+
+        vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln1" });
+        vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln1" });
+
+        vm4.invoke(WANTestBase.class, "doPuts", new Object[] {
+            testName + "_RR", 1000 });
+
+        vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_RR", 1000 });
+        vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_RR", 1000 });
+
+        vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] {
+            "ln1", 1000 });
+        vm5.invoke(WANTestBase.class, "validateQueueContents", new Object[] {
+            "ln1", 1000 });
+
+/*      }
+      finally {
+        exp1.remove();
+        exp2.remove();
+        exp3.remove();
+      }
+*/      
+    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+  
+//  public void test_DR_PGS_ORDERPOLICY_PARTITION_EXPECTException(){
+//    
+//  }
+//  public void test_DR_PGS_DISKSTORE_NAME_PROVIDED_VALIDATE_DISK(){
+//    
+//  }
+//  public void test_DR_PGS_DISKSTORE_NAME_NOT_PROVIDED_VALIDATE_DISK(){
+//    
+//  }
+//  
+//  public void test_DR_PGS_START_STOP_START(){
+//    
+//  }
+//
+//  public void test_DR_PGS_PERSISTENCE_START_STOP_START(){
+//    
+//  }
+//  
+//  public void test_DR_PGS_START_PAUSE_STOP(){
+//    
+//  }
+//
+//  public void test_DR_PGS_START_PAUSE_RESUME_VALIDATE_RECEIVER(){
+//    
+//  }
+
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_1Nodes_Put_Receiver() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", null, isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+        true, 10, 100, false, false, null, true});
+
+      vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln1"});
+      
+      vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+        1000});
+     
+      vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000});
+      
+      vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      
+      vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_RR", 1000});
+    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_2Nodes_Put_Receiver() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", null, isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap()  });
+      vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln1", isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+        true, 10, 100, false, false, null, true});
+      vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+      true, 10, 100, false, false, null, true});
+
+      vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln1"});
+      vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln1"});
+      
+      vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+        1000 });
+     
+      vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      
+      vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      vm5.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      
+      vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_RR", 1000 });
+    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_2Nodes_EMPTY_Put_Receiver() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", null, isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", Scope.DISTRIBUTED_ACK, DataPolicy.EMPTY, isOffHeap()  });
+      vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln1", Scope.DISTRIBUTED_ACK, DataPolicy.REPLICATE, isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+        true, 10, 100, false, false, null, true});
+      vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+      true, 10, 100, false, false, null, true});
+
+      vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln1"});
+      vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln1"});
+      
+      vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+        1000 });
+     
+//      vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+//        1000 });
+      vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      
+      vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      vm5.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      
+      vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_RR", 1000 });
+    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PR_PGS_4Nodes_Put_Receiver_2Nodes() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      
+      vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", null, isOffHeap()  });
+      vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", null, isOffHeap()  });
+      
+      vm2.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 1, 100, isOffHeap()  });
+      vm3.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 1, 100, isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      
+      vm4.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+          testName + "_PR", "ln", 1, 100, isOffHeap()  });
+      vm5.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+          testName + "_PR", "ln", 1, 100, isOffHeap()  });
+      vm6.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+          testName + "_PR", "ln", 1, 100, isOffHeap()  });
+      vm7.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+          testName + "_PR", "ln", 1, 100, isOffHeap()  });
+
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 10, 100, false, false, null, true });
+      vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 10, 100, false, false, null, true });
+      vm6.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 10, 100, false, false, null, true });
+      vm7.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 10, 100, false, false, null, true });
+
+      vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln"});
+      vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln"});
+      vm6.invoke(WANTestBase.class, "startSender", new Object[] { "ln"});
+      vm7.invoke(WANTestBase.class, "startSender", new Object[] { "ln"});
+      
+      vm4.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm5.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm6.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm7.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+
+      vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+        1000 });
+      vm5.invoke(WANTestBase.class, "doNextPuts", new Object[] { testName + "_PR",
+        1000, 2000 });
+      
+      vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_PR",
+        1000 });
+      vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      
+      vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln",
+        0 });
+      vm5.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln",
+        0 });
+      
+/*      ExpectedException exp1 = addExpectedException(CacheClosedException.class
+          .getName());
+      try {*/
+        vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_RR", 1000 });
+        vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_RR", 1000 });
+        vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_PR", 1000 });
+        vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_PR", 1000 });
+/*      }
+      finally {
+        exp1.remove();
+      }
+*/    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_NOMANUALSTART_4Nodes_Put_ValidateReceiver() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", null, isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+          true, 10, 100, false, false, null, false });
+      vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+          true, 10, 100, false, false, null, false });
+      vm6.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+          true, 10, 100, false, false, null, false });
+      vm7.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+          true, 10, 100, false, false, null, false });
+    
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap()  });
+      vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap()  });
+      vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap()  });
+      vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln1", isOffHeap()  });
+
+      vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+        1000 });
+     
+      vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      vm6.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      vm7.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+
+      
+      vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      vm5.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      vm6.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      vm7.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      
+/*      ExpectedException exp1 = addExpectedException(CacheClosedException.class
+          .getName());
+      try {*/
+        vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_RR", 1000 });
+/*      }
+      finally {
+        exp1.remove();
+      }*/
+    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_4Nodes_Put_CLOSE4NODESCACHE_RECREATE_PUT_ValidateReceiver()
+      throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 100, 10, false, false, null, true });
+      vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 100, 10, false, false, null, true });
+      vm6.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 100, 10, false, false, null, true });
+      vm7.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 100, 10, false, false, null, true });
+
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+
+      vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+      vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+      vm6.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+      vm7.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+      vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", null, isOffHeap()  });
+      vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", null, isOffHeap()  });
+
+      // before doing any puts, let the senders be running in order to ensure
+      // that
+      // not a single event will be lost
+      vm4.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm5.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm6.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm7.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+
+      vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+          1000 });
+      vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+          testName + "_RR", 1000 });
+
+/*      ExpectedException exp1 = addExpectedException(CacheClosedException.class
+          .getName());
+      try {*/
+        vm4.invoke(WANTestBase.class, "killSender", new Object[] {});
+        vm5.invoke(WANTestBase.class, "killSender", new Object[] {});
+        vm6.invoke(WANTestBase.class, "killSender", new Object[] {});
+        vm7.invoke(WANTestBase.class, "killSender", new Object[] {});
+/*      }
+      finally {
+        exp1.remove();
+      }*/
+
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 100, 10, false, false, null, true });
+      vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 100, 10, false, false, null, true });
+      vm6.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 100, 10, false, false, null, true });
+      vm7.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+          true, 100, 10, false, false, null, true });
+
+      vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+      vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+          testName + "_RR", "ln", isOffHeap()  });
+
+      vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+      vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+      vm6.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+      vm7.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+      vm4.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm5.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm6.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      vm7.invoke(WANTestBase.class, "waitForSenderRunningState",
+          new Object[] { "ln" });
+      // ------------------------------------------------------------------------------------
+
+      vm4.invoke(WANTestBase.class, "doNextPuts", new Object[] {
+          testName + "_RR", 1000, 2000 });
+
+      // verify all buckets drained on all sender nodes.
+      vm4.invoke(WANTestBase.class,
+          "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+      vm5.invoke(WANTestBase.class,
+          "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+      vm6.invoke(WANTestBase.class,
+          "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+      vm7.invoke(WANTestBase.class,
+          "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+
+/*      exp1 = addExpectedException(CacheClosedException.class.getName());
+      try {*/
+        vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_RR", 2000 });
+        vm3.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+            testName + "_RR", 2000 });
+/*      }
+      finally {
+        exp1.remove();
+      }*/
+    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_NO_ACK_PGS_2Nodes_Put_ValidateQueue_Receiver() throws Exception {
+    try {
+      Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+          "createFirstLocatorWithDSId", new Object[] { 1 });
+      Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+          "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+      vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+      vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", null, isOffHeap()  });
+      
+      vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+      
+      vm4.invoke(WANTestBase.class,
+          "createReplicatedRegion", new Object[] { testName + "_RR", "ln1",
+              Scope.DISTRIBUTED_NO_ACK, DataPolicy.REPLICATE, isOffHeap()  });
+      vm5.invoke(WANTestBase.class,
+          "createReplicatedRegion", new Object[] { testName + "_RR", "ln1",
+              Scope.DISTRIBUTED_NO_ACK, DataPolicy.REPLICATE, isOffHeap()   });
+      
+      vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+        true, 10, 100, false, false, null, true});
+      vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln1", 2,
+      true, 10, 100, false, false, null, true});
+
+      vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln1"});
+      vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln1"});
+      
+      vm4.invoke(WANTestBase.class, "doPuts", new Object[] { testName + "_RR",
+        1000 });
+     
+      vm4.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      vm5.invoke(WANTestBase.class, "validateRegionSize", new Object[] { testName + "_RR",
+        1000 });
+      
+      vm4.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      vm5.invoke(WANTestBase.class, "validateQueueContents", new Object[] { "ln1",
+        0 });
+      
+      vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_RR", 1000 });
+    }
+    catch (Exception e) {
+      fail("Unexpected exception", e);
+    }
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_2NODES_1NODESDOWN_Validate_Receiver() throws Exception {
+
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+    vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+    vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+    vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2, true,
+        100, 10, false, false, null, true });
+    vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2, true,
+        100, 10, false, false, null, true });
+
+    vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+    vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+
+    vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+    vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+    vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+
+    vm4.invoke(WANTestBase.class, "waitForSenderRunningState",
+        new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "waitForSenderRunningState",
+        new Object[] { "ln" });
+
+    pauseWaitCriteria(60000);
+    
+/*    ExpectedException exp1 = addExpectedException(CacheClosedException.class
+        .getName());
+    try {*/
+      AsyncInvocation inv1 = vm4.invokeAsync(
+          ReplicatedRegion_ParallelWANPropogationDUnitTest.class, "doPuts0", new Object[] {
+              testName + "_RR", 1000 });
+      pause(1000);
+      AsyncInvocation inv2 = vm5.invokeAsync(WANTestBase.class, "killSender");
+      try {
+        inv1.join();
+        inv2.join();
+      }
+      catch (Exception e) {
+        fail("UnExpected Exception", e);
+      }
+/*    }
+    finally {
+      exp1.remove();
+    }*/
+
+    Integer size = (Integer)vm4.invoke(WANTestBase.class,
+        "getQueueContentSize", new Object[] { "ln" });
+    getLogWriter().info("The size of the queue is in vm4 " + size);
+    
+
+    vm4.invoke(WANTestBase.class,
+        "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+    
+    size = (Integer)vm4.invoke(WANTestBase.class,
+        "getQueueContentSize", new Object[] { "ln" });
+    getLogWriter().info("The size of the queue is in vm4 " + size);
+    
+    vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_RR", 1000 });
+  }
+  
+  /**Below test is disabled intentionally
+  1> In this release 8.0, for rolling upgrade support queue name is changed to old style
+  2>Comman parallel sender for different non colocated regions is not supported in 8.0 so no need to bother about 
+      ParallelGatewaySenderQueue#convertPathToName
+  3> We have to enabled it in next release
+  4> Version based rolling upgrade support should be provided. based on the version of the gemfire QSTRING should be used between 8.0 
+     and version prior to 8.0*/
+  public void DISABLED_test_DR_PGS_4NODES_2NODESDOWN_Validate_Receiver() throws Exception {
+
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+    vm3.invoke(WANTestBase.class, "createReceiver", new Object[] { nyPort });
+
+    vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm5.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm6.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+    vm7.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+    vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2, true,
+        100, 10, false, false, null, true });
+    vm5.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2, true,
+        100, 10, false, false, null, true });
+    vm6.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2, true,
+        100, 10, false, false, null, true });
+    vm7.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2, true,
+        100, 10, false, false, null, true });
+
+    vm4.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+    vm5.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+    vm6.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+    vm7.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+
+    vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm6.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+    vm7.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+    vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+    vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap()  });
+
+    vm4.invoke(WANTestBase.class, "waitForSenderRunningState",
+        new Object[] { "ln" });
+    vm5.invoke(WANTestBase.class, "waitForSenderRunningState",
+        new Object[] { "ln" });
+    vm6.invoke(WANTestBase.class, "waitForSenderRunningState",
+        new Object[] { "ln" });
+    vm7.invoke(WANTestBase.class, "waitForSenderRunningState",
+        new Object[] { "ln" });
+
+    pauseWaitCriteria(60000);
+/*    ExpectedException exp1 = addExpectedException(CacheClosedException.class
+        .getName());
+    try */{
+      AsyncInvocation inv1 = vm7.invokeAsync(
+          ReplicatedRegion_ParallelWANPropogationDUnitTest.class, "doPuts0", new Object[] {
+              testName + "_RR", 10000 });
+      pauseWaitCriteria(1000);
+      AsyncInvocation inv2 = vm4.invokeAsync(WANTestBase.class, "killSender");
+      pauseWaitCriteria(2000);
+      AsyncInvocation inv3 = vm6.invokeAsync(
+          ReplicatedRegion_ParallelWANPropogationDUnitTest.class, "doPuts1", new Object[] {
+              testName + "_RR", 10000 });
+      pauseWaitCriteria(1500);
+      AsyncInvocation inv4 = vm5.invokeAsync(WANTestBase.class, "killSender");
+      try {
+        inv1.join();
+        inv2.join();
+        inv3.join();
+        inv4.join();
+      }
+      catch (Exception e) {
+        fail("UnExpected Exception", e);
+      }
+    }/*
+    finally {
+      exp1.remove();
+    }*/
+
+    vm6.invoke(WANTestBase.class,
+        "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+    vm7.invoke(WANTestBase.class,
+        "validateParallelSenderQueueAllBucketsDrained", new Object[] { "ln" });
+
+    vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_RR", 10000 });
+    
+  }
+  
+  public static void doPuts0(String regionName, int numPuts) {
+    ExpectedException exp = addExpectedException(ForceReattemptException.class
+        .getName());
+    ExpectedException exp1 = addExpectedException(CacheClosedException.class
+        .getName());
+    try {
+
+      Region r = cache.getRegion(Region.SEPARATOR + regionName);
+      assertNotNull(r);
+      for (long i = 0; i < numPuts; i++) {
+        getLogWriter().info("Put : key : " + i);
+        r.put(i, "0_" + i);
+      }
+    } finally {
+      exp.remove();
+      exp1.remove();
+    }
+  }
+  
+  public static void doPuts1(String regionName, int numPuts){
+    ExpectedException exp = addExpectedException(ForceReattemptException.class
+        .getName());
+    ExpectedException exp1 = addExpectedException(CacheClosedException.class
+        .getName());
+    try {
+
+      Region r = cache.getRegion(Region.SEPARATOR + regionName);
+      assertNotNull(r);
+      for (long i = 0; i < numPuts; i++) {
+        getLogWriter().info("Put : key : " + i);
+        r.put(i, "1_" + i);
+      }
+    } finally {
+      exp.remove();
+      exp1.remove();
+    }
+  }
+  
+  public static void doPuts2(String regionName, int numPuts){
+    ExpectedException exp = addExpectedException(ForceReattemptException.class
+        .getName());
+    ExpectedException exp1 = addExpectedException(CacheClosedException.class
+        .getName());
+    try {
+      Region r = cache.getRegion(Region.SEPARATOR + regionName);
+      assertNotNull(r);
+      for (long i = 0; i < numPuts; i++) {
+        getLogWriter().info("Put : key : " + i);
+        r.put(i, "2_" + i);
+      }
+    } finally {
+      exp.remove();
+      exp1.remove();
+    }
+  }
+    
+  /**
+   * Test to validate that put on DR with no ack on multiple nodes are propogated to parallelqueue on multiple nodes 
+   */
+  
+  /**
+   * Test to validate that the single put in DR is propoagted to remote site through paralleHatewaySender
+   */
+  
+  
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/SenderWithTransportFilterDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/SenderWithTransportFilterDUnitTest.java b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/SenderWithTransportFilterDUnitTest.java
new file mode 100644
index 0000000..4e15ab7
--- /dev/null
+++ b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/SenderWithTransportFilterDUnitTest.java
@@ -0,0 +1,266 @@
+/*=========================================================================
+ * 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.internal.cache.wan.misc;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.Properties;
+import java.util.zip.Adler32;
+import java.util.zip.CheckedInputStream;
+import java.util.zip.CheckedOutputStream;
+
+import com.gemstone.gemfire.cache.CacheFactory;
+import com.gemstone.gemfire.cache.DiskStore;
+import com.gemstone.gemfire.cache.DiskStoreFactory;
+import com.gemstone.gemfire.cache.wan.GatewayReceiver;
+import com.gemstone.gemfire.cache.wan.GatewayReceiverFactory;
+import com.gemstone.gemfire.cache.wan.GatewaySenderFactory;
+import com.gemstone.gemfire.cache.wan.GatewayTransportFilter;
+import com.gemstone.gemfire.distributed.internal.DistributionConfig;
+import com.gemstone.gemfire.distributed.internal.InternalDistributedSystem;
+import com.gemstone.gemfire.internal.AvailablePortHelper;
+import com.gemstone.gemfire.internal.cache.wan.CompressionInputStream;
+import com.gemstone.gemfire.internal.cache.wan.CompressionOutputStream;
+import com.gemstone.gemfire.internal.cache.wan.InternalGatewaySenderFactory;
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase;
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase.MyLocatorCallback;
+
+import dunit.VM;
+
+public class SenderWithTransportFilterDUnitTest extends WANTestBase {
+
+  private static final long serialVersionUID = 1L;
+
+  public SenderWithTransportFilterDUnitTest(String name) {
+    super(name);
+  }
+
+  public void setUp() throws Exception {
+    super.setUp();
+  }
+
+  public void testSerialSenderWithTansportFilter() {
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(SenderWithTransportFilterDUnitTest.class,
+        "createReceiverWithTransportFilters", new Object[] { nyPort });
+    vm2.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", null, isOffHeap() });
+
+    vm3.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+    vm3.invoke(SenderWithTransportFilterDUnitTest.class,
+        "createSenderWithTransportFilter", new Object[] { "ln", 2, false, 100,
+            1, false, false, true });
+
+    vm3.invoke(WANTestBase.class, "createReplicatedRegion", new Object[] {
+        testName + "_RR", "ln", isOffHeap() });
+
+    vm3.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+    vm3.invoke(WANTestBase.class, "doPuts",
+        new Object[] { testName + "_RR", 100 });
+
+    vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_RR", 100 });
+  }
+
+  public void testParallelSenderWithTansportFilter() {
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(SenderWithTransportFilterDUnitTest.class,
+        "createReceiverWithTransportFilters", new Object[] { nyPort });
+    vm2.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", null, 0, 10, isOffHeap() });
+
+    vm3.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+    vm3.invoke(SenderWithTransportFilterDUnitTest.class,
+        "createSenderWithTransportFilter", new Object[] { "ln", 2, true, 100,
+            1, false, false, true });
+
+    vm3.invoke(WANTestBase.class, "createPartitionedRegion", new Object[] {
+        testName + "_PR", "ln", 0, 10, isOffHeap() });
+
+    vm3.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+    vm3.invoke(WANTestBase.class, "doPuts",
+        new Object[] { testName + "_PR", 100 });
+
+    vm2.invoke(WANTestBase.class, "validateRegionSize", new Object[] {
+        testName + "_PR", 100 });
+  }
+  
+  public static int createReceiverWithTransportFilters(int locPort) {
+    WANTestBase test = new WANTestBase(testName);
+    Properties props = new Properties();
+    props.setProperty(DistributionConfig.MCAST_PORT_NAME, "0");
+    props.setProperty(DistributionConfig.LOCATORS_NAME, "localhost[" + locPort
+        + "]");
+
+    InternalDistributedSystem ds = test.getSystem(props);
+    cache = CacheFactory.create(ds);
+    GatewayReceiverFactory fact = cache.createGatewayReceiverFactory();
+    int port = AvailablePortHelper.getRandomAvailablePortForDUnitSite();
+    fact.setStartPort(port);
+    fact.setEndPort(port);
+    ArrayList<GatewayTransportFilter> transportFilters = new ArrayList<GatewayTransportFilter>();
+    transportFilters.add(new CheckSumTranportFilter("CheckSumTranportFilter"));
+    transportFilters.add(new CompressionTranportFilter("CompressionTranportFilter"));
+    if (!transportFilters.isEmpty()) {
+      for (GatewayTransportFilter filter : transportFilters) {
+        fact.addGatewayTransportFilter(filter);
+      }
+    }
+    GatewayReceiver receiver = fact.create();
+    try {
+      receiver.start();
+    }
+    catch (IOException e) {
+      e.printStackTrace();
+      fail("Test " + test.getName()
+          + " failed to start GatewayRecevier on port " + port);
+    }
+    return port;
+  }
+
+  public static void createSenderWithTransportFilter(String dsName,
+      int remoteDsId, boolean isParallel, Integer maxMemory, Integer batchSize,
+      boolean isConflation, boolean isPersistent, boolean isManulaStart) {
+    File persistentDirectory = new File(dsName + "_disk_"
+        + System.currentTimeMillis() + "_" + VM.getCurrentVMNum());
+    persistentDirectory.mkdir();
+    DiskStoreFactory dsf = cache.createDiskStoreFactory();
+    File[] dirs1 = new File[] { persistentDirectory };
+
+    if (isParallel) {
+      GatewaySenderFactory gateway = cache
+          .createGatewaySenderFactory();
+      gateway.setParallel(true);
+      gateway.setMaximumQueueMemory(maxMemory);
+      gateway.setBatchSize(batchSize);
+      ((InternalGatewaySenderFactory)gateway).setLocatorDiscoveryCallback(new MyLocatorCallback());
+      ArrayList<GatewayTransportFilter> transportFilters = new ArrayList<GatewayTransportFilter>();
+      transportFilters.add(new CompressionTranportFilter("CompressionTranportFilter"));
+      transportFilters.add(new CheckSumTranportFilter("CheckSumTranportFilter"));
+      if (!transportFilters.isEmpty()) {
+        for (GatewayTransportFilter filter : transportFilters) {
+          gateway.addGatewayTransportFilter(filter);
+        }
+      }
+      if (isPersistent) {
+        gateway.setPersistenceEnabled(true);
+        gateway.setDiskStoreName(dsf.setDiskDirs(dirs1).create(dsName)
+            .getName());
+      }
+      else {
+        DiskStore store = dsf.setDiskDirs(dirs1).create(dsName);
+        gateway.setDiskStoreName(store.getName());
+      }
+      gateway.setBatchConflationEnabled(isConflation);
+      gateway.create(dsName, remoteDsId);
+
+    }
+    else {
+      GatewaySenderFactory gateway = cache
+          .createGatewaySenderFactory();
+      gateway.setMaximumQueueMemory(maxMemory);
+      gateway.setBatchSize(batchSize);
+      gateway.setManualStart(isManulaStart);
+      ((InternalGatewaySenderFactory)gateway)
+          .setLocatorDiscoveryCallback(new MyLocatorCallback());
+      ArrayList<GatewayTransportFilter> transportFilters = new ArrayList<GatewayTransportFilter>();
+      transportFilters.add(new CompressionTranportFilter("CompressionTranportFilter"));
+      transportFilters.add(new CheckSumTranportFilter("CheckSumTranportFilter"));
+      if (!transportFilters.isEmpty()) {
+        for (GatewayTransportFilter filter : transportFilters) {
+          gateway.addGatewayTransportFilter(filter);
+        }
+      }
+      gateway.setBatchConflationEnabled(isConflation);
+      if (isPersistent) {
+        gateway.setPersistenceEnabled(true);
+        gateway.setDiskStoreName(dsf.setDiskDirs(dirs1).create(dsName)
+            .getName());
+      }
+      else {
+        DiskStore store = dsf.setDiskDirs(dirs1).create(dsName);
+        gateway.setDiskStoreName(store.getName());
+      }
+      gateway.create(dsName, remoteDsId);
+    }
+  }
+
+  static class CompressionTranportFilter implements GatewayTransportFilter {
+
+    private String name;
+    
+    public CompressionTranportFilter(String name){
+      this.name = name;
+    }
+    
+    public String toString(){
+      return this.name;
+    }
+    public InputStream getInputStream(InputStream stream) {
+      return new CompressionInputStream(stream);
+      // return new ZipInputStream(stream);
+    }
+
+    public OutputStream getOutputStream(OutputStream stream) {
+      return new CompressionOutputStream(stream);
+      // return new ZipOutputStream(stream);
+    }
+
+    public void close() {
+      // TODO Auto-generated method stub
+    }
+
+  }
+  
+  static class CheckSumTranportFilter implements GatewayTransportFilter {
+
+    Adler32 checker = new Adler32();
+    
+    private String name;
+    
+    public CheckSumTranportFilter(String name){
+      this.name = name;
+    }
+    
+    public String toString(){
+      return this.name;
+    }
+    public InputStream getInputStream(InputStream stream) {
+      return new CheckedInputStream(stream, checker);
+      // return new ZipInputStream(stream);
+    }
+
+    public OutputStream getOutputStream(OutputStream stream) {
+      return new CheckedOutputStream(stream, checker);
+      // return new ZipOutputStream(stream);
+    }
+
+    public void close() {
+      // TODO Auto-generated method stub
+    }
+
+  }
+  
+}

http://git-wip-us.apache.org/repos/asf/incubator-geode/blob/6df75241/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/ShutdownAllPersistentGatewaySenderDUnitTest.java
----------------------------------------------------------------------
diff --git a/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/ShutdownAllPersistentGatewaySenderDUnitTest.java b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/ShutdownAllPersistentGatewaySenderDUnitTest.java
new file mode 100644
index 0000000..211f726
--- /dev/null
+++ b/gemfire-wan/src/test/java/com/gemstone/gemfire/internal/cache/wan/misc/ShutdownAllPersistentGatewaySenderDUnitTest.java
@@ -0,0 +1,197 @@
+/*=========================================================================
+ * 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.internal.cache.wan.misc;
+
+import java.util.Set;
+
+import com.gemstone.gemfire.admin.AdminDistributedSystemFactory;
+import com.gemstone.gemfire.admin.AdminException;
+import com.gemstone.gemfire.admin.DistributedSystemConfig;
+import com.gemstone.gemfire.admin.internal.AdminDistributedSystemImpl;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.internal.cache.CacheObserverAdapter;
+import com.gemstone.gemfire.internal.cache.CacheObserverHolder;
+import com.gemstone.gemfire.internal.cache.LocalRegion;
+import com.gemstone.gemfire.internal.cache.wan.WANTestBase;
+
+import dunit.AsyncInvocation;
+import dunit.DistributedTestCase;
+import dunit.SerializableRunnable;
+import dunit.VM;
+
+public class ShutdownAllPersistentGatewaySenderDUnitTest extends WANTestBase {
+  private static final long MAX_WAIT = 70000;
+
+  private static final int NUM_KEYS = 1000;
+
+  public ShutdownAllPersistentGatewaySenderDUnitTest(String name) {
+    super(name);
+  }
+  
+  @Override
+  public void setUp() throws Exception {
+    super.setUp();
+    addExpectedException("Cache is being closed by ShutdownAll");
+  }
+
+  private static final long serialVersionUID = 1L;
+
+  public void testGatewaySender() throws Exception {
+    addExpectedException("Cache is shutting down");
+
+    Integer lnPort = (Integer)vm0.invoke(WANTestBase.class,
+        "createFirstLocatorWithDSId", new Object[] { 1 });
+
+    Integer nyPort = (Integer)vm1.invoke(WANTestBase.class,
+        "createFirstRemoteLocator", new Object[] { 2, lnPort });
+
+    vm2.invoke(WANTestBase.class, "createCache", new Object[] { nyPort });
+    vm3.invoke(WANTestBase.class, "createCache", new Object[] { nyPort });
+    vm2.invoke(WANTestBase.class, "createReceiverAfterCache",
+        new Object[] { nyPort });
+
+    vm2.invoke(WANTestBase.class, "createPersistentPartitionedRegion",
+        new Object[] { testName + "_PR", "ln", 1, 100, isOffHeap() });
+    vm3.invoke(WANTestBase.class, "createPersistentPartitionedRegion",
+        new Object[] { testName + "_PR", "ln", 1, 100, isOffHeap() });
+
+    vm4.invoke(WANTestBase.class, "createCache", new Object[] { lnPort });
+
+    vm4.invoke(WANTestBase.class, "createSender", new Object[] { "ln", 2,
+        false, 100, 400, false, false, null, true });
+
+    vm4.invoke(WANTestBase.class, "startSender", new Object[] { "ln" });
+
+    vm4.invoke(WANTestBase.class, "createPersistentPartitionedRegion",
+        new Object[] { testName + "_PR", "ln", 1, 100, isOffHeap() });
+
+    // set the CacheObserver to block the ShutdownAll
+    SerializableRunnable waitAtShutdownAll = new SerializableRunnable() {
+      @Override
+      public void run() {
+        LocalRegion.ISSUE_CALLBACKS_TO_CACHE_OBSERVER = true;
+        CacheObserverHolder.setInstance(new CacheObserverAdapter() {
+          @Override
+          public void beforeShutdownAll() {
+            final Region region = cache.getRegion(testName + "_PR");
+            waitForCriterion(new WaitCriterion() {
+              @Override
+              public boolean done() {
+                return region.size() >= 2;
+              }
+
+              @Override
+              public String description() {
+                return "Wait for wan to have processed several events";
+              }
+            }, 30000, 100, true);
+          }
+        });
+      }
+    };
+    vm2.invoke(waitAtShutdownAll);
+    vm3.invoke(waitAtShutdownAll);
+    
+    AsyncInvocation vm4_future = vm4.invokeAsync(WANTestBase.class, "doPuts",
+        new Object[] { testName + "_PR", NUM_KEYS });
+
+    // ShutdownAll will be suspended at observer, so puts will continue
+    AsyncInvocation future = shutDownAllMembers(vm2, 2, MAX_WAIT);
+    future.join(MAX_WAIT);
+
+    // now restart vm1 with gatewayHub
+    getLogWriter().info("restart in VM2");
+    vm2.invoke(WANTestBase.class, "createCache", new Object[] { nyPort });
+    vm3.invoke(WANTestBase.class, "createCache", new Object[] { nyPort });
+    AsyncInvocation vm3_future = vm3.invokeAsync(WANTestBase.class,
+        "createPersistentPartitionedRegion", new Object[] { testName + "_PR",
+            "ln", 1, 100, isOffHeap() });
+    vm2.invoke(WANTestBase.class, "createPersistentPartitionedRegion",
+        new Object[] { testName + "_PR", "ln", 1, 100, isOffHeap() });
+    vm3_future.join(MAX_WAIT);
+
+    vm3.invoke(new SerializableRunnable() {
+      public void run() {
+        final Region region = cache.getRegion(testName + "_PR");
+        cache.getLogger().info(
+            "vm1's region size before restart gatewayhub is " + region.size());
+      }
+    });
+    vm2.invoke(WANTestBase.class, "createReceiverAfterCache",
+        new Object[] { nyPort });
+
+    // wait for vm0 to finish its work
+    vm4_future.join(MAX_WAIT);
+    vm4.invoke(new SerializableRunnable() {
+      public void run() {
+        Region region = cache.getRegion(testName + "_PR");
+        assertEquals(NUM_KEYS, region.size());
+      }
+    });
+
+    // verify the other side (vm1)'s entries received from gateway
+    vm2.invoke(new SerializableRunnable() {
+      public void run() {
+        final Region region = cache.getRegion(testName + "_PR");
+
+        cache.getLogger().info(
+            "vm1's region size after restart gatewayhub is " + region.size());
+        waitForCriterion(new WaitCriterion() {
+          public boolean done() {
+            Object lastvalue = region.get(NUM_KEYS - 1);
+            if (lastvalue != null && lastvalue.equals(NUM_KEYS - 1)) {
+              region.getCache().getLogger().info(
+                  "Last key has arrived, its value is " + lastvalue
+                      + ", end of wait.");
+              return true;
+            }
+            else
+              return (region.size() == NUM_KEYS);
+          }
+
+          public String description() {
+            return "Waiting for destination region to reach size: " + NUM_KEYS
+                + ", current is " + region.size();
+          }
+        }, MAX_WAIT, 100, true);
+        assertEquals(NUM_KEYS, region.size());
+      }
+    });
+
+  }
+
+  private AsyncInvocation shutDownAllMembers(VM vm, final int expnum, final long timeout) {
+      AsyncInvocation future = vm.invokeAsync(new SerializableRunnable("Shutdown all the members") {
+
+      public void run() {
+        DistributedSystemConfig config;
+        AdminDistributedSystemImpl adminDS = null;
+        try {
+          config = AdminDistributedSystemFactory.defineDistributedSystem(cache
+              .getDistributedSystem(), "");
+          adminDS = (AdminDistributedSystemImpl)AdminDistributedSystemFactory
+              .getDistributedSystem(config);
+          adminDS.connect();
+          Set members = adminDS.shutDownAllMembers(timeout);
+          int num = members == null ? 0 : members.size();
+          assertEquals(expnum, num);
+        }
+        catch (AdminException e) {
+          throw new RuntimeException(e);
+        }
+        finally {
+          if (adminDS != null) {
+            adminDS.disconnect();
+          }
+        }
+      }
+    });
+    return future;
+  }
+
+}


Mime
View raw message