geode-cq/src/distributedTest/java/org/apache/geode/internal/cache/ha/HADispatcherDUnitTest.java - geode - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
  * agreements. See the NOTICE file distributed with this work for additional information regarding
  * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance with the License. You may obtain a
  * copy of the License at
  *
  * http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software distributed under the License
  * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  * or implied. See the License for the specific language governing permissions and limitations under
  * the License.
  */
 package org.apache.geode.internal.cache.ha;

 import static org.apache.geode.cache.Region.SEPARATOR;
 import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
 import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
 import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
 import static org.apache.geode.test.dunit.Assert.assertEquals;
 import static org.apache.geode.test.dunit.Assert.assertNotNull;
 import static org.apache.geode.test.dunit.Assert.assertTrue;
 import static org.apache.geode.test.dunit.Assert.fail;
 import static org.apache.geode.test.dunit.Host.getHost;
 import static org.apache.geode.test.dunit.LogWriterUtils.getLogWriter;
 import static org.apache.geode.test.dunit.NetworkUtils.getServerHostName;

 import java.io.IOException;
 import java.util.Iterator;
 import java.util.Properties;

 import org.junit.Test;
 import org.junit.experimental.categories.Category;

 import org.apache.geode.cache.AttributesFactory;
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.CacheException;
 import org.apache.geode.cache.CacheFactory;
 import org.apache.geode.cache.CacheListener;
 import org.apache.geode.cache.DataPolicy;
 import org.apache.geode.cache.Declarable;
 import org.apache.geode.cache.EntryEvent;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionAttributes;
 import org.apache.geode.cache.Scope;
 import org.apache.geode.cache.client.internal.PoolImpl;
 import org.apache.geode.cache.query.CqAttributes;
 import org.apache.geode.cache.query.CqAttributesFactory;
 import org.apache.geode.cache.query.CqException;
 import org.apache.geode.cache.query.CqExistsException;
 import org.apache.geode.cache.query.CqListener;
 import org.apache.geode.cache.query.CqQuery;
 import org.apache.geode.cache.query.QueryService;
 import org.apache.geode.cache.query.RegionNotFoundException;
 import org.apache.geode.cache.query.cq.dunit.CqQueryTestListener;
 import org.apache.geode.cache.util.CacheListenerAdapter;
 import org.apache.geode.cache30.CacheSerializableRunnable;
 import org.apache.geode.cache30.ClientServerTestCase;
 import org.apache.geode.distributed.DistributedSystem;
 import org.apache.geode.internal.cache.CacheServerImpl;
 import org.apache.geode.internal.cache.Conflatable;
 import org.apache.geode.internal.cache.HARegion;
 import org.apache.geode.internal.cache.LocalRegion;
 import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
 import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
 import org.apache.geode.internal.cache.tier.sockets.ConflationDUnitTestHelper;
 import org.apache.geode.internal.cache.tier.sockets.HAEventWrapper;
 import org.apache.geode.test.awaitility.GeodeAwaitility;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.WaitCriterion;
 import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
 import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

 /**
  * This Dunit test is to verify that when the dispatcher of CS dispatches the Event , the peer's
  * HARegionQueue should get the events removed from the HA RegionQueues assuming the QRM thread has
  * acted upon by that time This is done in the following steps 1. start server1 and server2 2. start
  * client1 and client2 3. perform put operation from client1 4. check the entry in the regionque of
  * client2 on server2.It should be present. 5. Wait till client2 receives event 6. Make sure that
  * QRM is envoked 7. Again the entry in the regionque of client2 on server2.It should not be
  * present. 8. close client1 and client2 9. close server1 and server2
  */
 @Category({ClientSubscriptionTest.class})
 public class HADispatcherDUnitTest extends JUnit4DistributedTestCase {

   private static final String REGION_NAME = HADispatcherDUnitTest.class.getSimpleName() + "_region";
   private static final Object dummyObj = "dummyObject";
   private static final String KEY1 = "KEY1";
   private static final String VALUE1 = "VALUE1";
   private static final String KEY2 = "KEY2";
   private static final String VALUE2 = "VALUE2";

   private static Cache cache = null;
   private static volatile boolean isObjectPresent = false;
   private static volatile boolean waitFlag = true;

   private VM server1 = null;
   private VM server2 = null;
   private VM client1 = null;
   private VM client2 = null;
   private int PORT1;
   private int PORT2;

   @Override
   public final void postSetUp() throws Exception {
     String serverHostName = getServerHostName(getHost(0));

     // Server1 VM
     server1 = getHost(0).getVM(0);

     // Server2 VM
     server2 = getHost(0).getVM(1);

     // Client 1 VM
     client1 = getHost(0).getVM(2);

     // client 2 VM
     client2 = getHost(0).getVM(3);

     PORT1 = ((Integer) server1.invoke(() -> createServerCache(new Boolean(false)))).intValue();

     server1.invoke(() -> ConflationDUnitTestHelper.setIsSlowStart());
     server1.invoke(() -> makeDispatcherSlow());
     server1.invoke(() -> setQRMslow());

     PORT2 = ((Integer) server2.invoke(() -> createServerCache(new Boolean(true)))).intValue();

     client1.invoke(() -> CacheServerTestUtil.disableShufflingOfEndpoints());
     client2.invoke(() -> CacheServerTestUtil.disableShufflingOfEndpoints());
     client1.invoke(() -> createClientCache(serverHostName, new Integer(PORT1), new Integer(PORT2),
         new Boolean(false)));
     client2.invoke(() -> createClientCache(serverHostName, new Integer(PORT1), new Integer(PORT2),
         new Boolean(true)));
   }

   @Override
   public final void preTearDown() throws Exception {
     client1.invoke(() -> closeCache());
     client2.invoke(() -> closeCache());
     // close server
     server1.invoke(() -> resetQRMslow());
     server1.invoke(() -> closeCache());
     server2.invoke(() -> closeCache());
   }

   @Test
   public void testDispatcher() throws Exception {
     clientPut(client1, KEY1, VALUE1);
     // Waiting in the client2 till it receives the event for the key.
     checkFromClient(client2);

     // performing check in the regionqueue of the server2
     checkFromServer(server2, KEY1);

     // For CQ Only.
     // performing put from the client1
     clientPut(client1, KEY2, VALUE2);
     checkFromClient(client2);
     checkFromServer(server2, KEY2);

     getLogWriter().info("testDispatcher() completed successfully");
   }

   /**
    * This is to test the serialization mechanism of ClientUpdateMessage. Added after CQ support.
    * This could be done in different way, by overflowing the HARegion queue.
    */
   @Test
   public void testClientUpdateMessageSerialization() throws Exception {
     // Update Value.
     clientPut(client1, KEY1, VALUE1);
     getLogWriter().fine(">>>>>>>> after clientPut(c1, k1, v1)");
     // Waiting in the client2 till it receives the event for the key.
     checkFromClient(client2);
     getLogWriter().fine("after checkFromClient(c2)");

     // performing check in the regionqueue of the server2
     checkFromServer(server2, KEY1);
     getLogWriter().fine("after checkFromServer(s2, k1)");

     // UPDATE.
     clientPut(client1, KEY1, VALUE1);
     getLogWriter().fine("after clientPut 2 (c1, k1, v1)");
     // Waiting in the client2 till it receives the event for the key.
     checkFromClient(client2);
     getLogWriter().fine("after checkFromClient 2 (c2)");

     // performing check in the regionqueue of the server2
     checkFromServer(server2, KEY1);
     getLogWriter().fine("after checkFromServer 2 (s2, k1)");

     getLogWriter().info("testClientUpdateMessageSerialization() completed successfully");
   }

   private void closeCache() {
     if (cache != null && !cache.isClosed()) {
       cache.close();
       cache.getDistributedSystem().disconnect();
     }
   }

   private void setQRMslow() throws InterruptedException {
     int oldMessageSyncInterval = cache.getMessageSyncInterval();
     cache.setMessageSyncInterval(6);
     Thread.sleep((oldMessageSyncInterval + 1) * 1000);
   }

   private void resetQRMslow() {
     cache.setMessageSyncInterval(HARegionQueue.DEFAULT_MESSAGE_SYNC_INTERVAL);
   }

   private void makeDispatcherSlow() {
     System.setProperty("slowStartTimeForTesting", "5000");
   }

   private void clientPut(VM vm, final Object key, final Object value) {
     // performing put from the client1
     vm.invoke(new CacheSerializableRunnable("putFromClient") {
       @Override
       public void run2() throws CacheException {
         Region region = cache.getRegion(SEPARATOR + REGION_NAME);
         assertNotNull(region);
         region.put(key, value);
       }
     });
   }

   private void checkFromClient(VM vm) {
     // Waiting in the client till it receives the event for the key.
     vm.invoke(new CacheSerializableRunnable("checkFromClient") {
       @Override
       public void run2() throws CacheException {
         Region region = cache.getRegion(SEPARATOR + REGION_NAME);
         assertNotNull(region);
         cache.getLogger().fine("starting the wait");
         synchronized (dummyObj) {
           while (waitFlag) {
             try {
               dummyObj.wait(30000);
             } catch (InterruptedException e) {
               fail("interrupted", e);
             }
           }
         }
         cache.getLogger().fine("wait over...waitFlag=" + waitFlag);
         if (waitFlag)
           fail("test failed");
       }
     });
   }

   private void checkFromServer(VM vm, final Object key) {
     vm.invoke(new CacheSerializableRunnable("checkFromServer") {
       @Override
       public void run2() throws CacheException {
         Iterator iter = cache.getCacheServers().iterator();
         CacheServerImpl server = (CacheServerImpl) iter.next();
         Iterator iter_prox =
             server.getAcceptor().getCacheClientNotifier().getClientProxies().iterator();
         isObjectPresent = false;

         while (iter_prox.hasNext()) {
           final CacheClientProxy proxy = (CacheClientProxy) iter_prox.next();
           HARegion region = (HARegion) proxy.getHARegion();
           assertNotNull(region);
           final HARegionQueue regionQueue = region.getOwner();

           WaitCriterion wc = new WaitCriterion() {
             @Override
             public boolean done() {
               int sz = regionQueue.size();
               cache.getLogger().fine("regionQueue.size()::" + sz);
               return sz == 0 || !proxy.isConnected();
             }

             @Override
             public String description() {
               return "regionQueue not empty with size " + regionQueue.size() + " for proxy "
                   + proxy;
             }
           };
           GeodeAwaitility.await().untilAsserted(wc);

           cache.getLogger().fine("processed a proxy");
         }
       }
     });
   }

   private void createCache(Properties props) {
     DistributedSystem ds = getSystem(props);
     assertNotNull(ds);
     ds.disconnect();
     ds = getSystem(props);
     cache = CacheFactory.create(ds);
     assertNotNull(cache);
   }

   private Integer createServerCache(Boolean isListenerPresent) throws IOException {
     createCache(new Properties());
     AttributesFactory factory = new AttributesFactory();
     factory.setScope(Scope.DISTRIBUTED_ACK);
     factory.setDataPolicy(DataPolicy.REPLICATE);
     if (isListenerPresent.booleanValue() == true) {
       CacheListener serverListener = new HAServerListener();
       factory.setCacheListener(serverListener);
     }
     RegionAttributes attrs = factory.create();
     cache.createRegion(REGION_NAME, attrs);
     CacheServerImpl server = (CacheServerImpl) cache.addCacheServer();
     assertNotNull(server);
     int port = getRandomAvailableTCPPort();
     server.setPort(port);
     server.setNotifyBySubscription(true);
     server.start();
     return new Integer(server.getPort());
   }

   private void createClientCache(String hostName, Integer port1, Integer port2,
       Boolean isListenerPresent) throws CqException, CqExistsException, RegionNotFoundException {
     int PORT1 = port1.intValue();
     int PORT2 = port2.intValue();
     Properties props = new Properties();
     props.setProperty(MCAST_PORT, "0");
     props.setProperty(LOCATORS, "");
     createCache(props);
     AttributesFactory factory = new AttributesFactory();
     factory.setScope(Scope.DISTRIBUTED_ACK);
     ClientServerTestCase.configureConnectionPool(factory, hostName, new int[] {PORT1, PORT2}, true,
         -1, 2, null);
     if (isListenerPresent.booleanValue() == true) {
       CacheListener clientListener = new HAClientListener();
       factory.setCacheListener(clientListener);
     }
     RegionAttributes attrs = factory.create();
     cache.createRegion(REGION_NAME, attrs);
     Region region = cache.getRegion(SEPARATOR + REGION_NAME);
     assertNotNull(region);

     {
       LocalRegion lr = (LocalRegion) region;
       final PoolImpl pool = (PoolImpl) (lr.getServerProxy().getPool());
       WaitCriterion ev = new WaitCriterion() {
         @Override
         public boolean done() {
           return pool.getPrimary() != null;
         }

         @Override
         public String description() {
           return null;
         }
       };
       GeodeAwaitility.await().untilAsserted(ev);
       ev = new WaitCriterion() {
         @Override
         public boolean done() {
           return pool.getRedundants().size() >= 1;
         }

         @Override
         public String description() {
           return null;
         }
       };
       GeodeAwaitility.await().untilAsserted(ev);

       assertNotNull(pool.getPrimary());
       assertTrue("backups=" + pool.getRedundants() + " expected=" + 1,
           pool.getRedundants().size() >= 1);
       assertEquals(PORT1, pool.getPrimaryPort());
     }

     region.registerInterest(KEY1);

     // Register CQ.
     createCQ();
   }

   private void createCQ() throws CqException, CqExistsException, RegionNotFoundException {
     QueryService cqService = null;
     try {
       cqService = cache.getQueryService();
     } catch (Exception cqe) {
       cqe.printStackTrace();
       fail("Failed to getCQService.");
     }

     // Create CQ Attributes.
     CqAttributesFactory cqf = new CqAttributesFactory();
     CqListener[] cqListeners = {new CqQueryTestListener(getLogWriter())};
     cqf.initCqListeners(cqListeners);
     CqAttributes cqa = cqf.create();

     String cqName = "CQForHARegionQueueTest";
     String queryStr = "Select * from " + SEPARATOR + REGION_NAME;

     // Create CQ.
     CqQuery cq1 = cqService.newCq(cqName, queryStr, cqa);
     cq1.execute();
   }

   /**
    * This is the client listener which notifies the waiting thread when it receives the event.
    */
   private static class HAClientListener extends CacheListenerAdapter implements Declarable {

     @Override
     public void afterCreate(EntryEvent event) {
       synchronized (dummyObj) {
         try {
           Object value = event.getNewValue();
           if (value.equals(VALUE1)) {
             waitFlag = false;
             dummyObj.notifyAll();
           }
         } catch (Exception e) {
           e.printStackTrace();
         }
       }
     }

     @Override
     public void init(Properties props) {}
   }

   /**
    * This is the server listener which ensures that regionqueue is properly populated
    */
   private static class HAServerListener extends CacheListenerAdapter {
     @Override
     public void afterCreate(EntryEvent event) {
       Cache cache = event.getRegion().getCache();
       Iterator iter = cache.getCacheServers().iterator();
       CacheServerImpl server = (CacheServerImpl) iter.next();
       isObjectPresent = false;

       // The event not be there in the region first time; try couple of time.
       // This should have been replaced by listener on the HARegion and doing wait for event arrival
       // in that.
       while (true) {
         for (Iterator iter_prox =
             server.getAcceptor().getCacheClientNotifier().getClientProxies().iterator(); iter_prox
                 .hasNext();) {
           CacheClientProxy proxy = (CacheClientProxy) iter_prox.next();
           HARegion regionForQueue = (HARegion) proxy.getHARegion();

           for (Iterator itr = regionForQueue.values().iterator(); itr.hasNext();) {
             Object obj = itr.next();
             if (obj instanceof HAEventWrapper) {
               Conflatable confObj = (Conflatable) obj;
               if (KEY1.equals(confObj.getKeyToConflate())
                   || KEY2.equals(confObj.getKeyToConflate())) {
                 isObjectPresent = true;
               }
             }
           }
         }

         if (isObjectPresent) {
           break; // From while.
         }

         try {
           Thread.sleep(10);
         } catch (InterruptedException e) {
           fail("interrupted", e);
         }
       }
     }
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more contributor license
	* agreements. See the NOTICE file distributed with this work for additional information regarding
	* copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance with the License. You may obtain a
	* copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software distributed under the License
	* is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
	* or implied. See the License for the specific language governing permissions and limitations under
	* the License.
	*/
	package org.apache.geode.internal.cache.ha;

	import static org.apache.geode.cache.Region.SEPARATOR;
	import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
	import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT;
	import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
	import static org.apache.geode.test.dunit.Assert.assertEquals;
	import static org.apache.geode.test.dunit.Assert.assertNotNull;
	import static org.apache.geode.test.dunit.Assert.assertTrue;
	import static org.apache.geode.test.dunit.Assert.fail;
	import static org.apache.geode.test.dunit.Host.getHost;
	import static org.apache.geode.test.dunit.LogWriterUtils.getLogWriter;
	import static org.apache.geode.test.dunit.NetworkUtils.getServerHostName;

	import java.io.IOException;
	import java.util.Iterator;
	import java.util.Properties;

	import org.junit.Test;
	import org.junit.experimental.categories.Category;

	import org.apache.geode.cache.AttributesFactory;
	import org.apache.geode.cache.Cache;
	import org.apache.geode.cache.CacheException;
	import org.apache.geode.cache.CacheFactory;
	import org.apache.geode.cache.CacheListener;
	import org.apache.geode.cache.DataPolicy;
	import org.apache.geode.cache.Declarable;
	import org.apache.geode.cache.EntryEvent;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionAttributes;
	import org.apache.geode.cache.Scope;
	import org.apache.geode.cache.client.internal.PoolImpl;
	import org.apache.geode.cache.query.CqAttributes;
	import org.apache.geode.cache.query.CqAttributesFactory;
	import org.apache.geode.cache.query.CqException;
	import org.apache.geode.cache.query.CqExistsException;
	import org.apache.geode.cache.query.CqListener;
	import org.apache.geode.cache.query.CqQuery;
	import org.apache.geode.cache.query.QueryService;
	import org.apache.geode.cache.query.RegionNotFoundException;
	import org.apache.geode.cache.query.cq.dunit.CqQueryTestListener;
	import org.apache.geode.cache.util.CacheListenerAdapter;
	import org.apache.geode.cache30.CacheSerializableRunnable;
	import org.apache.geode.cache30.ClientServerTestCase;
	import org.apache.geode.distributed.DistributedSystem;
	import org.apache.geode.internal.cache.CacheServerImpl;
	import org.apache.geode.internal.cache.Conflatable;
	import org.apache.geode.internal.cache.HARegion;
	import org.apache.geode.internal.cache.LocalRegion;
	import org.apache.geode.internal.cache.tier.sockets.CacheClientProxy;
	import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
	import org.apache.geode.internal.cache.tier.sockets.ConflationDUnitTestHelper;
	import org.apache.geode.internal.cache.tier.sockets.HAEventWrapper;
	import org.apache.geode.test.awaitility.GeodeAwaitility;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.WaitCriterion;
	import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
	import org.apache.geode.test.junit.categories.ClientSubscriptionTest;

	/**
	* This Dunit test is to verify that when the dispatcher of CS dispatches the Event , the peer's
	* HARegionQueue should get the events removed from the HA RegionQueues assuming the QRM thread has
	* acted upon by that time This is done in the following steps 1. start server1 and server2 2. start
	* client1 and client2 3. perform put operation from client1 4. check the entry in the regionque of
	* client2 on server2.It should be present. 5. Wait till client2 receives event 6. Make sure that
	* QRM is envoked 7. Again the entry in the regionque of client2 on server2.It should not be
	* present. 8. close client1 and client2 9. close server1 and server2
	*/
	@Category({ClientSubscriptionTest.class})
	public class HADispatcherDUnitTest extends JUnit4DistributedTestCase {

	private static final String REGION_NAME = HADispatcherDUnitTest.class.getSimpleName() + "_region";
	private static final Object dummyObj = "dummyObject";
	private static final String KEY1 = "KEY1";
	private static final String VALUE1 = "VALUE1";
	private static final String KEY2 = "KEY2";
	private static final String VALUE2 = "VALUE2";

	private static Cache cache = null;
	private static volatile boolean isObjectPresent = false;
	private static volatile boolean waitFlag = true;

	private VM server1 = null;
	private VM server2 = null;
	private VM client1 = null;
	private VM client2 = null;
	private int PORT1;
	private int PORT2;

	@Override
	public final void postSetUp() throws Exception {
	String serverHostName = getServerHostName(getHost(0));

	// Server1 VM
	server1 = getHost(0).getVM(0);

	// Server2 VM
	server2 = getHost(0).getVM(1);

	// Client 1 VM
	client1 = getHost(0).getVM(2);

	// client 2 VM
	client2 = getHost(0).getVM(3);

	PORT1 = ((Integer) server1.invoke(() -> createServerCache(new Boolean(false)))).intValue();

	server1.invoke(() -> ConflationDUnitTestHelper.setIsSlowStart());
	server1.invoke(() -> makeDispatcherSlow());
	server1.invoke(() -> setQRMslow());

	PORT2 = ((Integer) server2.invoke(() -> createServerCache(new Boolean(true)))).intValue();

	client1.invoke(() -> CacheServerTestUtil.disableShufflingOfEndpoints());
	client2.invoke(() -> CacheServerTestUtil.disableShufflingOfEndpoints());
	client1.invoke(() -> createClientCache(serverHostName, new Integer(PORT1), new Integer(PORT2),
	new Boolean(false)));
	client2.invoke(() -> createClientCache(serverHostName, new Integer(PORT1), new Integer(PORT2),
	new Boolean(true)));
	}

	@Override
	public final void preTearDown() throws Exception {
	client1.invoke(() -> closeCache());
	client2.invoke(() -> closeCache());
	// close server
	server1.invoke(() -> resetQRMslow());
	server1.invoke(() -> closeCache());
	server2.invoke(() -> closeCache());
	}

	@Test
	public void testDispatcher() throws Exception {
	clientPut(client1, KEY1, VALUE1);
	// Waiting in the client2 till it receives the event for the key.
	checkFromClient(client2);

	// performing check in the regionqueue of the server2
	checkFromServer(server2, KEY1);

	// For CQ Only.
	// performing put from the client1
	clientPut(client1, KEY2, VALUE2);
	checkFromClient(client2);
	checkFromServer(server2, KEY2);

	getLogWriter().info("testDispatcher() completed successfully");
	}

	/**
	* This is to test the serialization mechanism of ClientUpdateMessage. Added after CQ support.
	* This could be done in different way, by overflowing the HARegion queue.
	*/
	@Test
	public void testClientUpdateMessageSerialization() throws Exception {
	// Update Value.
	clientPut(client1, KEY1, VALUE1);
	getLogWriter().fine(">>>>>>>> after clientPut(c1, k1, v1)");
	// Waiting in the client2 till it receives the event for the key.
	checkFromClient(client2);
	getLogWriter().fine("after checkFromClient(c2)");

	// performing check in the regionqueue of the server2
	checkFromServer(server2, KEY1);
	getLogWriter().fine("after checkFromServer(s2, k1)");

	// UPDATE.
	clientPut(client1, KEY1, VALUE1);
	getLogWriter().fine("after clientPut 2 (c1, k1, v1)");
	// Waiting in the client2 till it receives the event for the key.
	checkFromClient(client2);
	getLogWriter().fine("after checkFromClient 2 (c2)");

	// performing check in the regionqueue of the server2
	checkFromServer(server2, KEY1);
	getLogWriter().fine("after checkFromServer 2 (s2, k1)");

	getLogWriter().info("testClientUpdateMessageSerialization() completed successfully");
	}

	private void closeCache() {
	if (cache != null && !cache.isClosed()) {
	cache.close();
	cache.getDistributedSystem().disconnect();
	}
	}

	private void setQRMslow() throws InterruptedException {
	int oldMessageSyncInterval = cache.getMessageSyncInterval();
	cache.setMessageSyncInterval(6);
	Thread.sleep((oldMessageSyncInterval + 1) * 1000);
	}

	private void resetQRMslow() {
	cache.setMessageSyncInterval(HARegionQueue.DEFAULT_MESSAGE_SYNC_INTERVAL);
	}

	private void makeDispatcherSlow() {
	System.setProperty("slowStartTimeForTesting", "5000");
	}

	private void clientPut(VM vm, final Object key, final Object value) {
	// performing put from the client1
	vm.invoke(new CacheSerializableRunnable("putFromClient") {
	@Override
	public void run2() throws CacheException {
	Region region = cache.getRegion(SEPARATOR + REGION_NAME);
	assertNotNull(region);
	region.put(key, value);
	}
	});
	}

	private void checkFromClient(VM vm) {
	// Waiting in the client till it receives the event for the key.
	vm.invoke(new CacheSerializableRunnable("checkFromClient") {
	@Override
	public void run2() throws CacheException {
	Region region = cache.getRegion(SEPARATOR + REGION_NAME);
	assertNotNull(region);
	cache.getLogger().fine("starting the wait");
	synchronized (dummyObj) {
	while (waitFlag) {
	try {
	dummyObj.wait(30000);
	} catch (InterruptedException e) {
	fail("interrupted", e);
	}
	}
	}
	cache.getLogger().fine("wait over...waitFlag=" + waitFlag);
	if (waitFlag)
	fail("test failed");
	}
	});
	}

	private void checkFromServer(VM vm, final Object key) {
	vm.invoke(new CacheSerializableRunnable("checkFromServer") {
	@Override
	public void run2() throws CacheException {
	Iterator iter = cache.getCacheServers().iterator();
	CacheServerImpl server = (CacheServerImpl) iter.next();
	Iterator iter_prox =
	server.getAcceptor().getCacheClientNotifier().getClientProxies().iterator();
	isObjectPresent = false;

	while (iter_prox.hasNext()) {
	final CacheClientProxy proxy = (CacheClientProxy) iter_prox.next();
	HARegion region = (HARegion) proxy.getHARegion();
	assertNotNull(region);
	final HARegionQueue regionQueue = region.getOwner();

	WaitCriterion wc = new WaitCriterion() {
	@Override
	public boolean done() {
	int sz = regionQueue.size();
	cache.getLogger().fine("regionQueue.size()::" + sz);
	return sz == 0 \|\| !proxy.isConnected();
	}

	@Override
	public String description() {
	return "regionQueue not empty with size " + regionQueue.size() + " for proxy "
	+ proxy;
	}
	};
	GeodeAwaitility.await().untilAsserted(wc);

	cache.getLogger().fine("processed a proxy");
	}
	}
	});
	}

	private void createCache(Properties props) {
	DistributedSystem ds = getSystem(props);
	assertNotNull(ds);
	ds.disconnect();
	ds = getSystem(props);
	cache = CacheFactory.create(ds);
	assertNotNull(cache);
	}

	private Integer createServerCache(Boolean isListenerPresent) throws IOException {
	createCache(new Properties());
	AttributesFactory factory = new AttributesFactory();
	factory.setScope(Scope.DISTRIBUTED_ACK);
	factory.setDataPolicy(DataPolicy.REPLICATE);
	if (isListenerPresent.booleanValue() == true) {
	CacheListener serverListener = new HAServerListener();
	factory.setCacheListener(serverListener);
	}
	RegionAttributes attrs = factory.create();
	cache.createRegion(REGION_NAME, attrs);
	CacheServerImpl server = (CacheServerImpl) cache.addCacheServer();
	assertNotNull(server);
	int port = getRandomAvailableTCPPort();
	server.setPort(port);
	server.setNotifyBySubscription(true);
	server.start();
	return new Integer(server.getPort());
	}

	private void createClientCache(String hostName, Integer port1, Integer port2,
	Boolean isListenerPresent) throws CqException, CqExistsException, RegionNotFoundException {
	int PORT1 = port1.intValue();
	int PORT2 = port2.intValue();
	Properties props = new Properties();
	props.setProperty(MCAST_PORT, "0");
	props.setProperty(LOCATORS, "");
	createCache(props);
	AttributesFactory factory = new AttributesFactory();
	factory.setScope(Scope.DISTRIBUTED_ACK);
	ClientServerTestCase.configureConnectionPool(factory, hostName, new int[] {PORT1, PORT2}, true,
	-1, 2, null);
	if (isListenerPresent.booleanValue() == true) {
	CacheListener clientListener = new HAClientListener();
	factory.setCacheListener(clientListener);
	}
	RegionAttributes attrs = factory.create();
	cache.createRegion(REGION_NAME, attrs);
	Region region = cache.getRegion(SEPARATOR + REGION_NAME);
	assertNotNull(region);

	{
	LocalRegion lr = (LocalRegion) region;
	final PoolImpl pool = (PoolImpl) (lr.getServerProxy().getPool());
	WaitCriterion ev = new WaitCriterion() {
	@Override
	public boolean done() {
	return pool.getPrimary() != null;
	}

	@Override
	public String description() {
	return null;
	}
	};
	GeodeAwaitility.await().untilAsserted(ev);
	ev = new WaitCriterion() {
	@Override
	public boolean done() {
	return pool.getRedundants().size() >= 1;
	}

	@Override
	public String description() {
	return null;
	}
	};
	GeodeAwaitility.await().untilAsserted(ev);

	assertNotNull(pool.getPrimary());
	assertTrue("backups=" + pool.getRedundants() + " expected=" + 1,
	pool.getRedundants().size() >= 1);
	assertEquals(PORT1, pool.getPrimaryPort());
	}

	region.registerInterest(KEY1);

	// Register CQ.
	createCQ();
	}

	private void createCQ() throws CqException, CqExistsException, RegionNotFoundException {
	QueryService cqService = null;
	try {
	cqService = cache.getQueryService();
	} catch (Exception cqe) {
	cqe.printStackTrace();
	fail("Failed to getCQService.");
	}

	// Create CQ Attributes.
	CqAttributesFactory cqf = new CqAttributesFactory();
	CqListener[] cqListeners = {new CqQueryTestListener(getLogWriter())};
	cqf.initCqListeners(cqListeners);
	CqAttributes cqa = cqf.create();

	String cqName = "CQForHARegionQueueTest";
	String queryStr = "Select * from " + SEPARATOR + REGION_NAME;

	// Create CQ.
	CqQuery cq1 = cqService.newCq(cqName, queryStr, cqa);
	cq1.execute();
	}

	/**
	* This is the client listener which notifies the waiting thread when it receives the event.
	*/
	private static class HAClientListener extends CacheListenerAdapter implements Declarable {

	@Override
	public void afterCreate(EntryEvent event) {
	synchronized (dummyObj) {
	try {
	Object value = event.getNewValue();
	if (value.equals(VALUE1)) {
	waitFlag = false;
	dummyObj.notifyAll();
	}
	} catch (Exception e) {
	e.printStackTrace();
	}
	}
	}

	@Override
	public void init(Properties props) {}
	}

	/**
	* This is the server listener which ensures that regionqueue is properly populated
	*/
	private static class HAServerListener extends CacheListenerAdapter {
	@Override
	public void afterCreate(EntryEvent event) {
	Cache cache = event.getRegion().getCache();
	Iterator iter = cache.getCacheServers().iterator();
	CacheServerImpl server = (CacheServerImpl) iter.next();
	isObjectPresent = false;

	// The event not be there in the region first time; try couple of time.
	// This should have been replaced by listener on the HARegion and doing wait for event arrival
	// in that.
	while (true) {
	for (Iterator iter_prox =
	server.getAcceptor().getCacheClientNotifier().getClientProxies().iterator(); iter_prox
	.hasNext();) {
	CacheClientProxy proxy = (CacheClientProxy) iter_prox.next();
	HARegion regionForQueue = (HARegion) proxy.getHARegion();

	for (Iterator itr = regionForQueue.values().iterator(); itr.hasNext();) {
	Object obj = itr.next();
	if (obj instanceof HAEventWrapper) {
	Conflatable confObj = (Conflatable) obj;
	if (KEY1.equals(confObj.getKeyToConflate())
	\|\| KEY2.equals(confObj.getKeyToConflate())) {
	isObjectPresent = true;
	}
	}
	}
	}

	if (isObjectPresent) {
	break; // From while.
	}

	try {
	Thread.sleep(10);
	} catch (InterruptedException e) {
	fail("interrupted", e);
	}
	}
	}
	}
	}