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

 import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT;
 import static org.apache.geode.cache.RegionShortcut.REPLICATE;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.apache.geode.test.dunit.VM.getCurrentVMNum;
 import static org.apache.geode.test.dunit.VM.getVM;
 import static org.assertj.core.api.Assertions.assertThat;

 import java.io.File;
 import java.io.IOException;
 import java.io.Serializable;
 import java.io.UncheckedIOException;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Properties;
 import java.util.stream.Collectors;

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

 import org.apache.geode.cache.DiskStoreFactory;
 import org.apache.geode.cache.PartitionAttributesFactory;
 import org.apache.geode.cache.Region;
 import org.apache.geode.cache.RegionFactory;
 import org.apache.geode.cache.asyncqueue.AsyncEvent;
 import org.apache.geode.cache.asyncqueue.AsyncEventListener;
 import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
 import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
 import org.apache.geode.cache.asyncqueue.internal.InternalAsyncEventQueue;
 import org.apache.geode.cache.wan.GatewaySender;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.test.dunit.AsyncInvocation;
 import org.apache.geode.test.dunit.ThreadUtils;
 import org.apache.geode.test.dunit.VM;
 import org.apache.geode.test.dunit.rules.CacheRule;
 import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
 import org.apache.geode.test.dunit.rules.DistributedRule;
 import org.apache.geode.test.junit.categories.AEQTest;
 import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;

 /**
  * Extracted from {@link AsyncEventListenerDistributedTest}.
  */
 @Category(AEQTest.class)
 @SuppressWarnings("serial")
 public class ParallelSerialAsyncEventListenerStopStartDistributedTest implements Serializable {

   @Rule
   public DistributedRule distributedRule = new DistributedRule();

   @Rule
   public CacheRule cacheRule = new CacheRule();

   @Rule
   public DistributedRestoreSystemProperties restoreSystemProperties =
       new DistributedRestoreSystemProperties();

   @Rule
   public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();

   private String basePartitionedRegionName;
   private String baseReplicateRegionName;
   private String baseSerialAsyncEventQueueId;
   private String baseParallelAsyncEventQueueId;

   private VM vm0;
   private VM vm1;
   private VM vm2;
   private VM vm3;

   @Before
   public void setUp() throws Exception {
     vm0 = getVM(0);
     vm1 = getVM(1);
     vm2 = getVM(2);
     vm3 = getVM(3);

     String className = getClass().getSimpleName();
     basePartitionedRegionName = className + "_PR_";
     baseReplicateRegionName = className + "_RR_";
     baseSerialAsyncEventQueueId = className + "_serialAEQ_";
     baseParallelAsyncEventQueueId = className + "_parallelAEQ_";
   }

   /**
    * Override as needed to add to the configuration, such as off-heap-memory-size.
    */
   protected Properties getDistributedSystemProperties() {
     return new Properties();
   }

   /**
    * Override as needed to add to the configuration, such as regionFactory.setOffHeap(boolean).
    */
   protected RegionFactory<?, ?> configureRegion(RegionFactory<?, ?> regionFactory) {
     return regionFactory;
   }

   @Test
   public void testStopStartPersistentParallelAndSerialAsyncEventQueues() {
     // Create cache
     vm0.invoke(() -> createCache());
     vm1.invoke(() -> createCache());
     vm2.invoke(() -> createCache());
     vm3.invoke(() -> createCache());

     // Create several serial and parallel AEQs
     int numAEQs = 3;
     vm0.invoke(() -> createAsyncEventQueues(numAEQs));
     vm1.invoke(() -> createAsyncEventQueues(numAEQs));
     vm2.invoke(() -> createAsyncEventQueues(numAEQs));
     vm3.invoke(() -> createAsyncEventQueues(numAEQs));

     // Create replicated and partitioned regions attached to those AEQs
     int numRegions = 3;
     vm0.invoke(() -> createRegions(numRegions));
     vm1.invoke(() -> createRegions(numRegions));
     vm2.invoke(() -> createRegions(numRegions));
     vm3.invoke(() -> createRegions(numRegions));

     // Do puts into all replicated and partitioned regions
     vm0.invoke(() -> doPuts(numRegions, 1000));

     // Wait for all AEQs to be empty
     vm0.invoke(() -> waitForAsyncQueuesToEmpty());
     vm1.invoke(() -> waitForAsyncQueuesToEmpty());
     vm2.invoke(() -> waitForAsyncQueuesToEmpty());
     vm3.invoke(() -> waitForAsyncQueuesToEmpty());

     // Stop all AEQs
     vm0.invoke(() -> stopAsyncQueues());
     vm1.invoke(() -> stopAsyncQueues());
     vm2.invoke(() -> stopAsyncQueues());
     vm3.invoke(() -> stopAsyncQueues());

     // Start all AEQs
     AsyncInvocation startAeqsVm0 = vm0.invokeAsync(() -> startAsyncQueues());
     AsyncInvocation startAeqsVm1 = vm1.invokeAsync(() -> startAsyncQueues());
     AsyncInvocation startAeqsVm2 = vm2.invokeAsync(() -> startAsyncQueues());
     AsyncInvocation startAeqsVm3 = vm3.invokeAsync(() -> startAsyncQueues());

     // Wait for async tasks to complete
     ThreadUtils.join(startAeqsVm0, 120 * 1000);
     ThreadUtils.join(startAeqsVm1, 120 * 1000);
     ThreadUtils.join(startAeqsVm2, 120 * 1000);
     ThreadUtils.join(startAeqsVm3, 120 * 1000);
   }

   private void createAsyncEventQueues(int numAEQs) {
     for (int i = 0; i < numAEQs; i++) {
       createPersistentAsyncEventQueue(baseSerialAsyncEventQueueId + i,
           createDiskStoreName(baseSerialAsyncEventQueueId + i), false);
       createPersistentAsyncEventQueue(baseParallelAsyncEventQueueId + i,
           createDiskStoreName(baseParallelAsyncEventQueueId + i), true);
     }
   }

   private void createRegions(int numRegions) {
     for (int i = 0; i < numRegions; i++) {
       createReplicateRegion(baseReplicateRegionName + i, baseSerialAsyncEventQueueId + i);
       createPartitionedRegion(basePartitionedRegionName + i, baseParallelAsyncEventQueueId + i);
     }
   }

   private void waitForAsyncQueuesToEmpty() {
     for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
       await().until(() -> aeq.size() == 0);
     }
   }

   private void stopAsyncQueues() {
     for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
       InternalAsyncEventQueue iaeq = (InternalAsyncEventQueue) aeq;
       assertThat(iaeq.getSender().isRunning()).isTrue();
       iaeq.stop();
       assertThat(iaeq.getSender().isRunning()).isFalse();
     }
   }

   private void startAsyncQueues() {
     for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
       GatewaySender sender = ((InternalAsyncEventQueue) aeq).getSender();
       assertThat(sender.isRunning()).isFalse();
       sender.start();
       assertThat(sender.isRunning()).isTrue();
     }
   }

   private Collection<AsyncEventQueue> getAsyncEventQueues() {
     return getCache().getAsyncEventQueues().stream()
         .sorted(Comparator.comparing(AsyncEventQueue::getId)).collect(Collectors.toList());
   }

   private InternalCache getCache() {
     return cacheRule.getOrCreateCache(getDistributedSystemProperties());
   }

   private void createCache() {
     cacheRule.createCache(getDistributedSystemProperties());
   }

   private void createPartitionedRegion(String regionName, String asyncEventQueueId) {
     assertThat(regionName).isNotEmpty();
     assertThat(asyncEventQueueId).isNotEmpty();

     PartitionAttributesFactory<?, ?> partitionAttributesFactory = new PartitionAttributesFactory();

     RegionFactory<?, ?> regionFactory = getCache().createRegionFactory(PARTITION_REDUNDANT);
     regionFactory.addAsyncEventQueueId(asyncEventQueueId);
     regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

     configureRegion(regionFactory).create(regionName);
   }

   private void createReplicateRegion(String regionName, String asyncEventQueueId) {
     assertThat(regionName).isNotEmpty();
     assertThat(asyncEventQueueId).isNotEmpty();

     RegionFactory<?, ?> regionFactory = getCache().createRegionFactory(REPLICATE);
     regionFactory.addAsyncEventQueueId(asyncEventQueueId);

     configureRegion(regionFactory).create(regionName);
   }

   private void createDiskStore(String diskStoreName, String asyncEventQueueId) {
     assertThat(diskStoreName).isNotEmpty();
     assertThat(asyncEventQueueId).isNotEmpty();

     File directory = createDirectory(createDiskStoreName(asyncEventQueueId));

     DiskStoreFactory diskStoreFactory = getCache().createDiskStoreFactory();
     diskStoreFactory.setDiskDirs(new File[] {directory});

     diskStoreFactory.create(diskStoreName);
   }

   private File createDirectory(String name) {
     assertThat(name).isNotEmpty();

     File directory = new File(temporaryFolder.getRoot(), name);
     if (!directory.exists()) {
       try {
         return temporaryFolder.newFolder(name);
       } catch (IOException e) {
         throw new UncheckedIOException(e);
       }
     }
     return directory;
   }

   private String createDiskStoreName(String asyncEventQueueId) {
     assertThat(asyncEventQueueId).isNotEmpty();

     return asyncEventQueueId + "_disk_" + getCurrentVMNum();
   }

   private void createPersistentAsyncEventQueue(String asyncEventQueueId, String diskStoreName,
       boolean isParallel) {
     createDiskStore(diskStoreName, asyncEventQueueId);

     AsyncEventQueueFactory asyncEventQueueFactory = getCache().createAsyncEventQueueFactory();
     asyncEventQueueFactory.setDiskStoreName(diskStoreName);
     asyncEventQueueFactory.setParallel(isParallel);
     asyncEventQueueFactory.setPersistent(true);

     asyncEventQueueFactory.create(asyncEventQueueId, new TestAsyncEventListener());
   }

   private void doPuts(int numRegions, int numPuts) {
     for (int i = 0; i < numRegions; i++) {
       Region<Integer, Integer> rr = getCache().getRegion(baseReplicateRegionName + i);
       Region<Integer, Integer> pr = getCache().getRegion(basePartitionedRegionName + i);
       for (int j = 0; j < numPuts; j++) {
         rr.put(j, j);
         pr.put(j, j);
       }
     }
   }

   private static class TestAsyncEventListener implements AsyncEventListener {

     @Override
     public synchronized boolean processEvents(List<AsyncEvent> events) {
       return true;
     }
   }
 }
	/*
	* 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.asyncqueue;

	import static org.apache.geode.cache.RegionShortcut.PARTITION_REDUNDANT;
	import static org.apache.geode.cache.RegionShortcut.REPLICATE;
	import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
	import static org.apache.geode.test.dunit.VM.getCurrentVMNum;
	import static org.apache.geode.test.dunit.VM.getVM;
	import static org.assertj.core.api.Assertions.assertThat;

	import java.io.File;
	import java.io.IOException;
	import java.io.Serializable;
	import java.io.UncheckedIOException;
	import java.util.Collection;
	import java.util.Comparator;
	import java.util.List;
	import java.util.Properties;
	import java.util.stream.Collectors;

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

	import org.apache.geode.cache.DiskStoreFactory;
	import org.apache.geode.cache.PartitionAttributesFactory;
	import org.apache.geode.cache.Region;
	import org.apache.geode.cache.RegionFactory;
	import org.apache.geode.cache.asyncqueue.AsyncEvent;
	import org.apache.geode.cache.asyncqueue.AsyncEventListener;
	import org.apache.geode.cache.asyncqueue.AsyncEventQueue;
	import org.apache.geode.cache.asyncqueue.AsyncEventQueueFactory;
	import org.apache.geode.cache.asyncqueue.internal.InternalAsyncEventQueue;
	import org.apache.geode.cache.wan.GatewaySender;
	import org.apache.geode.internal.cache.InternalCache;
	import org.apache.geode.test.dunit.AsyncInvocation;
	import org.apache.geode.test.dunit.ThreadUtils;
	import org.apache.geode.test.dunit.VM;
	import org.apache.geode.test.dunit.rules.CacheRule;
	import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties;
	import org.apache.geode.test.dunit.rules.DistributedRule;
	import org.apache.geode.test.junit.categories.AEQTest;
	import org.apache.geode.test.junit.rules.serializable.SerializableTemporaryFolder;

	/**
	* Extracted from {@link AsyncEventListenerDistributedTest}.
	*/
	@Category(AEQTest.class)
	@SuppressWarnings("serial")
	public class ParallelSerialAsyncEventListenerStopStartDistributedTest implements Serializable {

	@Rule
	public DistributedRule distributedRule = new DistributedRule();

	@Rule
	public CacheRule cacheRule = new CacheRule();

	@Rule
	public DistributedRestoreSystemProperties restoreSystemProperties =
	new DistributedRestoreSystemProperties();

	@Rule
	public SerializableTemporaryFolder temporaryFolder = new SerializableTemporaryFolder();

	private String basePartitionedRegionName;
	private String baseReplicateRegionName;
	private String baseSerialAsyncEventQueueId;
	private String baseParallelAsyncEventQueueId;

	private VM vm0;
	private VM vm1;
	private VM vm2;
	private VM vm3;

	@Before
	public void setUp() throws Exception {
	vm0 = getVM(0);
	vm1 = getVM(1);
	vm2 = getVM(2);
	vm3 = getVM(3);

	String className = getClass().getSimpleName();
	basePartitionedRegionName = className + "_PR_";
	baseReplicateRegionName = className + "_RR_";
	baseSerialAsyncEventQueueId = className + "_serialAEQ_";
	baseParallelAsyncEventQueueId = className + "_parallelAEQ_";
	}

	/**
	* Override as needed to add to the configuration, such as off-heap-memory-size.
	*/
	protected Properties getDistributedSystemProperties() {
	return new Properties();
	}

	/**
	* Override as needed to add to the configuration, such as regionFactory.setOffHeap(boolean).
	*/
	protected RegionFactory<?, ?> configureRegion(RegionFactory<?, ?> regionFactory) {
	return regionFactory;
	}

	@Test
	public void testStopStartPersistentParallelAndSerialAsyncEventQueues() {
	// Create cache
	vm0.invoke(() -> createCache());
	vm1.invoke(() -> createCache());
	vm2.invoke(() -> createCache());
	vm3.invoke(() -> createCache());

	// Create several serial and parallel AEQs
	int numAEQs = 3;
	vm0.invoke(() -> createAsyncEventQueues(numAEQs));
	vm1.invoke(() -> createAsyncEventQueues(numAEQs));
	vm2.invoke(() -> createAsyncEventQueues(numAEQs));
	vm3.invoke(() -> createAsyncEventQueues(numAEQs));

	// Create replicated and partitioned regions attached to those AEQs
	int numRegions = 3;
	vm0.invoke(() -> createRegions(numRegions));
	vm1.invoke(() -> createRegions(numRegions));
	vm2.invoke(() -> createRegions(numRegions));
	vm3.invoke(() -> createRegions(numRegions));

	// Do puts into all replicated and partitioned regions
	vm0.invoke(() -> doPuts(numRegions, 1000));

	// Wait for all AEQs to be empty
	vm0.invoke(() -> waitForAsyncQueuesToEmpty());
	vm1.invoke(() -> waitForAsyncQueuesToEmpty());
	vm2.invoke(() -> waitForAsyncQueuesToEmpty());
	vm3.invoke(() -> waitForAsyncQueuesToEmpty());

	// Stop all AEQs
	vm0.invoke(() -> stopAsyncQueues());
	vm1.invoke(() -> stopAsyncQueues());
	vm2.invoke(() -> stopAsyncQueues());
	vm3.invoke(() -> stopAsyncQueues());

	// Start all AEQs
	AsyncInvocation startAeqsVm0 = vm0.invokeAsync(() -> startAsyncQueues());
	AsyncInvocation startAeqsVm1 = vm1.invokeAsync(() -> startAsyncQueues());
	AsyncInvocation startAeqsVm2 = vm2.invokeAsync(() -> startAsyncQueues());
	AsyncInvocation startAeqsVm3 = vm3.invokeAsync(() -> startAsyncQueues());

	// Wait for async tasks to complete
	ThreadUtils.join(startAeqsVm0, 120 * 1000);
	ThreadUtils.join(startAeqsVm1, 120 * 1000);
	ThreadUtils.join(startAeqsVm2, 120 * 1000);
	ThreadUtils.join(startAeqsVm3, 120 * 1000);
	}

	private void createAsyncEventQueues(int numAEQs) {
	for (int i = 0; i < numAEQs; i++) {
	createPersistentAsyncEventQueue(baseSerialAsyncEventQueueId + i,
	createDiskStoreName(baseSerialAsyncEventQueueId + i), false);
	createPersistentAsyncEventQueue(baseParallelAsyncEventQueueId + i,
	createDiskStoreName(baseParallelAsyncEventQueueId + i), true);
	}
	}

	private void createRegions(int numRegions) {
	for (int i = 0; i < numRegions; i++) {
	createReplicateRegion(baseReplicateRegionName + i, baseSerialAsyncEventQueueId + i);
	createPartitionedRegion(basePartitionedRegionName + i, baseParallelAsyncEventQueueId + i);
	}
	}

	private void waitForAsyncQueuesToEmpty() {
	for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
	await().until(() -> aeq.size() == 0);
	}
	}

	private void stopAsyncQueues() {
	for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
	InternalAsyncEventQueue iaeq = (InternalAsyncEventQueue) aeq;
	assertThat(iaeq.getSender().isRunning()).isTrue();
	iaeq.stop();
	assertThat(iaeq.getSender().isRunning()).isFalse();
	}
	}

	private void startAsyncQueues() {
	for (final AsyncEventQueue aeq : getAsyncEventQueues()) {
	GatewaySender sender = ((InternalAsyncEventQueue) aeq).getSender();
	assertThat(sender.isRunning()).isFalse();
	sender.start();
	assertThat(sender.isRunning()).isTrue();
	}
	}

	private Collection<AsyncEventQueue> getAsyncEventQueues() {
	return getCache().getAsyncEventQueues().stream()
	.sorted(Comparator.comparing(AsyncEventQueue::getId)).collect(Collectors.toList());
	}

	private InternalCache getCache() {
	return cacheRule.getOrCreateCache(getDistributedSystemProperties());
	}

	private void createCache() {
	cacheRule.createCache(getDistributedSystemProperties());
	}

	private void createPartitionedRegion(String regionName, String asyncEventQueueId) {
	assertThat(regionName).isNotEmpty();
	assertThat(asyncEventQueueId).isNotEmpty();

	PartitionAttributesFactory<?, ?> partitionAttributesFactory = new PartitionAttributesFactory();

	RegionFactory<?, ?> regionFactory = getCache().createRegionFactory(PARTITION_REDUNDANT);
	regionFactory.addAsyncEventQueueId(asyncEventQueueId);
	regionFactory.setPartitionAttributes(partitionAttributesFactory.create());

	configureRegion(regionFactory).create(regionName);
	}

	private void createReplicateRegion(String regionName, String asyncEventQueueId) {
	assertThat(regionName).isNotEmpty();
	assertThat(asyncEventQueueId).isNotEmpty();

	RegionFactory<?, ?> regionFactory = getCache().createRegionFactory(REPLICATE);
	regionFactory.addAsyncEventQueueId(asyncEventQueueId);

	configureRegion(regionFactory).create(regionName);
	}

	private void createDiskStore(String diskStoreName, String asyncEventQueueId) {
	assertThat(diskStoreName).isNotEmpty();
	assertThat(asyncEventQueueId).isNotEmpty();

	File directory = createDirectory(createDiskStoreName(asyncEventQueueId));

	DiskStoreFactory diskStoreFactory = getCache().createDiskStoreFactory();
	diskStoreFactory.setDiskDirs(new File[] {directory});

	diskStoreFactory.create(diskStoreName);
	}

	private File createDirectory(String name) {
	assertThat(name).isNotEmpty();

	File directory = new File(temporaryFolder.getRoot(), name);
	if (!directory.exists()) {
	try {
	return temporaryFolder.newFolder(name);
	} catch (IOException e) {
	throw new UncheckedIOException(e);
	}
	}
	return directory;
	}

	private String createDiskStoreName(String asyncEventQueueId) {
	assertThat(asyncEventQueueId).isNotEmpty();

	return asyncEventQueueId + "_disk_" + getCurrentVMNum();
	}

	private void createPersistentAsyncEventQueue(String asyncEventQueueId, String diskStoreName,
	boolean isParallel) {
	createDiskStore(diskStoreName, asyncEventQueueId);

	AsyncEventQueueFactory asyncEventQueueFactory = getCache().createAsyncEventQueueFactory();
	asyncEventQueueFactory.setDiskStoreName(diskStoreName);
	asyncEventQueueFactory.setParallel(isParallel);
	asyncEventQueueFactory.setPersistent(true);

	asyncEventQueueFactory.create(asyncEventQueueId, new TestAsyncEventListener());
	}

	private void doPuts(int numRegions, int numPuts) {
	for (int i = 0; i < numRegions; i++) {
	Region<Integer, Integer> rr = getCache().getRegion(baseReplicateRegionName + i);
	Region<Integer, Integer> pr = getCache().getRegion(basePartitionedRegionName + i);
	for (int j = 0; j < numPuts; j++) {
	rr.put(j, j);
	pr.put(j, j);
	}
	}
	}

	private static class TestAsyncEventListener implements AsyncEventListener {

	@Override
	public synchronized boolean processEvents(List<AsyncEvent> events) {
	return true;
	}
	}
	}