| /* |
| * 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.cache.Region.SEPARATOR; |
| import static org.apache.geode.distributed.ConfigurationProperties.DISTRIBUTED_SYSTEM_ID; |
| import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.MCAST_PORT; |
| import static org.apache.geode.distributed.ConfigurationProperties.OFF_HEAP_MEMORY_SIZE; |
| import static org.apache.geode.distributed.ConfigurationProperties.REMOTE_LOCATORS; |
| import static org.apache.geode.distributed.ConfigurationProperties.START_LOCATOR; |
| 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.assertTrue; |
| import static org.junit.Assert.fail; |
| |
| import java.io.DataInput; |
| import java.io.DataOutput; |
| import java.io.File; |
| import java.io.IOException; |
| import java.net.InetSocketAddress; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.StringTokenizer; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.atomic.AtomicInteger; |
| |
| import org.apache.geode.DataSerializable; |
| import org.apache.geode.DataSerializer; |
| import org.apache.geode.cache.AttributesFactory; |
| import org.apache.geode.cache.Cache; |
| import org.apache.geode.cache.CacheClosedException; |
| import org.apache.geode.cache.CacheFactory; |
| import org.apache.geode.cache.CacheLoader; |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.Declarable; |
| import org.apache.geode.cache.DiskStore; |
| import org.apache.geode.cache.DiskStoreFactory; |
| import org.apache.geode.cache.EntryEvent; |
| import org.apache.geode.cache.EntryOperation; |
| import org.apache.geode.cache.FixedPartitionAttributes; |
| import org.apache.geode.cache.FixedPartitionResolver; |
| import org.apache.geode.cache.LoaderHelper; |
| import org.apache.geode.cache.PartitionAttributesFactory; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionFactory; |
| import org.apache.geode.cache.Scope; |
| 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.AsyncEventQueueImpl; |
| import org.apache.geode.cache.asyncqueue.internal.AsyncEventQueueStats; |
| import org.apache.geode.cache.client.internal.LocatorDiscoveryCallbackAdapter; |
| import org.apache.geode.cache.control.RebalanceFactory; |
| import org.apache.geode.cache.control.RebalanceOperation; |
| import org.apache.geode.cache.control.RebalanceResults; |
| import org.apache.geode.cache.control.ResourceManager; |
| import org.apache.geode.cache.partition.PartitionRegionHelper; |
| import org.apache.geode.cache.persistence.PartitionOfflineException; |
| import org.apache.geode.cache.util.CacheListenerAdapter; |
| import org.apache.geode.cache.wan.GatewayEventFilter; |
| import org.apache.geode.cache.wan.GatewayEventSubstitutionFilter; |
| import org.apache.geode.cache.wan.GatewayReceiver; |
| import org.apache.geode.cache.wan.GatewayReceiverFactory; |
| import org.apache.geode.cache.wan.GatewaySender; |
| import org.apache.geode.cache.wan.GatewaySender.OrderPolicy; |
| import org.apache.geode.cache.wan.GatewaySenderFactory; |
| import org.apache.geode.distributed.Locator; |
| import org.apache.geode.distributed.internal.InternalDistributedSystem; |
| import org.apache.geode.distributed.internal.InternalLocator; |
| import org.apache.geode.internal.AvailablePortHelper; |
| import org.apache.geode.internal.cache.ForceReattemptException; |
| import org.apache.geode.internal.cache.GemFireCacheImpl; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.RegionQueue; |
| import org.apache.geode.internal.cache.control.InternalResourceManager; |
| import org.apache.geode.internal.cache.control.InternalResourceManager.ResourceObserver; |
| import org.apache.geode.internal.size.Sizeable; |
| import org.apache.geode.test.awaitility.GeodeAwaitility; |
| import org.apache.geode.test.dunit.Assert; |
| import org.apache.geode.test.dunit.Host; |
| import org.apache.geode.test.dunit.IgnoredException; |
| import org.apache.geode.test.dunit.Invoke; |
| import org.apache.geode.test.dunit.LogWriterUtils; |
| 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.dunit.rules.ClusterStartupRule; |
| import org.apache.geode.test.dunit.rules.DistributedRule; |
| |
| |
| /** |
| * @deprecated Please use {@link DistributedRule} and Geode User APIs or {@link ClusterStartupRule} |
| * instead. |
| */ |
| @Deprecated |
| public class AsyncEventQueueTestBase extends JUnit4DistributedTestCase { |
| |
| protected static Cache cache; |
| |
| protected static VM vm0; |
| |
| protected static VM vm1; |
| |
| protected static VM vm2; |
| |
| protected static VM vm3; |
| |
| protected static VM vm4; |
| |
| protected static AsyncEventListener eventListener1; |
| |
| private static final long MAX_WAIT = 60000; |
| |
| protected static GatewayEventFilter eventFilter; |
| |
| protected static boolean destroyFlag = false; |
| |
| protected static List<Integer> dispatcherThreads = new ArrayList<>(Arrays.asList(1, 3, 5)); |
| |
| // this will be set for each test method run with one of the values from above |
| // list |
| protected static int numDispatcherThreadsForTheRun = 1; |
| |
| public AsyncEventQueueTestBase() { |
| super(); |
| } |
| |
| @Override |
| public final void preSetUp() throws Exception { |
| final Host host = Host.getHost(0); |
| vm0 = host.getVM(0); |
| vm1 = host.getVM(1); |
| vm2 = host.getVM(2); |
| vm3 = host.getVM(3); |
| vm4 = host.getVM(4); |
| } |
| |
| @Override |
| public final void postSetUp() throws Exception { |
| // this is done to vary the number of dispatchers for sender |
| // during every test method run |
| shuffleNumDispatcherThreads(); |
| Invoke.invokeInEveryVM(() -> setNumDispatcherThreadsForTheRun(dispatcherThreads.get(0))); |
| } |
| |
| public static void shuffleNumDispatcherThreads() { |
| Collections.shuffle(dispatcherThreads); |
| } |
| |
| public static void setNumDispatcherThreadsForTheRun(int numThreads) { |
| numDispatcherThreadsForTheRun = numThreads; |
| } |
| |
| public static Integer createFirstLocatorWithDSId(int dsId) { |
| if (Locator.hasLocator()) { |
| Locator.getLocator().stop(); |
| } |
| AsyncEventQueueTestBase test = new AsyncEventQueueTestBase(); |
| int port = AvailablePortHelper.getRandomAvailableTCPPort(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| // props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId); |
| props.setProperty(LOCATORS, "localhost[" + port + "]"); |
| props.setProperty(START_LOCATOR, |
| "localhost[" + port + "],server=true,peer=true,hostname-for-clients=localhost"); |
| test.startLocatorDistributedSystem(props); |
| return port; |
| } |
| |
| public static Integer createFirstRemoteLocator(int dsId, int remoteLocPort) { |
| AsyncEventQueueTestBase test = new AsyncEventQueueTestBase(); |
| int port = AvailablePortHelper.getRandomAvailableTCPPort(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(DISTRIBUTED_SYSTEM_ID, "" + dsId); |
| props.setProperty(LOCATORS, "localhost[" + port + "]"); |
| props.setProperty(START_LOCATOR, |
| "localhost[" + port + "],server=true,peer=true,hostname-for-clients=localhost"); |
| props.setProperty(REMOTE_LOCATORS, "localhost[" + remoteLocPort + "]"); |
| test.startLocatorDistributedSystem(props); |
| return port; |
| } |
| |
| private void startLocatorDistributedSystem(Properties props) { |
| // Start start the locator with a LOCATOR_DM_TYPE and not a NORMAL_DM_TYPE |
| System.setProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE, "true"); |
| try { |
| getSystem(props); |
| } finally { |
| System.clearProperty(InternalLocator.FORCE_LOCATOR_DM_TYPE); |
| } |
| } |
| |
| public static void createReplicatedRegionWithAsyncEventQueue(String regionName, |
| String asyncQueueIds, Boolean offHeap) { |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| AttributesFactory fact = new AttributesFactory(); |
| addAsyncEventQueueIds(fact, asyncQueueIds); |
| fact.setDataPolicy(DataPolicy.REPLICATE); |
| fact.setOffHeap(offHeap); |
| RegionFactory regionFactory = cache.createRegionFactory(fact.create()); |
| Region r = regionFactory.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp1.remove(); |
| } |
| } |
| |
| public static void createReplicatedRegionWithCacheLoaderAndAsyncEventQueue(String regionName, |
| String asyncQueueIds) { |
| |
| AttributesFactory fact = new AttributesFactory(); |
| addAsyncEventQueueIds(fact, asyncQueueIds); |
| fact.setDataPolicy(DataPolicy.REPLICATE); |
| // set the CacheLoader |
| fact.setCacheLoader(new MyCacheLoader()); |
| RegionFactory regionFactory = cache.createRegionFactory(fact.create()); |
| Region r = regionFactory.create(regionName); |
| assertNotNull(r); |
| } |
| |
| private static void addAsyncEventQueueIds(AttributesFactory fact, String asyncQueueIds) { |
| if (asyncQueueIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(asyncQueueIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String asyncQueueId = tokenizer.nextToken(); |
| fact.addAsyncEventQueueId(asyncQueueId); |
| } |
| } |
| } |
| |
| public static void createReplicatedRegionWithSenderAndAsyncEventQueue(String regionName, |
| String senderIds, String asyncChannelId, Boolean offHeap) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| |
| AttributesFactory fact = new AttributesFactory(); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| fact.setDataPolicy(DataPolicy.REPLICATE); |
| fact.setOffHeap(offHeap); |
| fact.setScope(Scope.DISTRIBUTED_ACK); |
| RegionFactory regionFactory = cache.createRegionFactory(fact.create()); |
| regionFactory.addAsyncEventQueueId(asyncChannelId); |
| Region r = regionFactory.create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| } |
| } |
| |
| public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| String diskStoreName, boolean isDiskSynchronous) { |
| createAsyncEventQueue(asyncChannelId, isParallel, maxMemory, batchSize, isConflation, |
| isPersistent, diskStoreName, isDiskSynchronous, new MyAsyncEventListener()); |
| } |
| |
| public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| String diskStoreName, boolean isDiskSynchronous, |
| final AsyncEventListener asyncEventListener) { |
| createAsyncEventQueue(asyncChannelId, isParallel, maxMemory, batchSize, isConflation, |
| isPersistent, diskStoreName, isDiskSynchronous, numDispatcherThreadsForTheRun, |
| asyncEventListener); |
| } |
| |
| public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| String diskStoreName, boolean isDiskSynchronous, int numDispatcherThreads, |
| final AsyncEventListener asyncEventListener) { |
| createDiskStore(asyncChannelId, diskStoreName); |
| |
| AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory, |
| batchSize, isPersistent, diskStoreName); |
| factory.setDiskSynchronous(isDiskSynchronous); |
| factory.setBatchConflationEnabled(isConflation); |
| // set dispatcher threads |
| factory.setDispatcherThreads(numDispatcherThreads); |
| // Set GatewayEventSubstitutionFilter |
| AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener); |
| } |
| |
| private static void createDiskStore(String asyncChannelId, String diskStoreName) { |
| if (diskStoreName != null) { |
| File directory = new File( |
| asyncChannelId + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| directory.mkdir(); |
| File[] dirs1 = new File[] {directory}; |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| dsf.setDiskDirs(dirs1); |
| DiskStore ds = dsf.create(diskStoreName); |
| } |
| } |
| |
| public static void createAsyncEventQueueWithListener2(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isPersistent, String diskStoreName) { |
| |
| createDiskStore(asyncChannelId, diskStoreName); |
| |
| AsyncEventListener asyncEventListener = new MyAsyncEventListener2(); |
| |
| AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory, |
| batchSize, isPersistent, diskStoreName); |
| // set dispatcher threads |
| factory.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener); |
| } |
| |
| public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| String diskStoreName, boolean isDiskSynchronous, String asyncListenerClass) throws Exception { |
| createAsyncEventQueue(asyncChannelId, isParallel, maxMemory, batchSize, isConflation, |
| isPersistent, diskStoreName, isDiskSynchronous, asyncListenerClass, null); |
| } |
| |
| public static void createAsyncEventQueue(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| String diskStoreName, boolean isDiskSynchronous, String asyncListenerClass, |
| String substitutionFilterClass) throws Exception { |
| |
| createDiskStore(asyncChannelId, diskStoreName); |
| |
| AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory, |
| batchSize, isPersistent, diskStoreName); |
| factory.setDiskSynchronous(isDiskSynchronous); |
| factory.setBatchConflationEnabled(isConflation); |
| if (substitutionFilterClass != null) { |
| factory.setGatewayEventSubstitutionListener( |
| (GatewayEventSubstitutionFilter) getClass(substitutionFilterClass).newInstance()); |
| } |
| // set dispatcher threads |
| factory.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| AsyncEventQueue asyncChannel = factory.create(asyncChannelId, |
| (AsyncEventListener) getClass(asyncListenerClass).newInstance()); |
| } |
| |
| private static Class getClass(String simpleClassName) throws Exception { |
| String packagePrefix = "org.apache.geode.internal.cache.wan."; |
| String className = packagePrefix + simpleClassName; |
| Class clazz = null; |
| clazz = Class.forName(className); |
| return clazz; |
| } |
| |
| public static void createAsyncEventQueueWithCustomListener(String asyncChannelId, |
| boolean isParallel, Integer maxMemory, Integer batchSize, boolean isConflation, |
| boolean isPersistent, String diskStoreName, boolean isDiskSynchronous) { |
| createAsyncEventQueueWithCustomListener(asyncChannelId, isParallel, maxMemory, batchSize, |
| isConflation, isPersistent, diskStoreName, isDiskSynchronous, |
| GatewaySender.DEFAULT_DISPATCHER_THREADS); |
| } |
| |
| public static void createAsyncEventQueueWithCustomListener(String asyncChannelId, |
| boolean isParallel, Integer maxMemory, Integer batchSize, boolean isConflation, |
| boolean isPersistent, String diskStoreName, boolean isDiskSynchronous, int nDispatchers) { |
| |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| |
| try { |
| createDiskStore(asyncChannelId, diskStoreName); |
| |
| AsyncEventListener asyncEventListener = new CustomAsyncEventListener(); |
| |
| AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory, |
| batchSize, isPersistent, diskStoreName); |
| factory.setDispatcherThreads(nDispatchers); |
| AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener); |
| } finally { |
| exp.remove(); |
| } |
| } |
| |
| private static AsyncEventQueueFactory getInitialAsyncEventQueueFactory(boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isPersistent, String diskStoreName) { |
| AsyncEventQueueFactory factory = cache.createAsyncEventQueueFactory(); |
| factory.setBatchSize(batchSize); |
| factory.setPersistent(isPersistent); |
| factory.setDiskStoreName(diskStoreName); |
| factory.setMaximumQueueMemory(maxMemory); |
| factory.setParallel(isParallel); |
| return factory; |
| } |
| |
| public static void createConcurrentAsyncEventQueue(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| String diskStoreName, boolean isDiskSynchronous, int dispatcherThreads, OrderPolicy policy) { |
| |
| createDiskStore(asyncChannelId, diskStoreName); |
| |
| AsyncEventListener asyncEventListener = new MyAsyncEventListener(); |
| |
| AsyncEventQueueFactory factory = getInitialAsyncEventQueueFactory(isParallel, maxMemory, |
| batchSize, isPersistent, diskStoreName); |
| factory.setDiskSynchronous(isDiskSynchronous); |
| factory.setBatchConflationEnabled(isConflation); |
| factory.setOrderPolicy(policy); |
| AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener); |
| } |
| |
| public static String createAsyncEventQueueWithDiskStore(String asyncChannelId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isPersistent, String diskStoreName) { |
| |
| AsyncEventListener asyncEventListener = new MyAsyncEventListener(); |
| |
| File persistentDirectory = null; |
| if (diskStoreName == null) { |
| persistentDirectory = new File( |
| asyncChannelId + "_disk_" + System.currentTimeMillis() + "_" + VM.getCurrentVMNum()); |
| } else { |
| persistentDirectory = new File(diskStoreName); |
| } |
| LogWriterUtils.getLogWriter().info("The ds is : " + persistentDirectory.getName()); |
| persistentDirectory.mkdir(); |
| DiskStoreFactory dsf = cache.createDiskStoreFactory(); |
| File[] dirs1 = new File[] {persistentDirectory}; |
| |
| AsyncEventQueueFactory factory = cache.createAsyncEventQueueFactory(); |
| factory.setBatchSize(batchSize); |
| factory.setParallel(isParallel); |
| if (isPersistent) { |
| factory.setPersistent(isPersistent); |
| factory.setDiskStoreName(dsf.setDiskDirs(dirs1).create(asyncChannelId).getName()); |
| } |
| factory.setMaximumQueueMemory(maxMemory); |
| // set dispatcher threads |
| factory.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| AsyncEventQueue asyncChannel = factory.create(asyncChannelId, asyncEventListener); |
| return persistentDirectory.getName(); |
| } |
| |
| public static void pauseAsyncEventQueue(String asyncChannelId) { |
| AsyncEventQueue theChannel = null; |
| |
| Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncChannel : asyncEventChannels) { |
| if (asyncChannelId.equals(asyncChannel.getId())) { |
| theChannel = asyncChannel; |
| } |
| } |
| |
| ((AsyncEventQueueImpl) theChannel).getSender().pause(); |
| } |
| |
| public static void pauseAsyncEventQueueAndWaitForDispatcherToPause(String asyncChannelId) { |
| AsyncEventQueue theChannel = null; |
| |
| Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncChannel : asyncEventChannels) { |
| if (asyncChannelId.equals(asyncChannel.getId())) { |
| theChannel = asyncChannel; |
| break; |
| } |
| } |
| |
| ((AsyncEventQueueImpl) theChannel).getSender().pause(); |
| |
| ((AsyncEventQueueImpl) theChannel).getSender().getEventProcessor() |
| .waitForDispatcherToPause(); |
| } |
| |
| public static void resumeAsyncEventQueue(String asyncQueueId) { |
| AsyncEventQueue theQueue = null; |
| |
| Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncChannel : asyncEventChannels) { |
| if (asyncQueueId.equals(asyncChannel.getId())) { |
| theQueue = asyncChannel; |
| } |
| } |
| |
| ((AsyncEventQueueImpl) theQueue).getSender().resume(); |
| } |
| |
| public static void waitForAsyncEventQueueSize(String senderId, int numQueueEntries, |
| boolean localSize) { |
| await() |
| .untilAsserted(() -> checkAsyncEventQueueSize(senderId, numQueueEntries, localSize)); |
| } |
| |
| public static void checkAsyncEventQueueSize(String asyncQueueId, int numQueueEntries) { |
| checkAsyncEventQueueSize(asyncQueueId, numQueueEntries, false); |
| } |
| |
| public static void checkAsyncEventQueueSize(String asyncQueueId, int numQueueEntries, |
| boolean localSize) { |
| AsyncEventQueueImpl aeq = (AsyncEventQueueImpl) cache.getAsyncEventQueue(asyncQueueId); |
| GatewaySender sender = aeq.getSender(); |
| |
| if (sender.isParallel()) { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| Region queueRegion = queues.toArray(new RegionQueue[queues.size()])[0].getRegion(); |
| if (localSize) { |
| queueRegion = PartitionRegionHelper.getLocalData(queueRegion); |
| } |
| assertEquals(numQueueEntries, queueRegion.size()); |
| } else { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| assertEquals(numQueueEntries, size); |
| } |
| } |
| |
| public static void createPartitionedRegion(String regionName, String senderIds, |
| Integer redundantCopies, Integer totalNumBuckets) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(PartitionOfflineException.class.getName()); |
| try { |
| AttributesFactory fact = new AttributesFactory(); |
| if (senderIds != null) { |
| StringTokenizer tokenizer = new StringTokenizer(senderIds, ","); |
| while (tokenizer.hasMoreTokens()) { |
| String senderId = tokenizer.nextToken(); |
| // GatewaySender sender = cache.getGatewaySender(senderId); |
| // assertNotNull(sender); |
| fact.addGatewaySenderId(senderId); |
| } |
| } |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setRedundantCopies(redundantCopies); |
| pfact.setRecoveryDelay(0); |
| fact.setPartitionAttributes(pfact.create()); |
| Region r = cache.createRegionFactory(fact.create()).create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| public static void createPartitionedRegionWithAsyncEventQueue(String regionName, |
| String asyncEventQueueId, Boolean offHeap) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(PartitionOfflineException.class.getName()); |
| try { |
| AttributesFactory fact = new AttributesFactory(); |
| |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(16); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId) |
| .create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| public static void createFixedPartitionedRegionWithAsyncEventQueue(String regionName, |
| String asyncEventQueueId, String partitionName, final List<String> allPartitions, |
| boolean offHeap) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(PartitionOfflineException.class.getName()); |
| try { |
| AttributesFactory fact = new AttributesFactory(); |
| |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(16); |
| pfact.addFixedPartitionAttributes( |
| FixedPartitionAttributes.createFixedPartition(partitionName, true)); |
| pfact.setPartitionResolver(new MyFixedPartitionResolver(allPartitions)); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId) |
| .create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| public static void createColocatedPartitionedRegionWithAsyncEventQueue(String regionName, |
| String asyncEventQueueId, Integer totalNumBuckets, String colocatedWith) { |
| |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(PartitionOfflineException.class.getName()); |
| try { |
| AttributesFactory fact = new AttributesFactory(); |
| |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(totalNumBuckets); |
| pfact.setColocatedWith(colocatedWith); |
| fact.setPartitionAttributes(pfact.create()); |
| Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId) |
| .create(regionName); |
| assertNotNull(r); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| public static void createPartitionedRegionWithCacheLoaderAndAsyncQueue(String regionName, |
| String asyncEventQueueId) { |
| |
| AttributesFactory fact = new AttributesFactory(); |
| |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(16); |
| fact.setPartitionAttributes(pfact.create()); |
| // set the CacheLoader implementation |
| fact.setCacheLoader(new MyCacheLoader()); |
| Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId) |
| .create(regionName); |
| assertNotNull(r); |
| } |
| |
| /** |
| * Create PartitionedRegion with 1 redundant copy |
| */ |
| public static void createPRWithRedundantCopyWithAsyncEventQueue(String regionName, |
| String asyncEventQueueId, Boolean offHeap) throws InterruptedException { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| |
| |
| CountDownLatch recoveryDone = new CountDownLatch(2); |
| |
| ResourceObserver observer = new InternalResourceManager.ResourceObserverAdapter() { |
| @Override |
| public void recoveryFinished(Region region) { |
| recoveryDone.countDown(); |
| } |
| }; |
| InternalResourceManager.setResourceObserver(observer); |
| |
| |
| try { |
| AttributesFactory fact = new AttributesFactory(); |
| |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(16); |
| pfact.setRedundantCopies(1); |
| fact.setPartitionAttributes(pfact.create()); |
| fact.setOffHeap(offHeap); |
| Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId) |
| .create(regionName); |
| assertNotNull(r); |
| recoveryDone.await(); |
| } finally { |
| exp.remove(); |
| } |
| } |
| |
| public static void createPartitionedRegionAccessorWithAsyncEventQueue(String regionName, |
| String asyncEventQueueId) { |
| AttributesFactory fact = new AttributesFactory(); |
| PartitionAttributesFactory pfact = new PartitionAttributesFactory(); |
| pfact.setTotalNumBuckets(16); |
| pfact.setLocalMaxMemory(0); |
| fact.setPartitionAttributes(pfact.create()); |
| Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId(asyncEventQueueId) |
| .create(regionName); |
| // fact.create()).create(regionName); |
| assertNotNull(r); |
| } |
| |
| protected static void createCache(Integer locPort) { |
| AsyncEventQueueTestBase test = new AsyncEventQueueTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| } |
| |
| public static void createCacheWithoutLocator(Integer mCastPort) { |
| AsyncEventQueueTestBase test = new AsyncEventQueueTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "" + mCastPort); |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| } |
| |
| public static void checkAsyncEventQueueStats(String queueId, final int queueSize, |
| int secondaryQueueSize, final int eventsReceived, final int eventsQueued, |
| final int eventsDistributed) { |
| Set<AsyncEventQueue> asyncQueues = cache.getAsyncEventQueues(); |
| AsyncEventQueue queue = null; |
| boolean isParallel = false; |
| for (AsyncEventQueue q : asyncQueues) { |
| isParallel = q.isParallel(); |
| if (q.getId().equals(queueId)) { |
| queue = q; |
| break; |
| } |
| } |
| final AsyncEventQueueStats statistics = ((AsyncEventQueueImpl) queue).getStatistics(); |
| await() |
| .untilAsserted( |
| () -> assertEquals("Expected queue entries: " + queueSize + " but actual entries: " |
| + statistics.getEventQueueSize(), queueSize, statistics.getEventQueueSize())); |
| if (isParallel) { |
| await().untilAsserted(() -> { |
| assertEquals( |
| "Expected events in the secondary queue is " + secondaryQueueSize + ", but actual is " |
| + statistics.getSecondaryEventQueueSize(), |
| secondaryQueueSize, statistics.getSecondaryEventQueueSize()); |
| }); |
| } else { |
| // for serial queue, evenvSecondaryQueueSize is not used |
| assertEquals(0, statistics.getSecondaryEventQueueSize()); |
| } |
| assertEquals(queueSize, statistics.getEventQueueSize()); |
| assertEquals(eventsReceived, statistics.getEventsReceived()); |
| assertEquals(eventsQueued, statistics.getEventsQueued()); |
| assert (statistics.getEventsDistributed() >= eventsDistributed); |
| } |
| |
| public static void checkAsyncEventQueueConflatedStats(String asyncEventQueueId, |
| final int eventsConflated) { |
| Set<AsyncEventQueue> queues = cache.getAsyncEventQueues(); |
| AsyncEventQueue queue = null; |
| for (AsyncEventQueue q : queues) { |
| if (q.getId().equals(asyncEventQueueId)) { |
| queue = q; |
| break; |
| } |
| } |
| final AsyncEventQueueStats statistics = ((AsyncEventQueueImpl) queue).getStatistics(); |
| assertEquals(eventsConflated, statistics.getEventsNotQueuedConflated()); |
| } |
| |
| public static void checkAsyncEventQueueStats_Failover(String asyncEventQueueId, |
| final int eventsReceived) { |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| AsyncEventQueue queue = null; |
| for (AsyncEventQueue q : asyncEventQueues) { |
| if (q.getId().equals(asyncEventQueueId)) { |
| queue = q; |
| break; |
| } |
| } |
| final AsyncEventQueueStats statistics = ((AsyncEventQueueImpl) queue).getStatistics(); |
| |
| assertEquals(eventsReceived, statistics.getEventsReceived()); |
| assertEquals(eventsReceived, |
| (statistics.getEventsQueued() + statistics.getUnprocessedTokensAddedByPrimary() |
| + statistics.getUnprocessedEventsRemovedByPrimary())); |
| } |
| |
| public static void checkAsyncEventQueueBatchStats(String asyncQueueId, final int batches) { |
| Set<AsyncEventQueue> queues = cache.getAsyncEventQueues(); |
| AsyncEventQueue queue = null; |
| for (AsyncEventQueue q : queues) { |
| if (q.getId().equals(asyncQueueId)) { |
| queue = q; |
| break; |
| } |
| } |
| final AsyncEventQueueStats statistics = ((AsyncEventQueueImpl) queue).getStatistics(); |
| assert (statistics.getBatchesDistributed() >= batches); |
| assertEquals(0, statistics.getBatchesRedistributed()); |
| } |
| |
| public static void checkAsyncEventQueueUnprocessedStats(String asyncQueueId, int events) { |
| Set<AsyncEventQueue> asyncQueues = cache.getAsyncEventQueues(); |
| AsyncEventQueue queue = null; |
| for (AsyncEventQueue q : asyncQueues) { |
| if (q.getId().equals(asyncQueueId)) { |
| queue = q; |
| break; |
| } |
| } |
| final AsyncEventQueueStats statistics = ((AsyncEventQueueImpl) queue).getStatistics(); |
| assertEquals(events, (statistics.getUnprocessedEventsAddedBySecondary() |
| + statistics.getUnprocessedTokensRemovedBySecondary())); |
| assertEquals(events, (statistics.getUnprocessedEventsRemovedByPrimary() |
| + statistics.getUnprocessedTokensAddedByPrimary())); |
| } |
| |
| public static void setRemoveFromQueueOnException(String senderId, boolean removeFromQueue) { |
| Set<GatewaySender> senders = cache.getGatewaySenders(); |
| GatewaySender sender = null; |
| for (GatewaySender s : senders) { |
| if (s.getId().equals(senderId)) { |
| sender = s; |
| break; |
| } |
| } |
| assertNotNull(sender); |
| ((AbstractGatewaySender) sender).setRemoveFromQueueOnException(removeFromQueue); |
| } |
| |
| public static void unsetRemoveFromQueueOnException(String senderId) { |
| Set<GatewaySender> senders = cache.getGatewaySenders(); |
| GatewaySender sender = null; |
| for (GatewaySender s : senders) { |
| if (s.getId().equals(senderId)) { |
| sender = s; |
| break; |
| } |
| } |
| assertNotNull(sender); |
| ((AbstractGatewaySender) sender).setRemoveFromQueueOnException(false); |
| } |
| |
| public static void waitForSenderToBecomePrimary(String senderId) { |
| Set<GatewaySender> senders = ((GemFireCacheImpl) cache).getAllGatewaySenders(); |
| final GatewaySender sender = getGatewaySenderById(senders, senderId); |
| WaitCriterion wc = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return sender != null && ((AbstractGatewaySender) sender).isPrimary(); |
| } |
| |
| @Override |
| public String description() { |
| return "Expected sender primary state to be true but is false"; |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(wc); |
| } |
| |
| private static GatewaySender getGatewaySenderById(Set<GatewaySender> senders, String senderId) { |
| for (GatewaySender s : senders) { |
| if (s.getId().equals(senderId)) { |
| return s; |
| } |
| } |
| // if none of the senders matches with the supplied senderid, return null |
| return null; |
| } |
| |
| public static void createSender(String dsName, int remoteDsId, boolean isParallel, |
| Integer maxMemory, Integer batchSize, boolean isConflation, boolean isPersistent, |
| GatewayEventFilter filter, boolean isManulaStart) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| try { |
| 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); |
| gateway.setManualStart(isManulaStart); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| ((InternalGatewaySenderFactory) gateway) |
| .setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| if (filter != null) { |
| eventFilter = filter; |
| gateway.addGatewayEventFilter(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); |
| // set dispatcher threads |
| gateway.setDispatcherThreads(numDispatcherThreadsForTheRun); |
| ((InternalGatewaySenderFactory) gateway) |
| .setLocatorDiscoveryCallback(new MyLocatorCallback()); |
| if (filter != null) { |
| eventFilter = filter; |
| gateway.addGatewayEventFilter(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); |
| } |
| } finally { |
| exln.remove(); |
| } |
| } |
| |
| public static int createReceiver(int locPort) { |
| AsyncEventQueueTestBase test = new AsyncEventQueueTestBase(); |
| Properties props = test.getDistributedSystemProperties(); |
| props.setProperty(MCAST_PORT, "0"); |
| props.setProperty(LOCATORS, "localhost[" + locPort + "]"); |
| |
| InternalDistributedSystem ds = test.getSystem(props); |
| cache = CacheFactory.create(ds); |
| GatewayReceiverFactory fact = cache.createGatewayReceiverFactory(); |
| int port = AvailablePortHelper.getRandomAvailableTCPPort(); |
| fact.setStartPort(port); |
| fact.setEndPort(port); |
| fact.setManualStart(true); |
| 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 String makePath(String[] strings) { |
| StringBuilder sb = new StringBuilder(); |
| for (final String string : strings) { |
| sb.append(string); |
| sb.append(File.separator); |
| } |
| return sb.toString(); |
| } |
| |
| /** |
| * Do a rebalance and verify balance was improved. If evictionPercentage > 0 (the default) then we |
| * have heapLRU and this can cause simulate and rebalance results to differ if eviction kicks in |
| * between. (See BUG 44899). |
| */ |
| public static void doRebalance() { |
| ResourceManager resMan = cache.getResourceManager(); |
| boolean heapEviction = (resMan.getEvictionHeapPercentage() > 0); |
| RebalanceFactory factory = resMan.createRebalanceFactory(); |
| try { |
| RebalanceResults simulateResults = null; |
| if (!heapEviction) { |
| LogWriterUtils.getLogWriter().info("Calling rebalance simulate"); |
| RebalanceOperation simulateOp = factory.simulate(); |
| simulateResults = simulateOp.getResults(); |
| } |
| |
| LogWriterUtils.getLogWriter().info("Starting rebalancing"); |
| RebalanceOperation rebalanceOp = factory.start(); |
| RebalanceResults rebalanceResults = rebalanceOp.getResults(); |
| |
| } catch (InterruptedException e) { |
| Assert.fail("Interrupted", e); |
| } |
| } |
| |
| public static void doPuts(String regionName, int numPuts) { |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(InterruptedException.class.getName()); |
| IgnoredException exp2 = |
| IgnoredException.addIgnoredException(GatewaySenderException.class.getName()); |
| try { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numPuts; i++) { |
| r.put(i, i); |
| } |
| } finally { |
| exp1.remove(); |
| exp2.remove(); |
| } |
| // for (long i = 0; i < numPuts; i++) { |
| // r.destroy(i); |
| // } |
| } |
| |
| public static void doHeavyPuts(String regionName, int numPuts) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numPuts; i++) { |
| r.put(i, new byte[1024 * 1024]); |
| } |
| } |
| |
| /** |
| * To be used for CacheLoader related tests |
| */ |
| public static void doGets(String regionName, int numGets) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numGets; i++) { |
| r.get(i); |
| } |
| } |
| |
| public static void doPutsFrom(String regionName, int from, int numPuts) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = from; i < numPuts; i++) { |
| r.put(i, i); |
| } |
| } |
| |
| public static void doPutAll(String regionName, int numPuts, int size) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = 0; i < numPuts; i++) { |
| Map putAllMap = new HashMap(); |
| for (long j = 0; j < size; j++) { |
| putAllMap.put((size * i) + j, i); |
| } |
| r.putAll(putAllMap, "putAllCallback"); |
| putAllMap.clear(); |
| } |
| } |
| |
| public static void putGivenKeyValue(String regionName, Map keyValues) { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (Object key : keyValues.keySet()) { |
| r.put(key, keyValues.get(key)); |
| } |
| } |
| |
| public static void doNextPuts(String regionName, int start, int numPuts) { |
| // waitForSitesToUpdate(); |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(CacheClosedException.class.getName()); |
| try { |
| Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| for (long i = start; i < numPuts; i++) { |
| r.put(i, i); |
| } |
| } finally { |
| exp.remove(); |
| } |
| } |
| |
| public static void validateRegionSize(String regionName, final int regionSize) { |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(CacheClosedException.class.getName()); |
| try { |
| |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| WaitCriterion wc = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return r.keySet().size() == regionSize; |
| } |
| |
| @Override |
| public String description() { |
| return "Expected region entries: " + regionSize + " but actual entries: " |
| + r.keySet().size() + " present region keyset " + r.keySet(); |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(wc); |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| } |
| } |
| |
| /** |
| * Validate whether all the attributes set on AsyncEventQueueFactory are set on the sender |
| * underneath the AsyncEventQueue. |
| */ |
| public static void validateAsyncEventQueueAttributes(String asyncChannelId, int maxQueueMemory, |
| int batchSize, int batchTimeInterval, boolean isPersistent, String diskStoreName, |
| boolean isDiskSynchronous, boolean batchConflationEnabled) { |
| |
| AsyncEventQueue theChannel = null; |
| |
| Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncChannel : asyncEventChannels) { |
| if (asyncChannelId.equals(asyncChannel.getId())) { |
| theChannel = asyncChannel; |
| } |
| } |
| |
| GatewaySender theSender = ((AsyncEventQueueImpl) theChannel).getSender(); |
| assertEquals("maxQueueMemory", maxQueueMemory, theSender.getMaximumQueueMemory()); |
| assertEquals("batchSize", batchSize, theSender.getBatchSize()); |
| assertEquals("batchTimeInterval", batchTimeInterval, theSender.getBatchTimeInterval()); |
| assertEquals("isPersistent", isPersistent, theSender.isPersistenceEnabled()); |
| assertEquals("diskStoreName", diskStoreName, theSender.getDiskStoreName()); |
| assertEquals("isDiskSynchronous", isDiskSynchronous, theSender.isDiskSynchronous()); |
| assertEquals("batchConflation", batchConflationEnabled, theSender.isBatchConflationEnabled()); |
| } |
| |
| /** |
| * Validate whether all the attributes set on AsyncEventQueueFactory are set on the sender |
| * underneath the AsyncEventQueue. |
| */ |
| public static void validateConcurrentAsyncEventQueueAttributes(String asyncChannelId, |
| int maxQueueMemory, int batchSize, int batchTimeInterval, boolean isPersistent, |
| String diskStoreName, boolean isDiskSynchronous, boolean batchConflationEnabled, |
| int dispatcherThreads, OrderPolicy policy) { |
| |
| AsyncEventQueue theChannel = null; |
| |
| Set<AsyncEventQueue> asyncEventChannels = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncChannel : asyncEventChannels) { |
| if (asyncChannelId.equals(asyncChannel.getId())) { |
| theChannel = asyncChannel; |
| } |
| } |
| |
| GatewaySender theSender = ((AsyncEventQueueImpl) theChannel).getSender(); |
| assertEquals("maxQueueMemory", maxQueueMemory, theSender.getMaximumQueueMemory()); |
| assertEquals("batchSize", batchSize, theSender.getBatchSize()); |
| assertEquals("batchTimeInterval", batchTimeInterval, theSender.getBatchTimeInterval()); |
| assertEquals("isPersistent", isPersistent, theSender.isPersistenceEnabled()); |
| assertEquals("diskStoreName", diskStoreName, theSender.getDiskStoreName()); |
| assertEquals("isDiskSynchronous", isDiskSynchronous, theSender.isDiskSynchronous()); |
| assertEquals("batchConflation", batchConflationEnabled, theSender.isBatchConflationEnabled()); |
| assertEquals("dispatcherThreads", dispatcherThreads, theSender.getDispatcherThreads()); |
| assertEquals("orderPolicy", policy, theSender.getOrderPolicy()); |
| } |
| |
| public static void validateAsyncEventListener(String asyncQueueId, final int expectedSize) { |
| AsyncEventListener theListener = null; |
| |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncQueue : asyncEventQueues) { |
| if (asyncQueueId.equals(asyncQueue.getId())) { |
| theListener = asyncQueue.getAsyncEventListener(); |
| } |
| } |
| |
| final Map eventsMap = ((MyAsyncEventListener) theListener).getEventsMap(); |
| assertNotNull(eventsMap); |
| WaitCriterion wc = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return eventsMap.size() == expectedSize; |
| } |
| |
| @Override |
| public String description() { |
| return "Expected map entries: " + expectedSize + " but actual entries: " + eventsMap.size(); |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(wc); // TODO:Yogs |
| } |
| |
| public static void validateAsyncEventForOperationDetail(String asyncQueueId, |
| final int expectedSize, boolean isLoad, boolean isPutAll) { |
| |
| AsyncEventListener theListener = null; |
| |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncQueue : asyncEventQueues) { |
| if (asyncQueueId.equals(asyncQueue.getId())) { |
| theListener = asyncQueue.getAsyncEventListener(); |
| } |
| } |
| |
| final Map eventsMap = ((MyAsyncEventListener_CacheLoader) theListener).getEventsMap(); |
| assertNotNull(eventsMap); |
| WaitCriterion wc = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return eventsMap.size() == expectedSize; |
| } |
| |
| @Override |
| public String description() { |
| return "Expected map entries: " + expectedSize + " but actual entries: " + eventsMap.size(); |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(wc); // TODO:Yogs |
| Collection values = eventsMap.values(); |
| for (final Object value : values) { |
| AsyncEvent asyncEvent = (AsyncEvent) value; |
| if (isLoad) { |
| assertTrue(asyncEvent.getOperation().isLoad()); |
| } |
| if (isPutAll) { |
| assertTrue(asyncEvent.getOperation().isPutAll()); |
| } |
| } |
| } |
| |
| public static void validateCustomAsyncEventListener(String asyncQueueId, final int expectedSize) { |
| AsyncEventListener theListener = null; |
| |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncQueue : asyncEventQueues) { |
| if (asyncQueueId.equals(asyncQueue.getId())) { |
| theListener = asyncQueue.getAsyncEventListener(); |
| } |
| } |
| |
| final Map eventsMap = ((CustomAsyncEventListener) theListener).getEventsMap(); |
| assertNotNull(eventsMap); |
| WaitCriterion wc = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| return eventsMap.size() == expectedSize; |
| } |
| |
| @Override |
| public String description() { |
| return "Expected map entries: " + expectedSize + " but actual entries: " + eventsMap.size(); |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(wc); // TODO:Yogs |
| |
| for (final AsyncEvent event : (Iterable<AsyncEvent>) eventsMap.values()) { |
| assertTrue("possibleDuplicate should be true for event: " + event, |
| event.getPossibleDuplicate()); |
| } |
| } |
| |
| public static void waitForAsyncQueueToGetEmpty(String asyncQueueId) { |
| AsyncEventQueue theAsyncEventQueue = cache.getAsyncEventQueue(asyncQueueId); |
| final GatewaySender sender = ((AsyncEventQueueImpl) theAsyncEventQueue).getSender(); |
| |
| if (sender.isParallel()) { |
| final Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| |
| WaitCriterion wc = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| return size == 0; |
| } |
| |
| @Override |
| public String description() { |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| return "Expected queue size to be : " + 0 + " but actual entries: " + size; |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(wc); |
| |
| } else { |
| WaitCriterion wc = new WaitCriterion() { |
| @Override |
| public boolean done() { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| return size == 0; |
| } |
| |
| @Override |
| public String description() { |
| Set<RegionQueue> queues = ((AbstractGatewaySender) sender).getQueues(); |
| int size = 0; |
| for (RegionQueue q : queues) { |
| size += q.size(); |
| } |
| return "Expected queue size to be : " + 0 + " but actual entries: " + size; |
| } |
| }; |
| GeodeAwaitility.await().untilAsserted(wc); |
| } |
| } |
| |
| public static void verifyAsyncEventListenerForPossibleDuplicates(String asyncEventQueueId, |
| Set<Integer> bucketIds, int batchSize) { |
| AsyncEventListener theListener = null; |
| |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncQueue : asyncEventQueues) { |
| if (asyncEventQueueId.equals(asyncQueue.getId())) { |
| theListener = asyncQueue.getAsyncEventListener(); |
| } |
| } |
| |
| final Map<Integer, List<GatewaySenderEventImpl>> bucketToEventsMap = |
| ((MyAsyncEventListener2) theListener).getBucketToEventsMap(); |
| assertNotNull(bucketToEventsMap); |
| assertTrue(bucketIds.size() > 1); |
| |
| for (int bucketId : bucketIds) { |
| List<GatewaySenderEventImpl> eventsForBucket = bucketToEventsMap.get(bucketId); |
| LogWriterUtils.getLogWriter() |
| .info("Events for bucket: " + bucketId + " is " + eventsForBucket); |
| assertNotNull(eventsForBucket); |
| for (int i = 0; i < batchSize; i++) { |
| GatewaySenderEventImpl senderEvent = eventsForBucket.get(i); |
| assertTrue(senderEvent.getPossibleDuplicate()); |
| } |
| } |
| } |
| |
| public static void verifySubstitutionFilterInvocations(String asyncEventQueueId, |
| int expectedNumInvocations) { |
| AsyncEventQueue queue = cache.getAsyncEventQueue(asyncEventQueueId); |
| assertNotNull(queue); |
| |
| // Verify the GatewayEventSubstitutionFilter has been invoked the appropriate number of times |
| MyGatewayEventSubstitutionFilter filter = |
| (MyGatewayEventSubstitutionFilter) queue.getGatewayEventSubstitutionFilter(); |
| assertNotNull(filter); |
| assertEquals(expectedNumInvocations, filter.getNumInvocations()); |
| |
| // Verify the AsyncEventListener has received the substituted values |
| MyAsyncEventListener listener = (MyAsyncEventListener) queue.getAsyncEventListener(); |
| final Map eventsMap = listener.getEventsMap(); |
| assertNotNull(eventsMap); |
| assertEquals(expectedNumInvocations, eventsMap.size()); |
| |
| for (final Object o : eventsMap.entrySet()) { |
| Map.Entry<Integer, String> entry = (Map.Entry<Integer, String>) o; |
| assertEquals(MyGatewayEventSubstitutionFilter.SUBSTITUTION_PREFIX + entry.getKey(), |
| entry.getValue()); |
| } |
| } |
| |
| |
| public static void verifySubstitutionFilterToDataInvocations(String asyncEventQueueId, |
| int expectedToDataInvoations) { |
| AsyncEventQueue queue = cache.getAsyncEventQueue(asyncEventQueueId); |
| assertNotNull(queue); |
| |
| // Verify the GatewayEventSubstitutionFilter has been invoked the appropriate number of times |
| SizeableGatewayEventSubstitutionFilter filter = |
| (SizeableGatewayEventSubstitutionFilter) queue.getGatewayEventSubstitutionFilter(); |
| assertNotNull(filter); |
| assertEquals(expectedToDataInvoations, filter.getNumToDataInvocations()); |
| } |
| |
| public static AsyncEventListener getAsyncEventListener(String asyncEventQueueId) { |
| AsyncEventListener theListener = null; |
| |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncQueue : asyncEventQueues) { |
| if (asyncEventQueueId.equals(asyncQueue.getId())) { |
| return asyncQueue.getAsyncEventListener(); |
| } |
| } |
| return null; |
| } |
| |
| public static int getAsyncEventListenerMapSize(String asyncEventQueueId) { |
| AsyncEventListener theListener = getAsyncEventListener(asyncEventQueueId); |
| |
| final Map eventsMap = ((MyAsyncEventListener) theListener).getEventsMap(); |
| assertNotNull(eventsMap); |
| LogWriterUtils.getLogWriter().info("The events map size is " + eventsMap.size()); |
| return eventsMap.size(); |
| } |
| |
| public static int getAsyncEventQueueSize(String asyncEventQueueId) { |
| AsyncEventQueue theQueue = null; |
| |
| Set<AsyncEventQueue> asyncEventQueues = cache.getAsyncEventQueues(); |
| for (AsyncEventQueue asyncQueue : asyncEventQueues) { |
| if (asyncEventQueueId.equals(asyncQueue.getId())) { |
| theQueue = asyncQueue; |
| } |
| } |
| assertNotNull(theQueue); |
| return theQueue.size(); |
| } |
| |
| public static String getRegionFullPath(String regionName) { |
| final Region r = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(r); |
| return r.getFullPath(); |
| } |
| |
| public static Set<Integer> getAllPrimaryBucketsOnTheNode(String regionName) { |
| PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName); |
| return region.getDataStore().getAllLocalPrimaryBucketIds(); |
| } |
| |
| public static void addCacheListenerAndCloseCache(String regionName) { |
| final Region region = cache.getRegion(SEPARATOR + regionName); |
| assertNotNull(region); |
| CacheListenerAdapter cl = new CacheListenerAdapter() { |
| @Override |
| public void afterCreate(EntryEvent event) { |
| if ((Long) event.getKey() == 900) { |
| cache.getLogger().fine(" Gateway sender is killed by a test"); |
| cache.close(); |
| cache.getDistributedSystem().disconnect(); |
| } |
| } |
| }; |
| region.getAttributesMutator().addCacheListener(cl); |
| } |
| |
| public static Boolean killSender(String senderId) { |
| final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect"); |
| IgnoredException exp = |
| IgnoredException.addIgnoredException(CacheClosedException.class.getName()); |
| IgnoredException exp1 = |
| IgnoredException.addIgnoredException(ForceReattemptException.class.getName()); |
| try { |
| Set<GatewaySender> senders = cache.getGatewaySenders(); |
| AbstractGatewaySender sender = null; |
| for (GatewaySender s : senders) { |
| if (s.getId().equals(senderId)) { |
| sender = (AbstractGatewaySender) s; |
| break; |
| } |
| } |
| if (sender.isPrimary()) { |
| LogWriterUtils.getLogWriter().info("Gateway sender is killed by a test"); |
| cache.getDistributedSystem().disconnect(); |
| return Boolean.TRUE; |
| } |
| return Boolean.FALSE; |
| } finally { |
| exp.remove(); |
| exp1.remove(); |
| exln.remove(); |
| } |
| } |
| |
| public static Boolean killAsyncEventQueue(String asyncQueueId) { |
| Set<AsyncEventQueue> queues = cache.getAsyncEventQueues(); |
| AsyncEventQueueImpl queue = null; |
| for (AsyncEventQueue q : queues) { |
| if (q.getId().equals(asyncQueueId)) { |
| queue = (AsyncEventQueueImpl) q; |
| break; |
| } |
| } |
| if (queue.isPrimary()) { |
| LogWriterUtils.getLogWriter().info("AsyncEventQueue is killed by a test"); |
| cache.getDistributedSystem().disconnect(); |
| return Boolean.TRUE; |
| } |
| return Boolean.FALSE; |
| } |
| |
| public static void killSender() { |
| LogWriterUtils.getLogWriter().info("Gateway sender is going to be killed by a test"); |
| cache.close(); |
| cache.getDistributedSystem().disconnect(); |
| LogWriterUtils.getLogWriter().info("Gateway sender is killed by a test"); |
| } |
| |
| public static class MyLocatorCallback extends LocatorDiscoveryCallbackAdapter { |
| |
| private final Set discoveredLocators = new HashSet(); |
| |
| private final Set removedLocators = new HashSet(); |
| |
| @Override |
| public synchronized void locatorsDiscovered(List locators) { |
| discoveredLocators.addAll(locators); |
| notifyAll(); |
| } |
| |
| @Override |
| public synchronized void locatorsRemoved(List locators) { |
| removedLocators.addAll(locators); |
| notifyAll(); |
| } |
| |
| public boolean waitForDiscovery(InetSocketAddress locator, long time) |
| throws InterruptedException { |
| return waitFor(discoveredLocators, locator, time); |
| } |
| |
| public boolean waitForRemove(InetSocketAddress locator, long time) throws InterruptedException { |
| return waitFor(removedLocators, locator, time); |
| } |
| |
| private synchronized boolean waitFor(Set set, InetSocketAddress locator, long time) |
| throws InterruptedException { |
| long remaining = time; |
| long endTime = System.currentTimeMillis() + time; |
| while (!set.contains(locator) && remaining >= 0) { |
| wait(remaining); |
| remaining = endTime - System.currentTimeMillis(); |
| } |
| return set.contains(locator); |
| } |
| |
| public synchronized Set getDiscovered() { |
| return new HashSet(discoveredLocators); |
| } |
| |
| public synchronized Set getRemoved() { |
| return new HashSet(removedLocators); |
| } |
| } |
| |
| @Override |
| public final void postTearDown() throws Exception { |
| cleanupVM(); |
| vm0.invoke(AsyncEventQueueTestBase::cleanupVM); |
| vm1.invoke(AsyncEventQueueTestBase::cleanupVM); |
| vm2.invoke(AsyncEventQueueTestBase::cleanupVM); |
| vm3.invoke(AsyncEventQueueTestBase::cleanupVM); |
| vm4.invoke(AsyncEventQueueTestBase::cleanupVM); |
| } |
| |
| public static void cleanupVM() throws IOException { |
| closeCache(); |
| JUnit4DistributedTestCase.cleanDiskDirs(); |
| } |
| |
| public static void closeCache() throws IOException { |
| if (cache != null && !cache.isClosed()) { |
| cache.close(); |
| cache.getDistributedSystem().disconnect(); |
| cache = null; |
| } else { |
| AsyncEventQueueTestBase test = new AsyncEventQueueTestBase(); |
| if (test.isConnectedToDS()) { |
| test.getSystem().disconnect(); |
| } |
| } |
| } |
| |
| public static void shutdownLocator() { |
| AsyncEventQueueTestBase test = new AsyncEventQueueTestBase(); |
| test.getSystem().disconnect(); |
| } |
| |
| public static void printEventListenerMap() { |
| ((MyGatewaySenderEventListener) eventListener1).printMap(); |
| } |
| |
| @Override |
| public Properties getDistributedSystemProperties() { |
| // For now all WANTestBase tests allocate off-heap memory even though |
| // many of them never use it. |
| // The problem is that WANTestBase has static methods that create instances |
| // of WANTestBase (instead of instances of the subclass). So we can't override |
| // this method so that only the off-heap subclasses allocate off heap memory. |
| Properties props = new Properties(); |
| props.setProperty(OFF_HEAP_MEMORY_SIZE, "300m"); |
| return props; |
| } |
| |
| /** |
| * Returns true if the test should create off-heap regions. OffHeap tests should over-ride this |
| * method and return false. |
| */ |
| public boolean isOffHeap() { |
| return false; |
| } |
| |
| private static class MyFixedPartitionResolver implements FixedPartitionResolver { |
| |
| private final List<String> allPartitions; |
| |
| public MyFixedPartitionResolver(final List<String> allPartitions) { |
| this.allPartitions = allPartitions; |
| } |
| |
| @Override |
| public String getPartitionName(final EntryOperation opDetails, |
| @Deprecated final Set targetPartitions) { |
| int hash = Math.abs(opDetails.getKey().hashCode() % allPartitions.size()); |
| return allPartitions.get(hash); |
| } |
| |
| @Override |
| public Object getRoutingObject(final EntryOperation opDetails) { |
| return opDetails.getKey(); |
| } |
| |
| @Override |
| public String getName() { |
| return getClass().getName(); |
| } |
| |
| @Override |
| public void close() { |
| |
| } |
| } |
| } |
| |
| |
| class MyAsyncEventListener_CacheLoader implements AsyncEventListener { |
| private final Map eventsMap; |
| |
| public MyAsyncEventListener_CacheLoader() { |
| eventsMap = new ConcurrentHashMap(); |
| } |
| |
| @Override |
| public boolean processEvents(List<AsyncEvent> events) { |
| for (AsyncEvent event : events) { |
| eventsMap.put(event.getKey(), event); |
| } |
| return true; |
| } |
| |
| public Map getEventsMap() { |
| return eventsMap; |
| } |
| |
| @Override |
| public void close() {} |
| } |
| |
| |
| class MyCacheLoader implements CacheLoader, Declarable { |
| |
| @Override |
| public Object load(LoaderHelper helper) { |
| Long key = (Long) helper.getKey(); |
| return "LoadedValue" + "_" + key; |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public void init(Properties props) {} |
| |
| } |
| |
| |
| class SizeableGatewayEventSubstitutionFilter implements GatewayEventSubstitutionFilter, Declarable { |
| |
| private final AtomicInteger numToDataInvocations = new AtomicInteger(); |
| |
| protected static final String SUBSTITUTION_PREFIX = "substituted_"; |
| |
| @Override |
| public Object getSubstituteValue(EntryEvent event) { |
| return new GatewayEventSubstituteObject(this, SUBSTITUTION_PREFIX + event.getKey()); |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public void init(Properties properties) {} |
| |
| protected void incNumToDataInvocations() { |
| numToDataInvocations.incrementAndGet(); |
| } |
| |
| protected int getNumToDataInvocations() { |
| return numToDataInvocations.get(); |
| } |
| } |
| |
| |
| class GatewayEventSubstituteObject implements DataSerializable, Sizeable { |
| |
| private String id; |
| |
| private final SizeableGatewayEventSubstitutionFilter filter; |
| |
| public GatewayEventSubstituteObject(SizeableGatewayEventSubstitutionFilter filter, String id) { |
| this.filter = filter; |
| this.id = id; |
| } |
| |
| public String getId() { |
| return id; |
| } |
| |
| @Override |
| public void toData(DataOutput out) throws IOException { |
| filter.incNumToDataInvocations(); |
| DataSerializer.writeString(id, out); |
| } |
| |
| @Override |
| public void fromData(DataInput in) throws IOException, ClassNotFoundException { |
| id = DataSerializer.readString(in); |
| } |
| |
| @Override |
| public int getSizeInBytes() { |
| return 0; |
| } |
| |
| public String toString() { |
| return getClass().getSimpleName() + "[" + "id=" |
| + id + "]"; |
| } |
| } |
| |
| |
| class MyGatewayEventSubstitutionFilter implements GatewayEventSubstitutionFilter, Declarable { |
| |
| private final AtomicInteger numInvocations = new AtomicInteger(); |
| |
| protected static final String SUBSTITUTION_PREFIX = "substituted_"; |
| |
| @Override |
| public Object getSubstituteValue(EntryEvent event) { |
| numInvocations.incrementAndGet(); |
| return SUBSTITUTION_PREFIX + event.getKey(); |
| } |
| |
| @Override |
| public void close() {} |
| |
| @Override |
| public void init(Properties properties) {} |
| |
| protected int getNumInvocations() { |
| return numInvocations.get(); |
| } |
| } |
