geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/CacheClientNotifierDUnitTest.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.wan;

 import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_ID;
 import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_TIMEOUT;
 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.awaitility.GeodeAwaitility.await;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;

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

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

 import org.apache.geode.cache.CacheFactory;
 import org.apache.geode.cache.DiskStore;
 import org.apache.geode.cache.EvictionAction;
 import org.apache.geode.cache.EvictionAttributes;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionAttributes;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.RegionShortcut;
 import org.apache.geode.cache.Scope;
 import org.apache.geode.cache.client.Pool;
 import org.apache.geode.cache.client.PoolManager;
 import org.apache.geode.cache.server.CacheServer;
 import org.apache.geode.cache.server.ClientSubscriptionConfig;
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
 import org.apache.geode.internal.cache.CacheServerImpl;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.cache.InternalCacheServer;
 import org.apache.geode.internal.cache.ha.HAContainerRegion;
 import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
 import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
 import org.apache.geode.test.dunit.LogWriterUtils;
 import org.apache.geode.test.dunit.SerializableRunnable;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.junit.categories.WanTest;

 @Category({WanTest.class})
 public class CacheClientNotifierDUnitTest extends WANTestBase {

   private static final int NUM_KEYS = 10;

   private int createCacheServerWithCSC(VM vm, final boolean withCSC, final int capacity,
       final String policy, final String diskStoreName) {
     final int serverPort = getRandomAvailableTCPPort();

     SerializableRunnable createCacheServer = new SerializableRunnable() {
       @Override
       public void run() throws Exception {
         CacheServerImpl server = (CacheServerImpl) cache.addCacheServer();
         server.setPort(serverPort);
         if (withCSC) {
           if (diskStoreName != null) {
             DiskStore ds = cache.findDiskStore(diskStoreName);
             if (ds == null) {
               ds = cache.createDiskStoreFactory().create(diskStoreName);
             }
           }
           ClientSubscriptionConfig csc = server.getClientSubscriptionConfig();
           csc.setCapacity(capacity);
           csc.setEvictionPolicy(policy);
           csc.setDiskStoreName(diskStoreName);
           server.setHostnameForClients("localhost");
           // server.setGroups(new String[]{"serv"});
         }
         try {
           server.start();
         } catch (IOException e) {
           org.apache.geode.test.dunit.Assert.fail("Failed to start server ", e);
         }
       }
     };
     vm.invoke(createCacheServer);
     return serverPort;
   }

   private void checkCacheServer(VM vm, final int serverPort, final boolean withCSC,
       final int capacity) {
     SerializableRunnable checkCacheServer = new SerializableRunnable() {

       @Override
       public void run() throws Exception {
         List<InternalCacheServer> cacheServers =
             ((InternalCache) cache).getCacheServersAndGatewayReceiver();
         CacheServerImpl server = null;
         for (CacheServer cs : cacheServers) {
           if (cs.getPort() == serverPort) {
             server = (CacheServerImpl) cs;
             break;
           }
         }
         assertNotNull(server);
         CacheClientNotifier ccn = server.getAcceptor().getCacheClientNotifier();
         HAContainerRegion haContainer = (HAContainerRegion) ccn.getHaContainer();
         if (server.getAcceptor().isGatewayReceiver()) {
           assertNull(haContainer);
           return;
         }
         Region internalRegion = haContainer.getMapForTest();
         RegionAttributes ra = internalRegion.getAttributes();
         EvictionAttributes ea = ra.getEvictionAttributes();
         if (withCSC) {
           assertNotNull(ea);
           assertEquals(capacity, ea.getMaximum());
           assertEquals(EvictionAction.OVERFLOW_TO_DISK, ea.getAction());
         } else {
           assertNull(ea);
         }
       }
     };
     vm.invoke(checkCacheServer);
   }

   public static void closeACacheServer(final int serverPort) {
     List<CacheServer> cacheServers = cache.getCacheServers();
     CacheServerImpl server = null;
     for (CacheServer cs : cacheServers) {
       if (cs.getPort() == serverPort) {
         server = (CacheServerImpl) cs;
         break;
       }
     }
     assertNotNull(server);
     server.stop();
   }

   private void verifyRegionSize(VM vm, final int expect) {
     SerializableRunnable verifyRegionSize = new SerializableRunnable() {
       @Override
       public void run() throws Exception {
         final Region region = cache.getRegion(getTestMethodName() + "_PR");

         await().untilAsserted(() -> assertEquals(expect, region.size()));
       }
     };
     vm.invoke(verifyRegionSize);
   }

   /**
    * The test will start several cache servers, including gateway receivers. Shutdown them and
    * verify the CacheClientNotifier for each server is correct
    */
   @Test
   public void testNormalClient2MultipleCacheServer() throws Exception {
     doMultipleCacheServer(false);
   }

   public void doMultipleCacheServer(boolean durable) throws Exception {
     /* test scenario: */
     /* create 1 GatewaySender on vm5 */
     /* create 1 GatewayReceiver on vm2 */
     /* create 2 cache servers on vm2, one with overflow. */
     /* verify if the cache server2 still has the overflow attributes */
     /* create 1 cache client1 on vm3 to register interest on cache server1 */
     /* create 1 cache client2 on vm4 to register interest on cache server1 */
     /* do some puts to GatewaySender on vm5 */

     // start locators
     Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
     Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));

     // create receiver and cache servers will be at ny
     vm2.invoke(() -> WANTestBase.createCache(nyPort));
     int receiverPort = vm2.invoke(() -> WANTestBase.createReceiver());
     checkCacheServer(vm2, receiverPort, false, 0);

     // create PR for receiver
     vm2.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName() + "_PR",
         null, 1, 100, isOffHeap()));

     // create cache server1 with overflow
     int serverPort = createCacheServerWithCSC(vm2, true, 3, "entry", "DEFAULT");
     checkCacheServer(vm2, serverPort, true, 3);

     /*
      * GEODE-6344: not to create the 2nd cacheserver on the same jvm until refix of GEODE-1183,
      * otherwise the test might be hang
      * // create cache server 2
      * final int serverPort2 = createCacheServerWithCSC(vm2, false, 0, null, null);
      * // Currently, only the first cache server's overflow attributes will take effect
      * // It will be enhanced in GEODE-1102
      * checkCacheServer(vm2, serverPort2, true, 3);
      * LogService.getLogger().info("receiverPort=" + receiverPort + ",serverPort=" + serverPort
      * + ",serverPort2=" + serverPort2);
      */

     vm3.invoke(() -> createClientWithLocator(nyPort, "localhost", getTestMethodName() + "_PR",
         "123", durable));
     vm4.invoke(() -> createClientWithLocator(nyPort, "localhost", getTestMethodName() + "_PR",
         "124", durable));

     // create sender at ln
     vm5.invoke(() -> WANTestBase.createCache(lnPort));
     vm5.invoke(() -> WANTestBase.createSender("ln", 2, false, 100, 400, false, false, null, true));
     vm5.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName() + "_PR",
         "ln", 1, 100, isOffHeap()));
     vm5.invoke(() -> WANTestBase.startSender("ln"));
     vm5.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_PR", NUM_KEYS));

     /* verify */
     verifyRegionSize(vm5, NUM_KEYS);
     verifyRegionSize(vm2, NUM_KEYS);
     verifyRegionSize(vm4, NUM_KEYS);
     verifyRegionSize(vm3, NUM_KEYS);

     /*
      * GEODE-6344
      * // close a cache server, then re-test
      * vm2.invoke(() -> closeACacheServer(serverPort2));
      */

     vm5.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_PR", NUM_KEYS * 2));

     /* verify */
     verifyRegionSize(vm5, NUM_KEYS * 2);
     verifyRegionSize(vm2, NUM_KEYS * 2);
     verifyRegionSize(vm4, NUM_KEYS * 2);
     verifyRegionSize(vm3, NUM_KEYS * 2);

     disconnectAllFromDS();
   }

   public static void createClientWithLocator(int port0, String host, String regionName,
       String clientId, boolean isDurable) {
     WANTestBase test = new WANTestBase();
     Properties props = test.getDistributedSystemProperties();
     props.setProperty(MCAST_PORT, "0");
     props.setProperty(LOCATORS, "");
     if (isDurable) {
       props.setProperty(DURABLE_CLIENT_ID, clientId);
       props.setProperty(DURABLE_CLIENT_TIMEOUT, "" + 200);
     }

     InternalDistributedSystem ds = test.getSystem(props);
     cache = (InternalCache) CacheFactory.create(ds);

     assertNotNull(cache);
     CacheServerTestUtil.disableShufflingOfEndpoints();
     Pool p;
     try {
       p = PoolManager.createFactory().addLocator(host, port0).setPingInterval(250)
           .setSubscriptionEnabled(true).setSubscriptionRedundancy(-1).setSocketBufferSize(1000)
           .setMinConnections(6).setMaxConnections(10).setRetryAttempts(3).create(regionName);
     } finally {
       CacheServerTestUtil.enableShufflingOfEndpoints();
     }

     RegionFactory factory = cache.createRegionFactory(RegionShortcut.LOCAL)
         .setScope(Scope.DISTRIBUTED_NO_ACK)
         .setPoolName(p.getName());
     region = factory.create(regionName);
     region.registerInterest("ALL_KEYS");
     assertNotNull(region);
     if (isDurable) {
       cache.readyForEvents();
     }
     LogWriterUtils.getLogWriter()
         .info("Distributed Region " + regionName + " created Successfully :" + region.toString()
             + " in a " + (isDurable ? "durable" : "") + " client");
   }
 }
	/*
	* 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.wan;

	import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_ID;
	import static org.apache.geode.distributed.ConfigurationProperties.DURABLE_CLIENT_TIMEOUT;
	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.awaitility.GeodeAwaitility.await;
	import static org.junit.Assert.assertEquals;
	import static org.junit.Assert.assertNotNull;
	import static org.junit.Assert.assertNull;

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

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

	import org.apache.geode.cache.CacheFactory;
	import org.apache.geode.cache.DiskStore;
	import org.apache.geode.cache.EvictionAction;
	import org.apache.geode.cache.EvictionAttributes;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionAttributes;
	import org.apache.geode.cache.RegionFactory;
	import org.apache.geode.cache.RegionShortcut;
	import org.apache.geode.cache.Scope;
	import org.apache.geode.cache.client.Pool;
	import org.apache.geode.cache.client.PoolManager;
	import org.apache.geode.cache.server.CacheServer;
	import org.apache.geode.cache.server.ClientSubscriptionConfig;
	import org.apache.geode.distributed.internal.InternalDistributedSystem;
	import org.apache.geode.internal.cache.CacheServerImpl;
	import org.apache.geode.internal.cache.InternalCache;
	import org.apache.geode.internal.cache.InternalCacheServer;
	import org.apache.geode.internal.cache.ha.HAContainerRegion;
	import org.apache.geode.internal.cache.tier.sockets.CacheClientNotifier;
	import org.apache.geode.internal.cache.tier.sockets.CacheServerTestUtil;
	import org.apache.geode.test.dunit.LogWriterUtils;
	import org.apache.geode.test.dunit.SerializableRunnable;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.junit.categories.WanTest;

	@Category({WanTest.class})
	public class CacheClientNotifierDUnitTest extends WANTestBase {

	private static final int NUM_KEYS = 10;

	private int createCacheServerWithCSC(VM vm, final boolean withCSC, final int capacity,
	final String policy, final String diskStoreName) {
	final int serverPort = getRandomAvailableTCPPort();

	SerializableRunnable createCacheServer = new SerializableRunnable() {
	@Override
	public void run() throws Exception {
	CacheServerImpl server = (CacheServerImpl) cache.addCacheServer();
	server.setPort(serverPort);
	if (withCSC) {
	if (diskStoreName != null) {
	DiskStore ds = cache.findDiskStore(diskStoreName);
	if (ds == null) {
	ds = cache.createDiskStoreFactory().create(diskStoreName);
	}
	}
	ClientSubscriptionConfig csc = server.getClientSubscriptionConfig();
	csc.setCapacity(capacity);
	csc.setEvictionPolicy(policy);
	csc.setDiskStoreName(diskStoreName);
	server.setHostnameForClients("localhost");
	// server.setGroups(new String[]{"serv"});
	}
	try {
	server.start();
	} catch (IOException e) {
	org.apache.geode.test.dunit.Assert.fail("Failed to start server ", e);
	}
	}
	};
	vm.invoke(createCacheServer);
	return serverPort;
	}

	private void checkCacheServer(VM vm, final int serverPort, final boolean withCSC,
	final int capacity) {
	SerializableRunnable checkCacheServer = new SerializableRunnable() {

	@Override
	public void run() throws Exception {
	List<InternalCacheServer> cacheServers =
	((InternalCache) cache).getCacheServersAndGatewayReceiver();
	CacheServerImpl server = null;
	for (CacheServer cs : cacheServers) {
	if (cs.getPort() == serverPort) {
	server = (CacheServerImpl) cs;
	break;
	}
	}
	assertNotNull(server);
	CacheClientNotifier ccn = server.getAcceptor().getCacheClientNotifier();
	HAContainerRegion haContainer = (HAContainerRegion) ccn.getHaContainer();
	if (server.getAcceptor().isGatewayReceiver()) {
	assertNull(haContainer);
	return;
	}
	Region internalRegion = haContainer.getMapForTest();
	RegionAttributes ra = internalRegion.getAttributes();
	EvictionAttributes ea = ra.getEvictionAttributes();
	if (withCSC) {
	assertNotNull(ea);
	assertEquals(capacity, ea.getMaximum());
	assertEquals(EvictionAction.OVERFLOW_TO_DISK, ea.getAction());
	} else {
	assertNull(ea);
	}
	}
	};
	vm.invoke(checkCacheServer);
	}

	public static void closeACacheServer(final int serverPort) {
	List<CacheServer> cacheServers = cache.getCacheServers();
	CacheServerImpl server = null;
	for (CacheServer cs : cacheServers) {
	if (cs.getPort() == serverPort) {
	server = (CacheServerImpl) cs;
	break;
	}
	}
	assertNotNull(server);
	server.stop();
	}

	private void verifyRegionSize(VM vm, final int expect) {
	SerializableRunnable verifyRegionSize = new SerializableRunnable() {
	@Override
	public void run() throws Exception {
	final Region region = cache.getRegion(getTestMethodName() + "_PR");

	await().untilAsserted(() -> assertEquals(expect, region.size()));
	}
	};
	vm.invoke(verifyRegionSize);
	}

	/**
	* The test will start several cache servers, including gateway receivers. Shutdown them and
	* verify the CacheClientNotifier for each server is correct
	*/
	@Test
	public void testNormalClient2MultipleCacheServer() throws Exception {
	doMultipleCacheServer(false);
	}

	public void doMultipleCacheServer(boolean durable) throws Exception {
	/* test scenario: */
	/* create 1 GatewaySender on vm5 */
	/* create 1 GatewayReceiver on vm2 */
	/* create 2 cache servers on vm2, one with overflow. */
	/* verify if the cache server2 still has the overflow attributes */
	/* create 1 cache client1 on vm3 to register interest on cache server1 */
	/* create 1 cache client2 on vm4 to register interest on cache server1 */
	/* do some puts to GatewaySender on vm5 */

	// start locators
	Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
	Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));

	// create receiver and cache servers will be at ny
	vm2.invoke(() -> WANTestBase.createCache(nyPort));
	int receiverPort = vm2.invoke(() -> WANTestBase.createReceiver());
	checkCacheServer(vm2, receiverPort, false, 0);

	// create PR for receiver
	vm2.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName() + "_PR",
	null, 1, 100, isOffHeap()));

	// create cache server1 with overflow
	int serverPort = createCacheServerWithCSC(vm2, true, 3, "entry", "DEFAULT");
	checkCacheServer(vm2, serverPort, true, 3);

	/*
	* GEODE-6344: not to create the 2nd cacheserver on the same jvm until refix of GEODE-1183,
	* otherwise the test might be hang
	* // create cache server 2
	* final int serverPort2 = createCacheServerWithCSC(vm2, false, 0, null, null);
	* // Currently, only the first cache server's overflow attributes will take effect
	* // It will be enhanced in GEODE-1102
	* checkCacheServer(vm2, serverPort2, true, 3);
	* LogService.getLogger().info("receiverPort=" + receiverPort + ",serverPort=" + serverPort
	* + ",serverPort2=" + serverPort2);
	*/

	vm3.invoke(() -> createClientWithLocator(nyPort, "localhost", getTestMethodName() + "_PR",
	"123", durable));
	vm4.invoke(() -> createClientWithLocator(nyPort, "localhost", getTestMethodName() + "_PR",
	"124", durable));

	// create sender at ln
	vm5.invoke(() -> WANTestBase.createCache(lnPort));
	vm5.invoke(() -> WANTestBase.createSender("ln", 2, false, 100, 400, false, false, null, true));
	vm5.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName() + "_PR",
	"ln", 1, 100, isOffHeap()));
	vm5.invoke(() -> WANTestBase.startSender("ln"));
	vm5.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_PR", NUM_KEYS));

	/* verify */
	verifyRegionSize(vm5, NUM_KEYS);
	verifyRegionSize(vm2, NUM_KEYS);
	verifyRegionSize(vm4, NUM_KEYS);
	verifyRegionSize(vm3, NUM_KEYS);

	/*
	* GEODE-6344
	* // close a cache server, then re-test
	* vm2.invoke(() -> closeACacheServer(serverPort2));
	*/

	vm5.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_PR", NUM_KEYS * 2));

	/* verify */
	verifyRegionSize(vm5, NUM_KEYS * 2);
	verifyRegionSize(vm2, NUM_KEYS * 2);
	verifyRegionSize(vm4, NUM_KEYS * 2);
	verifyRegionSize(vm3, NUM_KEYS * 2);

	disconnectAllFromDS();
	}

	public static void createClientWithLocator(int port0, String host, String regionName,
	String clientId, boolean isDurable) {
	WANTestBase test = new WANTestBase();
	Properties props = test.getDistributedSystemProperties();
	props.setProperty(MCAST_PORT, "0");
	props.setProperty(LOCATORS, "");
	if (isDurable) {
	props.setProperty(DURABLE_CLIENT_ID, clientId);
	props.setProperty(DURABLE_CLIENT_TIMEOUT, "" + 200);
	}

	InternalDistributedSystem ds = test.getSystem(props);
	cache = (InternalCache) CacheFactory.create(ds);

	assertNotNull(cache);
	CacheServerTestUtil.disableShufflingOfEndpoints();
	Pool p;
	try {
	p = PoolManager.createFactory().addLocator(host, port0).setPingInterval(250)
	.setSubscriptionEnabled(true).setSubscriptionRedundancy(-1).setSocketBufferSize(1000)
	.setMinConnections(6).setMaxConnections(10).setRetryAttempts(3).create(regionName);
	} finally {
	CacheServerTestUtil.enableShufflingOfEndpoints();
	}

	RegionFactory factory = cache.createRegionFactory(RegionShortcut.LOCAL)
	.setScope(Scope.DISTRIBUTED_NO_ACK)
	.setPoolName(p.getName());
	region = factory.create(regionName);
	region.registerInterest("ALL_KEYS");
	assertNotNull(region);
	if (isDurable) {
	cache.readyForEvents();
	}
	LogWriterUtils.getLogWriter()
	.info("Distributed Region " + regionName + " created Successfully :" + region.toString()
	+ " in a " + (isDurable ? "durable" : "") + " client");
	}
	}