| /* |
| * 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.parallel; |
| |
| import static org.apache.geode.distributed.internal.DistributionConfig.OFF_HEAP_MEMORY_SIZE_NAME; |
| import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPorts; |
| import static org.apache.geode.internal.cache.tier.sockets.Message.MAX_MESSAGE_SIZE_PROPERTY; |
| import static org.apache.geode.internal.util.ArrayUtils.asList; |
| import static org.apache.geode.test.awaitility.GeodeAwaitility.await; |
| import static org.apache.geode.test.dunit.IgnoredException.addIgnoredException; |
| import static org.assertj.core.api.Assertions.assertThat; |
| import static org.assertj.core.api.Assertions.catchThrowable; |
| import static org.mockito.ArgumentMatchers.any; |
| import static org.mockito.Mockito.doAnswer; |
| import static org.mockito.Mockito.mock; |
| |
| import java.io.Serializable; |
| import java.util.ArrayList; |
| import java.util.Collections; |
| import java.util.HashMap; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Properties; |
| import java.util.Set; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Executors; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.stream.Collectors; |
| import java.util.stream.IntStream; |
| import java.util.stream.Stream; |
| |
| import org.junit.After; |
| import org.junit.Before; |
| import org.junit.Rule; |
| import org.junit.Test; |
| import org.junit.experimental.categories.Category; |
| import util.TestException; |
| |
| import org.apache.geode.GemFireIOException; |
| import org.apache.geode.cache.CacheWriter; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.RegionShortcut; |
| import org.apache.geode.cache.wan.GatewayEventFilter; |
| import org.apache.geode.cache.wan.GatewaySender; |
| import org.apache.geode.distributed.internal.ClusterDistributionManager; |
| import org.apache.geode.distributed.internal.DistributionMessage; |
| import org.apache.geode.distributed.internal.DistributionMessageObserver; |
| import org.apache.geode.internal.cache.BucketRegion; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.internal.cache.PartitionedRegion; |
| import org.apache.geode.internal.cache.PartitionedRegionDataStore; |
| import org.apache.geode.internal.cache.tier.sockets.MessageTooLargeException; |
| import org.apache.geode.internal.cache.wan.AbstractGatewaySender; |
| import org.apache.geode.internal.cache.wan.GatewaySenderException; |
| import org.apache.geode.internal.cache.wan.WANTestBase; |
| import org.apache.geode.internal.offheap.MemoryAllocatorImpl; |
| import org.apache.geode.internal.offheap.OffHeapClearRequired; |
| import org.apache.geode.test.awaitility.GeodeAwaitility; |
| import org.apache.geode.test.dunit.AsyncInvocation; |
| import org.apache.geode.test.dunit.RMIException; |
| import org.apache.geode.test.dunit.VM; |
| import org.apache.geode.test.dunit.rules.ClusterStartupRule; |
| import org.apache.geode.test.dunit.rules.DistributedRestoreSystemProperties; |
| import org.apache.geode.test.dunit.rules.MemberVM; |
| import org.apache.geode.test.junit.categories.WanTest; |
| import org.apache.geode.test.junit.rules.serializable.SerializableTestName; |
| |
| /** |
| * DUnit test for operations on ParallelGatewaySender |
| */ |
| @Category(WanTest.class) |
| public class ParallelGatewaySenderOperationsDistributedTest extends WANTestBase { |
| |
| @Rule |
| public SerializableTestName testName = new SerializableTestName(); |
| |
| @Rule |
| public ClusterStartupRule clusterStartupRule = new ClusterStartupRule(); |
| |
| @Rule |
| public DistributedRestoreSystemProperties restoreSystemProperties = |
| new DistributedRestoreSystemProperties(); |
| |
| @Before |
| public void setUp() throws Exception { |
| addIgnoredException("Broken pipe||Unexpected IOException"); |
| addIgnoredException("Connection reset"); |
| addIgnoredException("Connection refused"); |
| addIgnoredException("could not get remote locator information"); |
| } |
| |
| @After |
| public void tearDown() { |
| for (VM vm : asList(vm4, vm5, vm6, vm7)) { |
| vm.invoke(() -> DistributionMessageObserver.setInstance(null)); |
| } |
| } |
| |
| @Test(timeout = 300_000) |
| public void testStopOneConcurrentGatewaySenderWithSSL() { |
| Integer lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1)); |
| Integer nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort)); |
| |
| createCacheInVMs(nyPort, vm2, vm3); |
| vm2.invoke(() -> createReceiverWithSSL(nyPort)); |
| vm3.invoke(() -> createReceiverWithSSL(nyPort)); |
| |
| vm4.invoke(() -> createCacheWithSSL(lnPort)); |
| vm5.invoke(() -> createCacheWithSSL(lnPort)); |
| |
| vm4.invoke(() -> createConcurrentSender("ln", 2, true, 100, 10, false, true, null, true, 5, |
| GatewaySender.OrderPolicy.KEY)); |
| vm5.invoke(() -> createConcurrentSender("ln", 2, true, 100, 10, false, true, null, true, 5, |
| GatewaySender.OrderPolicy.KEY)); |
| |
| String regionName = getUniqueName() + "_PR"; |
| vm4.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm5.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| |
| vm2.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm3.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| |
| startSenderInVMs("ln", vm4, vm5); |
| |
| vm4.invoke(() -> doPuts(regionName, 10)); |
| |
| vm4.invoke(() -> stopSender("ln")); |
| vm4.invoke(() -> startSender("ln")); |
| |
| vm4.invoke(() -> doPuts(regionName, 10)); |
| |
| vm5.invoke(() -> stopSender("ln")); |
| vm5.invoke(() -> startSender("ln")); |
| } |
| |
| @Test |
| public void testParallelGatewaySenderWithoutStarting() { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, false); |
| |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| vm4.invoke(() -> verifySenderStoppedState("ln")); |
| vm5.invoke(() -> verifySenderStoppedState("ln")); |
| vm6.invoke(() -> verifySenderStoppedState("ln")); |
| vm7.invoke(() -> verifySenderStoppedState("ln")); |
| |
| validateRegionSizes(getUniqueName() + "_PR", 0, vm2, vm3); |
| } |
| |
| /** |
| * ParallelGatewaySender should not be started on Accessor node |
| * |
| * <p> |
| * TRAC #44323: NewWan: ParallelGatewaySender should not be started on Accessor Node |
| */ |
| @Test |
| public void testParallelGatewaySenderStartOnAccessorNode() { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, true, true); |
| |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 10)); |
| |
| vm4.invoke(() -> validateParallelSenderQueueAllBucketsDrained("ln")); |
| vm5.invoke(() -> validateParallelSenderQueueAllBucketsDrained("ln")); |
| |
| validateRegionSizes(getUniqueName() + "_PR", 10, vm2, vm3); |
| } |
| |
| /** |
| * Normal scenario in which the sender is paused in between. |
| */ |
| @Test |
| public void testParallelPropagationSenderPause() { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true); |
| |
| // make sure all the senders are running before doing any puts |
| waitForSendersRunning(); |
| |
| // FIRST RUN: now, the senders are started. So, start the puts |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 100)); |
| |
| // now, pause all of the senders |
| vm4.invoke(() -> pauseSender("ln")); |
| vm5.invoke(() -> pauseSender("ln")); |
| vm6.invoke(() -> pauseSender("ln")); |
| vm7.invoke(() -> pauseSender("ln")); |
| |
| // SECOND RUN: keep one thread doing puts to the region |
| vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // verify region size remains on remote vm and is restricted below a specified limit (i.e. |
| // number of puts in the first run) |
| vm2.invoke(() -> validateRegionSizeRemainsSame(getUniqueName() + "_PR", 100)); |
| } |
| |
| /** |
| * Normal scenario in which a paused sender is resumed. |
| */ |
| @Test |
| public void testParallelPropagationSenderResume() throws Exception { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true); |
| |
| // make sure all the senders are running before doing any puts |
| waitForSendersRunning(); |
| |
| int numPuts = 1000; |
| // now, the senders are started. So, start the puts |
| AsyncInvocation<Void> async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", numPuts)); |
| |
| // now, pause all of the senders |
| vm4.invoke(() -> pauseSender("ln")); |
| vm5.invoke(() -> pauseSender("ln")); |
| vm6.invoke(() -> pauseSender("ln")); |
| vm7.invoke(() -> pauseSender("ln")); |
| |
| // resume the senders |
| vm4.invoke(() -> resumeSender("ln")); |
| vm5.invoke(() -> resumeSender("ln")); |
| vm6.invoke(() -> resumeSender("ln")); |
| vm7.invoke(() -> resumeSender("ln")); |
| |
| async.await(2, TimeUnit.MINUTES); |
| validateParallelSenderQueueAllBucketsDrained(); |
| |
| // find the region size on remote vm |
| vm2.invoke(() -> validateRegionSize(getUniqueName() + "_PR", numPuts)); |
| } |
| |
| /** |
| * Negative scenario in which a sender that is stopped (and not paused) is resumed. Expected: |
| * resume is only valid for pause. If a sender which is stopped is resumed, it will not be started |
| * again. |
| */ |
| @Test |
| public void testParallelPropagationSenderResumeNegativeScenario() { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createCacheInVMs(nyPort, vm2, vm3); |
| createReceiverInVMs(vm2, vm3); |
| |
| createCacheInVMs(lnPort, vm4, vm5); |
| |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| |
| vm4.invoke( |
| () -> createPartitionedRegion(getUniqueName() + "_PR", "ln", 1, 100, isOffHeap())); |
| vm5.invoke( |
| () -> createPartitionedRegion(getUniqueName() + "_PR", "ln", 1, 100, isOffHeap())); |
| |
| vm2.invoke( |
| () -> createPartitionedRegion(getUniqueName() + "_PR", null, 1, 100, isOffHeap())); |
| vm3.invoke( |
| () -> createPartitionedRegion(getUniqueName() + "_PR", null, 1, 100, isOffHeap())); |
| |
| startSenderInVMs("ln", vm4, vm5); |
| |
| // wait till the senders are running |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| |
| // start the puts |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 100)); |
| |
| // let the queue drain completely |
| vm4.invoke(() -> validateQueueContents("ln", 0)); |
| |
| // stop the senders |
| vm4.invoke(() -> stopSender("ln")); |
| vm5.invoke(() -> stopSender("ln")); |
| |
| // now, try to resume a stopped sender |
| vm4.invoke(() -> resumeSender("ln")); |
| vm5.invoke(() -> resumeSender("ln")); |
| |
| // do more puts |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // validate region size on remote vm to contain only the events put in local site |
| // before the senders are stopped. |
| vm2.invoke(() -> validateRegionSize(getUniqueName() + "_PR", 100)); |
| } |
| |
| /** |
| * Normal scenario in which a sender is stopped. |
| */ |
| @Test |
| public void testParallelPropagationSenderStop() { |
| addIgnoredException("Broken pipe"); |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true); |
| |
| // make sure all the senders are running before doing any puts |
| waitForSendersRunning(); |
| |
| // FIRST RUN: now, the senders are started. So, do some of the puts |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 100)); |
| |
| // now, stop all of the senders |
| stopSenders(); |
| |
| // SECOND RUN: keep one thread doing puts |
| vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // verify region size remains on remote vm and is restricted below a specified limit (number of |
| // puts in the first run) |
| vm2.invoke(() -> validateRegionSizeRemainsSame(getUniqueName() + "_PR", 100)); |
| } |
| |
| /** |
| * Normal scenario in which a sender is stopped and then started again. |
| */ |
| @Test |
| public void testParallelPropagationSenderStartAfterStop() { |
| addIgnoredException("Broken pipe"); |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| String regionName = getUniqueName() + "_PR"; |
| |
| createCacheInVMs(nyPort, vm2, vm3); |
| createCacheInVMs(lnPort, vm4, vm5, vm6, vm7); |
| |
| vm2.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm3.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| |
| createReceiverInVMs(vm2, vm3); |
| |
| vm4.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm5.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm6.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm7.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| |
| startSenderInVMs("ln", vm4, vm5, vm6, vm7); |
| |
| // make sure all the senders are running before doing any puts |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| |
| // FIRST RUN: now, the senders are started. So, do some of the puts |
| vm4.invoke(() -> doPuts(regionName, 200)); |
| |
| // now, stop all of the senders |
| vm4.invoke(() -> stopSender("ln")); |
| vm5.invoke(() -> stopSender("ln")); |
| vm6.invoke(() -> stopSender("ln")); |
| vm7.invoke(() -> stopSender("ln")); |
| |
| // Region size on remote site should remain same and below the number of puts done in the FIRST |
| // RUN |
| vm2.invoke(() -> validateRegionSizeRemainsSame(regionName, 200)); |
| |
| // SECOND RUN: do some of the puts after the senders are stopped |
| vm4.invoke(() -> doPuts(regionName, 1000)); |
| |
| // Region size on remote site should remain same and below the number of puts done in the FIRST |
| // RUN |
| vm2.invoke(() -> validateRegionSizeRemainsSame(regionName, 200)); |
| |
| // start the senders again |
| startSenderInVMs("ln", vm4, vm5, vm6, vm7); |
| |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| |
| // Region size on remote site should remain same and below the number of puts done in the FIRST |
| // RUN |
| vm2.invoke(() -> validateRegionSizeRemainsSame(regionName, 200)); |
| |
| // SECOND RUN: do some more puts |
| vm4.invoke(() -> doPuts(regionName, 1000)); |
| |
| // verify all the buckets on all the sender nodes are drained |
| validateParallelSenderQueueAllBucketsDrained(); |
| |
| // verify the events propagate to remote site |
| vm2.invoke(() -> validateRegionSize(regionName, 1000)); |
| |
| vm4.invoke(() -> validateQueueSizeStat("ln", 0)); |
| vm5.invoke(() -> validateQueueSizeStat("ln", 0)); |
| vm6.invoke(() -> validateQueueSizeStat("ln", 0)); |
| vm7.invoke(() -> validateQueueSizeStat("ln", 0)); |
| } |
| |
| /** |
| * Normal scenario in which a sender is stopped and then started again. Differs from above test |
| * case in the way that when the sender is starting from stopped state, puts are simultaneously |
| * happening on the region by another thread. |
| */ |
| @Test |
| public void testParallelPropagationSenderStartAfterStop_Scenario2() throws Exception { |
| addIgnoredException("Broken pipe"); |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true); |
| |
| // make sure all the senders are running before doing any puts |
| waitForSendersRunning(); |
| |
| // FIRST RUN: now, the senders are started. So, do some of the puts |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 200)); |
| |
| // Make sure the puts make it to the remote side |
| vm2.invoke(() -> validateRegionSize(getUniqueName() + "_PR", 200)); |
| vm3.invoke(() -> validateRegionSize(getUniqueName() + "_PR", 200)); |
| |
| // now, stop all of the senders |
| stopSenders(); |
| |
| vm2.invoke(() -> validateRegionSize(getUniqueName() + "_PR", 200)); |
| vm3.invoke(() -> validateRegionSize(getUniqueName() + "_PR", 200)); |
| |
| // SECOND RUN: do some of the puts after the senders are stopped |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // Region size on remote site should remain same and below the number of puts done in the FIRST |
| // RUN |
| vm2.invoke(() -> validateRegionSizeRemainsSame(getUniqueName() + "_PR", 200)); |
| |
| // SECOND RUN: start async puts on region |
| boolean foundEventsDroppedDueToPrimarySenderNotRunning = false; |
| int count = 0; |
| |
| do { |
| stopSenders(); |
| AsyncInvocation<Void> async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 5000)); |
| |
| // when puts are happening by another thread, start the senders |
| startSenderInVMsAsync("ln", vm4, vm5, vm6, vm7); |
| |
| async.join(); |
| List<Integer> vm4List = vm4.invoke(() -> WANTestBase.getSenderStatsForDroppedEvents("ln")); |
| List<Integer> vm5List = vm5.invoke(() -> WANTestBase.getSenderStatsForDroppedEvents("ln")); |
| List<Integer> vm6List = vm6.invoke(() -> WANTestBase.getSenderStatsForDroppedEvents("ln")); |
| List<Integer> vm7List = vm7.invoke(() -> WANTestBase.getSenderStatsForDroppedEvents("ln")); |
| if (vm4List.get(0) + vm5List.get(0) + vm6List.get(0) + vm7List.get(0) > 0) { |
| foundEventsDroppedDueToPrimarySenderNotRunning = true; |
| } |
| count++; |
| } while (!foundEventsDroppedDueToPrimarySenderNotRunning && count < 5); |
| |
| // verify all the buckets on all the sender nodes are drained |
| validateParallelSenderQueueAllBucketsDrained(); |
| |
| // verify that the queue size ultimately becomes zero. That means all the events propagate to |
| // remote site. |
| |
| vm4.invoke(() -> validateQueueContents("ln", 0)); |
| } |
| |
| /** |
| * Normal scenario in which a sender is stopped and then started again on accessor node. |
| */ |
| @Test |
| public void testParallelPropagationSenderStartAfterStopOnAccessorNode() { |
| addIgnoredException("Broken pipe"); |
| addIgnoredException("Connection reset"); |
| addIgnoredException("Unexpected IOException"); |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, true, true); |
| |
| // make sure all the senders are not running on accessor nodes and running on non-accessor nodes |
| waitForSendersRunning(); |
| |
| // FIRST RUN: now, the senders are started. So, do some of the puts |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 200)); |
| |
| // now, stop all of the senders |
| stopSenders(); |
| |
| // SECOND RUN: do some of the puts after the senders are stopped |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // Region size on remote site should remain same and below the number of puts done in the FIRST |
| // RUN |
| vm2.invoke(() -> validateRegionSizeRemainsSame(getUniqueName() + "_PR", 200)); |
| |
| // start the senders again |
| startSenderInVMs("ln", vm4, vm5, vm6, vm7); |
| |
| // Region size on remote site should remain same and below the number of puts done in the FIRST |
| // RUN |
| vm2.invoke(() -> validateRegionSizeRemainsSame(getUniqueName() + "_PR", 200)); |
| |
| // SECOND RUN: do some more puts |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // verify all buckets drained only on non-accessor nodes. |
| vm4.invoke(() -> validateParallelSenderQueueAllBucketsDrained("ln")); |
| vm5.invoke(() -> validateParallelSenderQueueAllBucketsDrained("ln")); |
| |
| // verify the events propagate to remote site |
| vm2.invoke(() -> validateRegionSize(getUniqueName() + "_PR", 1000)); |
| } |
| |
| /** |
| * Normal scenario in which a combinations of start, pause, resume operations is tested |
| */ |
| @Test |
| public void testStartPauseResumeParallelGatewaySender() throws Exception { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true); |
| |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // Since puts are already done on userPR, it will have the buckets created. |
| // During sender start, it will wait until those buckets are created for shadowPR as well. |
| // Start the senders in async threads, so colocation of shadowPR will be complete and |
| // missing buckets will be created in PRHARedundancyProvider.createMissingBuckets(). |
| startSenderInVMsAsync("ln", vm4, vm5, vm6, vm7); |
| |
| waitForSendersRunning(); |
| |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 5000)); |
| |
| vm4.invoke(() -> pauseSender("ln")); |
| vm5.invoke(() -> pauseSender("ln")); |
| vm6.invoke(() -> pauseSender("ln")); |
| vm7.invoke(() -> pauseSender("ln")); |
| |
| vm4.invoke(() -> verifySenderPausedState("ln")); |
| vm5.invoke(() -> verifySenderPausedState("ln")); |
| vm6.invoke(() -> verifySenderPausedState("ln")); |
| vm7.invoke(() -> verifySenderPausedState("ln")); |
| |
| AsyncInvocation<Void> async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| vm4.invoke(() -> resumeSender("ln")); |
| vm5.invoke(() -> resumeSender("ln")); |
| vm6.invoke(() -> resumeSender("ln")); |
| vm7.invoke(() -> resumeSender("ln")); |
| |
| vm4.invoke(() -> verifySenderResumedState("ln")); |
| vm5.invoke(() -> verifySenderResumedState("ln")); |
| vm6.invoke(() -> verifySenderResumedState("ln")); |
| vm7.invoke(() -> verifySenderResumedState("ln")); |
| |
| async.await(); |
| |
| // verify all buckets drained on all sender nodes. |
| validateParallelSenderQueueAllBucketsDrained(); |
| |
| validateRegionSizes(getUniqueName() + "_PR", 5000, vm2, vm3); |
| } |
| |
| /** |
| * Since the sender is attached to a region and in use, it can not be destroyed. Hence, exception |
| * is thrown by the sender API. |
| */ |
| @Test |
| public void testDestroyParallelGatewaySenderExceptionScenario() { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true); |
| |
| // make sure all the senders are running before doing any puts |
| waitForSendersRunning(); |
| |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // try destroying on couple of nodes |
| Throwable caughtException = catchThrowable(() -> vm4.invoke(() -> destroySender("ln"))); |
| |
| assertThat(caughtException).isInstanceOf(RMIException.class); |
| assertThat(caughtException.getCause()).isInstanceOf(GatewaySenderException.class); |
| |
| caughtException = catchThrowable(() -> vm5.invoke(() -> destroySender("ln"))); |
| |
| assertThat(caughtException).isInstanceOf(RMIException.class); |
| assertThat(caughtException.getCause()).isInstanceOf(GatewaySenderException.class); |
| |
| vm2.invoke(() -> validateRegionSize(getUniqueName() + "_PR", 1000)); |
| } |
| |
| @Test |
| public void testDestroyParallelGatewaySender() { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true); |
| |
| // make sure all the senders are running |
| waitForSendersRunning(); |
| |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| // stop the sender and remove from region before calling destroy on it |
| stopSenders(); |
| |
| vm4.invoke(() -> removeSenderFromTheRegion("ln", getUniqueName() + "_PR")); |
| vm5.invoke(() -> removeSenderFromTheRegion("ln", getUniqueName() + "_PR")); |
| vm6.invoke(() -> removeSenderFromTheRegion("ln", getUniqueName() + "_PR")); |
| vm7.invoke(() -> removeSenderFromTheRegion("ln", getUniqueName() + "_PR")); |
| |
| vm4.invoke(() -> destroySender("ln")); |
| vm5.invoke(() -> destroySender("ln")); |
| vm6.invoke(() -> destroySender("ln")); |
| vm7.invoke(() -> destroySender("ln")); |
| |
| vm4.invoke(() -> verifySenderDestroyed("ln", true)); |
| vm5.invoke(() -> verifySenderDestroyed("ln", true)); |
| vm6.invoke(() -> verifySenderDestroyed("ln", true)); |
| vm7.invoke(() -> verifySenderDestroyed("ln", true)); |
| } |
| |
| @Test |
| public void destroyParallelGatewaySenderShouldNotStopDispatchingFromOtherSendersAttachedToTheRegion() { |
| String site2SenderId = "site2-sender"; |
| String site3SenderId = "site3-sender"; |
| String regionName = testName.getMethodName(); |
| int[] ports = getRandomAvailableTCPPorts(3); |
| int site1Port = ports[0]; |
| int site2Port = ports[1]; |
| int site3Port = ports[2]; |
| Set<String> site1RemoteLocators = |
| Stream.of("localhost[" + site2Port + "]", "localhost[" + site3Port + "]") |
| .collect(Collectors.toSet()); |
| Set<String> site2RemoteLocators = |
| Stream.of("localhost[" + site1Port + "]", "localhost[" + site3Port + "]") |
| .collect(Collectors.toSet()); |
| Set<String> site3RemoteLocators = |
| Stream.of("localhost[" + site1Port + "]", "localhost[" + site2Port + "]") |
| .collect(Collectors.toSet()); |
| |
| // Start 3 sites. |
| vm0.invoke(() -> createLocator(1, site1Port, |
| Collections.singleton("localhost[" + site1Port + "]"), site1RemoteLocators)); |
| vm1.invoke(() -> createLocator(2, site2Port, |
| Collections.singleton("localhost[" + site2Port + "]"), site2RemoteLocators)); |
| vm2.invoke(() -> createLocator(3, site3Port, |
| Collections.singleton("localhost[" + site3Port + "]"), site3RemoteLocators)); |
| |
| // Create the cache on the 3 sites. |
| createCacheInVMs(site1Port, vm3); |
| createCacheInVMs(site2Port, vm4); |
| createCacheInVMs(site3Port, vm5); |
| |
| // Create receiver and region on sites 2 and 3. |
| asList(vm4, vm5).forEach(vm -> vm.invoke(() -> { |
| createReceiver(); |
| createPartitionedRegion(regionName, null, 1, 113, isOffHeap()); |
| })); |
| |
| // Create senders and partitioned region on site 1. |
| vm3.invoke(() -> { |
| createSender(site2SenderId, 2, true, 100, 20, false, false, null, false); |
| createSender(site3SenderId, 3, true, 100, 20, false, false, null, false); |
| waitForSenderRunningState(site2SenderId); |
| waitForSenderRunningState(site3SenderId); |
| |
| createPartitionedRegion(regionName, String.join(",", site2SenderId, site3SenderId), 1, 113, |
| isOffHeap()); |
| }); |
| |
| // #################################################################################### // |
| |
| final int FIRST_BATCH = 100; |
| final int SECOND_BATCH = 200; |
| final Map<String, String> firstBatch = new HashMap<>(); |
| IntStream.range(0, FIRST_BATCH).forEach(i -> firstBatch.put("Key" + i, "Value" + i)); |
| final Map<String, String> secondBatch = new HashMap<>(); |
| IntStream.range(FIRST_BATCH, SECOND_BATCH) |
| .forEach(i -> secondBatch.put("Key" + i, "Value" + i)); |
| |
| // Insert first batch and wait until the queues are empty. |
| vm3.invoke(() -> { |
| cache.getRegion(regionName).putAll(firstBatch); |
| checkQueueSize(site2SenderId, 0); |
| checkQueueSize(site3SenderId, 0); |
| }); |
| |
| // Wait until sites 2 and 3 have received all updates. |
| asList(vm4, vm5).forEach(vm -> vm.invoke(() -> { |
| Region<String, String> region = cache.getRegion(regionName); |
| await().untilAsserted(() -> assertThat(region.size()).isEqualTo(FIRST_BATCH)); |
| firstBatch.forEach((key, value) -> assertThat(region.get(key)).isEqualTo(value)); |
| })); |
| |
| // Stop sender to site3, remove it from the region and destroy it. |
| vm3.invoke(() -> { |
| stopSender(site3SenderId); |
| removeSenderFromTheRegion(site3SenderId, regionName); |
| destroySender(site3SenderId); |
| verifySenderDestroyed(site3SenderId, true); |
| }); |
| |
| // Insert second batch and wait until the queue is empty. |
| vm3.invoke(() -> { |
| cache.getRegion(regionName).putAll(secondBatch); |
| checkQueueSize(site2SenderId, 0); |
| }); |
| |
| // Site 3 should only have the first batch. |
| vm5.invoke(() -> { |
| Region<String, String> region = cache.getRegion(regionName); |
| await().untilAsserted(() -> assertThat(region.size()).isEqualTo(FIRST_BATCH)); |
| firstBatch.forEach((key, value) -> assertThat(region.get(key)).isEqualTo(value)); |
| }); |
| |
| // Site 2 should have both batches. |
| vm4.invoke(() -> { |
| Region<String, String> region = cache.getRegion(regionName); |
| await().untilAsserted(() -> assertThat(region.size()).isEqualTo(SECOND_BATCH)); |
| firstBatch.forEach((key, value) -> assertThat(region.get(key)).isEqualTo(value)); |
| secondBatch.forEach((key, value) -> assertThat(region.get(key)).isEqualTo(value)); |
| }); |
| } |
| |
| @Test |
| public void testParallelGatewaySenderMessageTooLargeException() { |
| vm4.invoke(() -> System.setProperty(MAX_MESSAGE_SIZE_PROPERTY, String.valueOf(1024 * 1024))); |
| |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| // Create and start sender with reduced maximum message size and 1 dispatcher thread |
| String regionName = getUniqueName() + "_PR"; |
| vm4.invoke(() -> createCache(lnPort)); |
| vm4.invoke(() -> setNumDispatcherThreadsForTheRun(1)); |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 100, false, false, null, false)); |
| vm4.invoke(() -> createPartitionedRegion(regionName, "ln", 0, 100, isOffHeap())); |
| |
| // Do puts |
| int numPuts = 200; |
| vm4.invoke(() -> doPuts(regionName, numPuts, new byte[11000])); |
| validateRegionSizes(regionName, numPuts, vm4); |
| |
| // Start receiver |
| addIgnoredException(MessageTooLargeException.class.getName(), vm4); |
| addIgnoredException(GemFireIOException.class.getName(), vm4); |
| |
| vm2.invoke(() -> createCache(nyPort)); |
| vm2.invoke(() -> createPartitionedRegion(regionName, null, 0, 100, isOffHeap())); |
| vm2.invoke(WANTestBase::createReceiver); |
| |
| validateRegionSizes(regionName, numPuts, vm2); |
| |
| vm4.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| GeodeAwaitility.await() |
| .untilAsserted(() -> assertThat(System.getProperty(MAX_MESSAGE_SIZE_PROPERTY)) |
| .isEqualTo(String.valueOf(1024 * 1024))); |
| GeodeAwaitility.await().untilAsserted( |
| () -> assertThat(sender.getStatistics().getBatchesResized()).isGreaterThan(0)); |
| }); |
| } |
| |
| @Test |
| public void testMultiSiteReplication() { |
| String site1to2SenderId = "site1to2-sender"; |
| String site1to3SenderId = "site1to3-sender"; |
| String site2to3SenderId = "site2to3-sender"; |
| String regionName = testName.getMethodName(); |
| int[] ports = getRandomAvailableTCPPorts(3); |
| int site1Port = ports[0]; |
| int site2Port = ports[1]; |
| int site3Port = ports[2]; |
| Set<String> site1RemoteLocators = |
| Stream.of("localhost[" + site2Port + "]", "localhost[" + site3Port + "]") |
| .collect(Collectors.toSet()); |
| Set<String> site2RemoteLocators = |
| Stream.of("localhost[" + site1Port + "]", "localhost[" + site3Port + "]") |
| .collect(Collectors.toSet()); |
| Set<String> site3RemoteLocators = |
| Stream.of("localhost[" + site1Port + "]", "localhost[" + site2Port + "]") |
| .collect(Collectors.toSet()); |
| |
| // Start 3 sites. |
| vm0.invoke(() -> createLocator(1, site1Port, |
| Collections.singleton("localhost[" + site1Port + "]"), site1RemoteLocators)); |
| vm1.invoke(() -> createLocator(2, site2Port, |
| Collections.singleton("localhost[" + site2Port + "]"), site2RemoteLocators)); |
| vm2.invoke(() -> createLocator(3, site3Port, |
| Collections.singleton("localhost[" + site3Port + "]"), site3RemoteLocators)); |
| |
| // Create the cache on the 3 sites. |
| createCacheInVMs(site1Port, vm3); |
| createCacheInVMs(site2Port, vm4); |
| createCacheInVMs(site3Port, vm5); |
| |
| // Create senders and partitioned region on site 1. |
| vm3.invoke(() -> { |
| createSender(site1to2SenderId, 2, true, 100, 20, false, false, null, false); |
| createSender(site1to3SenderId, 3, true, 100, 20, false, false, null, false); |
| waitForSenderRunningState(site1to2SenderId); |
| waitForSenderRunningState(site1to3SenderId); |
| |
| createPartitionedRegion(regionName, String.join(",", site1to2SenderId, site1to3SenderId), 1, |
| 113, |
| isOffHeap()); |
| }); |
| |
| // Create receiver, sender and partitioned region on site 2. |
| vm4.invoke(() -> { |
| createReceiver(); |
| createSender(site2to3SenderId, 3, true, 100, 20, false, false, null, false); |
| waitForSenderRunningState(site2to3SenderId); |
| |
| createPartitionedRegion(regionName, String.join(",", site2to3SenderId), 1, 113, |
| isOffHeap()); |
| }); |
| |
| // Create receiver and partitioned region on site 3. |
| vm5.invoke(() -> { |
| createReceiver(); |
| createPartitionedRegion(regionName, null, 1, 113, isOffHeap()); |
| }); |
| |
| // Do puts |
| vm3.invoke(() -> doPuts(regionName, 200)); |
| validateRegionSizes(regionName, 200, vm3, vm4, vm5); |
| |
| // Stop sender from site 1 to site 3. |
| vm3.invoke(() -> stopSender(site1to3SenderId)); |
| |
| // Do puts again |
| vm3.invoke(() -> doPuts(regionName, 1000)); |
| validateRegionSizes(regionName, 1000, vm3, vm4, vm5); |
| } |
| |
| @Test |
| public void testParallelGatewaySenderConcurrentPutClearNoOffheapOrphans() { |
| MemberVM locator = clusterStartupRule.startLocatorVM(1, new Properties()); |
| Properties properties = new Properties(); |
| properties.put(OFF_HEAP_MEMORY_SIZE_NAME, "100"); |
| MemberVM server = clusterStartupRule.startServerVM(2, properties, locator.getPort()); |
| final String regionName = "portfolios"; |
| final String gatewaySenderId = "ln"; |
| |
| server.invoke(() -> { |
| addIgnoredException("could not get remote locator"); |
| InternalCache cache = ClusterStartupRule.getCache(); |
| GatewaySender sender = |
| cache.createGatewaySenderFactory().setParallel(true).create(gatewaySenderId, 1); |
| Region<Object, Object> userRegion = |
| cache.createRegionFactory(RegionShortcut.PARTITION).setOffHeap(true) |
| .addGatewaySenderId("ln").create(regionName); |
| PartitionedRegion shadowRegion = (PartitionedRegion) ((AbstractGatewaySender) sender) |
| .getEventProcessor().getQueue().getRegion(); |
| @SuppressWarnings("unchecked") |
| CacheWriter<Object, Object> mockCacheWriter = mock(CacheWriter.class); |
| CountDownLatch cacheWriterLatch = new CountDownLatch(1); |
| CountDownLatch shadowRegionClearLatch = new CountDownLatch(1); |
| |
| doAnswer(invocation -> { |
| // The cache writer is invoked between when the region entry is created with value of |
| // Token.REMOVED_PHASE_1 and when it is replaced with the actual GatewaySenderEvent. |
| // We use this hook to trigger the clear logic when the token is still in the |
| // region entry. |
| cacheWriterLatch.countDown(); |
| // Wait until the clear is complete before putting the actual GatewaySenderEvent in the |
| // region entry. |
| shadowRegionClearLatch.await(); |
| return null; |
| }).when(mockCacheWriter).beforeCreate(any()); |
| |
| shadowRegion.setCacheWriter(mockCacheWriter); |
| |
| ExecutorService service = Executors.newFixedThreadPool(2); |
| |
| List<Callable<Object>> callables = new ArrayList<>(); |
| |
| // In one thread, we are going to put some test data in the user region, |
| // which will eventually put the GatewaySenderEvent into the shadow region |
| callables.add(Executors.callable(() -> { |
| try { |
| userRegion.put("testKey", "testValue"); |
| } catch (Exception ex) { |
| throw new RuntimeException(ex); |
| } |
| })); |
| |
| // In another thread, we are clear the shadow region's buckets. If the region entry is a |
| // Token.REMOVED_PHASE_1 at this time, a leak is possible. We can guarantee that the |
| // Token.REMOVED_PHASE_1 is present by using the mocked cache writer defined |
| // above. |
| callables.add(Executors.callable(() -> { |
| try { |
| OffHeapClearRequired.doWithOffHeapClear(() -> { |
| // Wait for the cache writer to be invoked to release this countdown latch. |
| // This guarantees that the region entry will contain a Token.REMOVED_PHASE_1. |
| try { |
| cacheWriterLatch.await(); |
| } catch (InterruptedException e) { |
| throw new TestException( |
| "Thread was interrupted while waiting for mocked cache writer to be invoked"); |
| } |
| |
| clearShadowBucketRegions(shadowRegion); |
| |
| // Signal to the cache writer that the clear is complete and the put can continue. |
| shadowRegionClearLatch.countDown(); |
| }); |
| } catch (Exception ex) { |
| throw new RuntimeException(ex); |
| } |
| })); |
| |
| List<Future<Object>> futures = service.invokeAll(callables, 10, TimeUnit.SECONDS); |
| |
| for (Future<Object> future : futures) { |
| try { |
| future.get(); |
| } catch (Exception ex) { |
| throw new TestException( |
| "Exception thrown while executing put and clear concurrently", |
| ex); |
| } |
| } |
| |
| userRegion.close(); |
| |
| await("Waiting for off-heap to be freed").until( |
| () -> 0 == ((MemoryAllocatorImpl) cache.getOffHeapStore()).getOrphans(cache).size()); |
| }); |
| } |
| |
| @Test |
| public void testParallelGWSenderUpdateAttrWhileEntriesInQueue() { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createCacheInVMs(nyPort, vm2, vm3); |
| |
| createCacheInVMs(lnPort, vm4, vm5, vm6, vm7); |
| |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| |
| createPartitionedRegions(false); |
| |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 1000)); |
| |
| updateBatchSize(50); |
| updateBatchTimeInterval(100); |
| |
| createReceiverInVMs(vm2, vm3); |
| |
| validateRegionSizes(getUniqueName() + "_PR", 1000, vm2, vm3); |
| |
| checkBatchSize(50); |
| checkBatchTimeInterval(100); |
| } |
| |
| @Test |
| public void testParallelGWSenderUpdateAttrWhilePutting() throws Exception { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createCacheInVMs(nyPort, vm2, vm3); |
| createReceiverInVMs(vm2, vm3); |
| |
| createCacheInVMs(lnPort, vm4, vm5, vm6, vm7); |
| |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| |
| createPartitionedRegions(false); |
| |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| |
| AsyncInvocation<Void> async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 5000)); |
| |
| updateBatchSize(100); |
| updateBatchTimeInterval(150); |
| |
| async.await(); |
| validateRegionSizes(getUniqueName() + "_PR", 5000, vm2, vm3); |
| |
| checkBatchSize(100); |
| checkBatchTimeInterval(150); |
| } |
| |
| @Test |
| public void testParallelGWSenderUpdateAttrWhilePaused() { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| createCacheInVMs(nyPort, vm2, vm3); |
| createReceiverInVMs(vm2, vm3); |
| |
| createCacheInVMs(lnPort, vm4, vm5, vm6, vm7); |
| |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| |
| createPartitionedRegions(false); |
| |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| |
| vm4.invoke(() -> pauseSender("ln")); |
| vm5.invoke(() -> pauseSender("ln")); |
| vm6.invoke(() -> pauseSender("ln")); |
| vm7.invoke(() -> pauseSender("ln")); |
| |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 5000)); |
| |
| updateGroupTransactionEvents(true); |
| updateBatchTimeInterval(200); |
| |
| validateRegionSizes(getUniqueName() + "_PR", 0, vm2, vm3); |
| |
| checkGroupTransactionEvents(true); |
| checkBatchTimeInterval(200); |
| |
| vm4.invoke(() -> resumeSender("ln")); |
| vm5.invoke(() -> resumeSender("ln")); |
| vm6.invoke(() -> resumeSender("ln")); |
| vm7.invoke(() -> resumeSender("ln")); |
| |
| validateRegionSizes(getUniqueName() + "_PR", 5000, vm2, vm3); |
| |
| } |
| |
| @Test |
| public void testParallelGWSenderUpdateFiltersWhilePutting() throws Exception { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| List<GatewayEventFilter> filters = new ArrayList<>(); |
| filters.add(new MyGatewayEventFilter_AfterAck()); |
| filters.add(new PDXGatewayEventFilter()); |
| |
| List<GatewayEventFilter> filters2 = new ArrayList<>(); |
| filters.add(new MyGatewayEventFilter()); |
| filters.add(new MyGatewayEventFilter_AfterAck()); |
| filters.add(new PDXGatewayEventFilter()); |
| |
| createCacheInVMs(nyPort, vm2, vm3); |
| createReceiverInVMs(vm2, vm3); |
| |
| createCacheInVMs(lnPort, vm4, vm5, vm6, vm7); |
| |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| |
| createPartitionedRegions(false); |
| |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| |
| AsyncInvocation<Void> async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 5000)); |
| checkBatchSize(10); |
| |
| updateBatchSize(100); |
| updateGatewayEventFilters(filters); |
| checkBatchSize(100); |
| updateGatewayEventFilters(filters2); |
| |
| async.await(); |
| |
| } |
| |
| |
| @Test |
| public void testParallelGWSenderUpdateFiltersWhilePuttingOnOneDispatcThread() throws Exception { |
| Integer[] locatorPorts = createLNAndNYLocators(); |
| Integer lnPort = locatorPorts[0]; |
| Integer nyPort = locatorPorts[1]; |
| |
| List<GatewayEventFilter> filters = new ArrayList<>(); |
| filters.add(new MyGatewayEventFilter_AfterAck()); |
| filters.add(new PDXGatewayEventFilter()); |
| |
| List<GatewayEventFilter> filters2 = new ArrayList<>(); |
| filters.add(new MyGatewayEventFilter()); |
| filters.add(new MyGatewayEventFilter_AfterAck()); |
| filters.add(new PDXGatewayEventFilter()); |
| |
| createCacheInVMs(nyPort, vm2, vm3); |
| createReceiverInVMs(vm2, vm3); |
| |
| createCacheInVMs(lnPort, vm4, vm5, vm6, vm7); |
| vm4.invoke(() -> setNumDispatcherThreadsForTheRun(1)); |
| vm5.invoke(() -> setNumDispatcherThreadsForTheRun(1)); |
| vm6.invoke(() -> setNumDispatcherThreadsForTheRun(1)); |
| vm7.invoke(() -> setNumDispatcherThreadsForTheRun(1)); |
| |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, false)); |
| |
| createPartitionedRegions(false); |
| |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| |
| AsyncInvocation<Void> async = vm4.invokeAsync(() -> doPuts(getUniqueName() + "_PR", 5000)); |
| checkBatchSize(10); |
| |
| updateBatchSize(100); |
| updateGatewayEventFilters(filters); |
| checkBatchSize(100); |
| updateGatewayEventFilters(filters2); |
| |
| async.await(); |
| |
| } |
| |
| /** |
| * Put entries in region after gateway sender is stopped. Count number of PQRM messages sent. |
| */ |
| @Test |
| public void testDroppedEventsSignalizationToSecondaryQueueWhileSenderStopped() { |
| int lnPort = vm0.invoke(() -> createFirstLocatorWithDSId(1)); |
| int nyPort = vm1.invoke(() -> createFirstRemoteLocator(2, lnPort)); |
| |
| createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true); |
| |
| // make sure all the senders are running before doing any puts |
| waitForSendersRunning(); |
| |
| // FIRST RUN: now, the senders are started. So, start the puts |
| vm4.invoke(() -> doPuts(getUniqueName() + "_PR", 100)); |
| |
| vm2.invoke(() -> validateRegionSizeRemainsSame(getUniqueName() + "_PR", 100)); |
| |
| stopSenders(); |
| |
| waitForAllSendersNotRunning(); |
| |
| vm4.invoke(() -> { |
| DistributionMessageObserver.setInstance(new CountSentPQRMObserver()); |
| }); |
| vm5.invoke(() -> { |
| DistributionMessageObserver.setInstance(new CountSentPQRMObserver()); |
| }); |
| vm6.invoke(() -> { |
| DistributionMessageObserver.setInstance(new CountSentPQRMObserver()); |
| }); |
| vm7.invoke(() -> { |
| DistributionMessageObserver.setInstance(new CountSentPQRMObserver()); |
| }); |
| |
| // SECOND RUN: keep one thread doing puts to the region |
| vm4.invoke(() -> doPutsFrom(getUniqueName() + "_PR", 100, 200)); |
| |
| vm2.invoke(() -> validateRegionSizeRemainsSame(getUniqueName() + "_PR", 100)); |
| |
| int parallelQueueRemovalMessageCountInVm4 = vm4.invoke(() -> { |
| CountSentPQRMObserver observer = |
| (CountSentPQRMObserver) DistributionMessageObserver.getInstance(); |
| return observer.getNumberOfSentPQRM(); |
| }); |
| |
| int parallelQueueRemovalMessageCountInVm5 = vm5.invoke(() -> { |
| CountSentPQRMObserver observer = |
| (CountSentPQRMObserver) DistributionMessageObserver.getInstance(); |
| return observer.getNumberOfSentPQRM(); |
| }); |
| |
| int parallelQueueRemovalMessageCountInVm6 = vm6.invoke(() -> { |
| CountSentPQRMObserver observer = |
| (CountSentPQRMObserver) DistributionMessageObserver.getInstance(); |
| return observer.getNumberOfSentPQRM(); |
| }); |
| |
| int parallelQueueRemovalMessageCountInVm7 = vm7.invoke(() -> { |
| CountSentPQRMObserver observer = |
| (CountSentPQRMObserver) DistributionMessageObserver.getInstance(); |
| return observer.getNumberOfSentPQRM(); |
| }); |
| |
| assertThat(parallelQueueRemovalMessageCountInVm4 + parallelQueueRemovalMessageCountInVm5 |
| + parallelQueueRemovalMessageCountInVm6 + parallelQueueRemovalMessageCountInVm7) |
| .isEqualTo(100); |
| |
| await().untilAsserted(() -> { |
| assertThat(vm4.invoke(() -> getSecondaryQueueSizeInStats("ln"))).isZero(); |
| assertThat(vm5.invoke(() -> getSecondaryQueueSizeInStats("ln"))).isZero(); |
| assertThat(vm6.invoke(() -> getSecondaryQueueSizeInStats("ln"))).isZero(); |
| assertThat(vm7.invoke(() -> getSecondaryQueueSizeInStats("ln"))).isZero(); |
| }); |
| |
| } |
| |
| private void clearShadowBucketRegions(PartitionedRegion shadowRegion) { |
| PartitionedRegionDataStore.BucketVisitor bucketVisitor = |
| (bucketId, r) -> ((BucketRegion) r).clearEntries(null); |
| |
| shadowRegion.getDataStore().visitBuckets(bucketVisitor); |
| } |
| |
| private void createSendersReceiversAndPartitionedRegion(Integer lnPort, Integer nyPort, |
| boolean createAccessors, boolean startSenders) { |
| createSendersAndReceivers(lnPort, nyPort); |
| |
| createPartitionedRegions(createAccessors); |
| |
| if (startSenders) { |
| startSenderInVMs("ln", vm4, vm5, vm6, vm7); |
| } |
| } |
| |
| private void createSendersAndReceivers(Integer lnPort, Integer nyPort) { |
| createCacheInVMs(nyPort, vm2, vm3); |
| createReceiverInVMs(vm2, vm3); |
| |
| createCacheInVMs(lnPort, vm4, vm5, vm6, vm7); |
| |
| vm4.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| vm5.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| vm6.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| vm7.invoke(() -> createSender("ln", 2, true, 100, 10, false, false, null, true)); |
| } |
| |
| private void createPartitionedRegions(boolean createAccessors) { |
| String regionName = getUniqueName() + "_PR"; |
| vm4.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm5.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| |
| if (createAccessors) { |
| vm6.invoke(() -> createPartitionedRegionAsAccessor(regionName, "ln", 1, 100)); |
| vm7.invoke(() -> createPartitionedRegionAsAccessor(regionName, "ln", 1, 100)); |
| } else { |
| vm6.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm7.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| } |
| |
| vm2.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| vm3.invoke(() -> createPartitionedRegion(regionName, "ln", 1, 100, isOffHeap())); |
| } |
| |
| private void updateBatchSize(int batchsize) { |
| vm4.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setBatchSize(batchsize); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm5.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setBatchSize(batchsize); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm6.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setBatchSize(batchsize); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm7.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setBatchSize(batchsize); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| } |
| |
| private void updateBatchTimeInterval(int batchTimeInterval) { |
| vm4.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setBatchTimeInterval(batchTimeInterval); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm5.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setBatchTimeInterval(batchTimeInterval); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm6.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setBatchTimeInterval(batchTimeInterval); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm7.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setBatchTimeInterval(batchTimeInterval); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| } |
| |
| private void updateGroupTransactionEvents(boolean groupTransactionEvents) { |
| vm4.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setGroupTransactionEvents(groupTransactionEvents); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm5.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setGroupTransactionEvents(groupTransactionEvents); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm6.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setGroupTransactionEvents(groupTransactionEvents); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm7.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setGroupTransactionEvents(groupTransactionEvents); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| |
| } |
| |
| private void updateGatewayEventFilters(List<GatewayEventFilter> filters) { |
| vm4.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setGatewayEventFilters(filters); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm5.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setGatewayEventFilters(filters); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm6.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setGatewayEventFilters(filters); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| vm7.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| boolean paused = false; |
| if (sender.isRunning() && !sender.isPaused()) { |
| sender.pause(); |
| paused = true; |
| } |
| sender.setGatewayEventFilters(filters); |
| if (paused) { |
| sender.resume(); |
| } |
| }); |
| } |
| |
| private void checkBatchSize(int batchsize) { |
| vm4.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.getBatchSize()).isEqualTo(batchsize); |
| }); |
| vm5.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.getBatchSize()).isEqualTo(batchsize); |
| }); |
| vm6.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.getBatchSize()).isEqualTo(batchsize); |
| }); |
| vm7.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.getBatchSize()).isEqualTo(batchsize); |
| }); |
| } |
| |
| private void checkBatchTimeInterval(int batchTimeInterval) { |
| vm4.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval); |
| }); |
| vm5.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval); |
| }); |
| vm6.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval); |
| }); |
| vm7.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.getBatchTimeInterval()).isEqualTo(batchTimeInterval); |
| }); |
| } |
| |
| private void checkGroupTransactionEvents(boolean groupTransactionEvents) { |
| vm4.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents); |
| }); |
| vm5.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents); |
| }); |
| vm6.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents); |
| }); |
| vm7.invoke(() -> { |
| AbstractGatewaySender sender = (AbstractGatewaySender) cache.getGatewaySender("ln"); |
| assertThat(sender.mustGroupTransactionEvents()).isEqualTo(groupTransactionEvents); |
| }); |
| } |
| |
| private void stopSenders() { |
| vm4.invoke(() -> stopSender("ln")); |
| vm5.invoke(() -> stopSender("ln")); |
| vm6.invoke(() -> stopSender("ln")); |
| vm7.invoke(() -> stopSender("ln")); |
| } |
| |
| private void waitForSendersRunning() { |
| vm4.invoke(() -> waitForSenderRunningState("ln")); |
| vm5.invoke(() -> waitForSenderRunningState("ln")); |
| vm6.invoke(() -> waitForSenderRunningState("ln")); |
| vm7.invoke(() -> waitForSenderRunningState("ln")); |
| } |
| |
| private void waitForAllSendersNotRunning() { |
| vm4.invoke(() -> waitForSenderNonRunningState("ln")); |
| vm5.invoke(() -> waitForSenderNonRunningState("ln")); |
| vm6.invoke(() -> waitForSenderNonRunningState("ln")); |
| vm7.invoke(() -> waitForSenderNonRunningState("ln")); |
| } |
| |
| private void validateParallelSenderQueueAllBucketsDrained() { |
| vm4.invoke(() -> validateParallelSenderQueueAllBucketsDrained("ln")); |
| vm5.invoke(() -> validateParallelSenderQueueAllBucketsDrained("ln")); |
| vm6.invoke(() -> validateParallelSenderQueueAllBucketsDrained("ln")); |
| vm7.invoke(() -> validateParallelSenderQueueAllBucketsDrained("ln")); |
| } |
| |
| private static class CountSentPQRMObserver extends DistributionMessageObserver |
| implements Serializable { |
| private final AtomicInteger numberOfSentPQRM = new AtomicInteger(0); |
| |
| @Override |
| public void beforeSendMessage(ClusterDistributionManager dm, DistributionMessage message) { |
| if (message instanceof ParallelQueueRemovalMessage) { |
| numberOfSentPQRM.addAndGet(message.getRecipients().size()); |
| } |
| } |
| |
| public int getNumberOfSentPQRM() { |
| return numberOfSentPQRM.get(); |
| } |
| } |
| |
| } |
