You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@geode.apache.org by GitBox <gi...@apache.org> on 2021/10/25 10:17:55 UTC
[GitHub] [geode] mivanac opened a new pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
mivanac opened a new pull request #7035:
URL: https://github.com/apache/geode/pull/7035
…new queue region
<!-- Thank you for submitting a contribution to Apache Geode. -->
<!-- In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
-->
### For all changes:
- [*] Is there a JIRA ticket associated with this PR? Is it referenced in the commit message?
- [*] Has your PR been rebased against the latest commit within the target branch (typically `develop`)?
- [*] Is your initial contribution a single, squashed commit?
- [*] Does `gradlew build` run cleanly?
- [*] Have you written or updated unit tests to verify your changes?
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
<!-- Note:
Please ensure that once the PR is submitted, check Concourse for build issues and
submit an update to your PR as soon as possible. If you need help, please send an
email to dev@geode.apache.org.
-->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] boglesby commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
boglesby commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r735835799
##########
File path: geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
##########
@@ -489,7 +489,21 @@ public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
boolean isAccessor = (userPR.getLocalMaxMemory() == 0);
final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());
- prQ = (PartitionedRegion) cache.getRegion(prQName);
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ if (prQ != null && prQ.isDestroyed()) {
+ PartitionedRegion oldPrQ = prQ;
+ while (oldPrQ == prQ) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ logger.info("wait for destroy to finish");
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ }
+ }
Review comment:
Can the region get into a state where it is never destroyed? You might need a way to break out of this loop.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] mivanac commented on pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
mivanac commented on pull request #7035:
URL: https://github.com/apache/geode/pull/7035#issuecomment-966222810
Hi @kirklund, @boglesby, @nabarunnag and @gesterzhou, could you review this PR.
Thanks,
Mario
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r737913250
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2088,210 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived() {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ LogWriterUtils.getLogWriter()
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ LogWriterUtils.getLogWriter().info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ LogWriterUtils.getLogWriter().info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ LogWriterUtils.getLogWriter().info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ LogWriterUtils.getLogWriter().info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ LogWriterUtils.getLogWriter().info("Start all the senders.");
+
+ AsyncInvocation inv1 = vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation inv2 = vm5.invokeAsync(() -> {
+ try {
+ Thread.sleep(200);
Review comment:
Please use `CountDownLatch` instead of `Thread.sleep`. Any uses of sleep are really just going to introduce more flaky tests that fail intermittently in precheckin runs.
You should be able to find lots of examples of uses of `CountDownLatch` within various distributed tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r743210190
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -315,6 +324,10 @@ protected SerializableRunnableIF waitForSenderRunnable() {
return () -> WANTestBase.waitForSenderRunningState("ln");
}
+ protected SerializableRunnableIF waitForSenderNonRunnable() {
Review comment:
`private`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
+ });
+
+ try {
+ waitForConditionInVM7.await();
+ } finally {
+ }
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM7 =
+ vm7.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ startSenderwithCleanQueuesInVM4.await();
+ startSenderwithCleanQueuesInVM5.await();
+ startSenderwithCleanQueuesInVM6.await();
+ startSenderwithCleanQueuesInVM7.await();
+
+ logger.info("Waiting for senders running.");
+ // wait for senders running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All the senders are now running...");
+
+ // ----------------------------------------------------------------------------------------------------
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ }
+
+ private static class BlockingDestroyRegionObserver extends DistributionMessageObserver {
+ private CountDownLatch startedBlocking = new CountDownLatch(1);
Review comment:
Make this `private final`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -39,6 +47,7 @@
public class ParallelWANPersistenceEnabledGatewaySenderDUnitTest extends WANTestBase {
private static final long serialVersionUID = 1L;
+ protected static final Logger logger = LogService.getLogger();
Review comment:
`logger` can be narrowed to `private`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
+ });
+
+ try {
+ waitForConditionInVM7.await();
+ } finally {
+ }
Review comment:
Delete the `try` and `finally`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
Review comment:
You should always use `GeodeAwaitility.getTimeOut` for any await type calls. The timeout was created and set to 5 minutes because lower values (even 2 minutes) results in intermittent flaky failures in the current environments (AWS or GCP):
```
import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
import static java.util.concurrent.TimeUnit.MINUTES;
```
```
observer.startedBlocking.await(getTimeout().toMinutes(), MINUTES);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] mivanac commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
mivanac commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r742285179
##########
File path: geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
##########
@@ -489,7 +489,21 @@ public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
boolean isAccessor = (userPR.getLocalMaxMemory() == 0);
final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());
- prQ = (PartitionedRegion) cache.getRegion(prQName);
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ if (prQ != null && prQ.isDestroyed()) {
+ PartitionedRegion oldPrQ = prQ;
+ while (oldPrQ == prQ) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ logger.info("wait for destroy to finish");
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ }
+ }
Review comment:
I think that this is not possible.
##########
File path: geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
##########
@@ -489,7 +489,21 @@ public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
boolean isAccessor = (userPR.getLocalMaxMemory() == 0);
final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());
- prQ = (PartitionedRegion) cache.getRegion(prQName);
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ if (prQ != null && prQ.isDestroyed()) {
+ PartitionedRegion oldPrQ = prQ;
+ while (oldPrQ == prQ) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ logger.info("wait for destroy to finish");
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ }
+ }
Review comment:
I think that this is not possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r735984826
##########
File path: geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
##########
@@ -489,7 +489,21 @@ public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
boolean isAccessor = (userPR.getLocalMaxMemory() == 0);
final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());
- prQ = (PartitionedRegion) cache.getRegion(prQName);
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ if (prQ != null && prQ.isDestroyed()) {
+ PartitionedRegion oldPrQ = prQ;
+ while (oldPrQ == prQ) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ logger.info("wait for destroy to finish");
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ }
Review comment:
I think this block needs some unit test coverage in addition to the dunit tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] mivanac commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
mivanac commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r742285179
##########
File path: geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
##########
@@ -489,7 +489,21 @@ public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
boolean isAccessor = (userPR.getLocalMaxMemory() == 0);
final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());
- prQ = (PartitionedRegion) cache.getRegion(prQName);
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ if (prQ != null && prQ.isDestroyed()) {
+ PartitionedRegion oldPrQ = prQ;
+ while (oldPrQ == prQ) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ logger.info("wait for destroy to finish");
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ }
+ }
Review comment:
I think that this is not possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] mivanac merged pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
mivanac merged pull request #7035:
URL: https://github.com/apache/geode/pull/7035
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] mivanac commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
mivanac commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r742285179
##########
File path: geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
##########
@@ -489,7 +489,21 @@ public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
boolean isAccessor = (userPR.getLocalMaxMemory() == 0);
final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());
- prQ = (PartitionedRegion) cache.getRegion(prQName);
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ if (prQ != null && prQ.isDestroyed()) {
+ PartitionedRegion oldPrQ = prQ;
+ while (oldPrQ == prQ) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ logger.info("wait for destroy to finish");
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ }
+ }
Review comment:
I think that this is not possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] mivanac commented on pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
mivanac commented on pull request #7035:
URL: https://github.com/apache/geode/pull/7035#issuecomment-976298129
Hi @kirklund,
can you comment, are the changes that you requested met.
Regards,
Mario
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r743210190
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -315,6 +324,10 @@ protected SerializableRunnableIF waitForSenderRunnable() {
return () -> WANTestBase.waitForSenderRunningState("ln");
}
+ protected SerializableRunnableIF waitForSenderNonRunnable() {
Review comment:
`private`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
+ });
+
+ try {
+ waitForConditionInVM7.await();
+ } finally {
+ }
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM7 =
+ vm7.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ startSenderwithCleanQueuesInVM4.await();
+ startSenderwithCleanQueuesInVM5.await();
+ startSenderwithCleanQueuesInVM6.await();
+ startSenderwithCleanQueuesInVM7.await();
+
+ logger.info("Waiting for senders running.");
+ // wait for senders running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All the senders are now running...");
+
+ // ----------------------------------------------------------------------------------------------------
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ }
+
+ private static class BlockingDestroyRegionObserver extends DistributionMessageObserver {
+ private CountDownLatch startedBlocking = new CountDownLatch(1);
Review comment:
Make this `private final`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -39,6 +47,7 @@
public class ParallelWANPersistenceEnabledGatewaySenderDUnitTest extends WANTestBase {
private static final long serialVersionUID = 1L;
+ protected static final Logger logger = LogService.getLogger();
Review comment:
`logger` can be narrowed to `private`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
+ });
+
+ try {
+ waitForConditionInVM7.await();
+ } finally {
+ }
Review comment:
Delete the `try` and `finally`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
Review comment:
You should always use `GeodeAwaitility.getTimeOut` for any await type calls. The timeout was created and set to 5 minutes because lower values (even 2 minutes) results in intermittent flaky failures in the current environments (AWS or GCP):
```
import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
import static java.util.concurrent.TimeUnit.MINUTES;
```
```
observer.startedBlocking.await(getTimeout().toMinutes(), MINUTES);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r743210190
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -315,6 +324,10 @@ protected SerializableRunnableIF waitForSenderRunnable() {
return () -> WANTestBase.waitForSenderRunningState("ln");
}
+ protected SerializableRunnableIF waitForSenderNonRunnable() {
Review comment:
`private`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
+ });
+
+ try {
+ waitForConditionInVM7.await();
+ } finally {
+ }
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM7 =
+ vm7.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ startSenderwithCleanQueuesInVM4.await();
+ startSenderwithCleanQueuesInVM5.await();
+ startSenderwithCleanQueuesInVM6.await();
+ startSenderwithCleanQueuesInVM7.await();
+
+ logger.info("Waiting for senders running.");
+ // wait for senders running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All the senders are now running...");
+
+ // ----------------------------------------------------------------------------------------------------
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ }
+
+ private static class BlockingDestroyRegionObserver extends DistributionMessageObserver {
+ private CountDownLatch startedBlocking = new CountDownLatch(1);
Review comment:
Make this `private final`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -39,6 +47,7 @@
public class ParallelWANPersistenceEnabledGatewaySenderDUnitTest extends WANTestBase {
private static final long serialVersionUID = 1L;
+ protected static final Logger logger = LogService.getLogger();
Review comment:
`logger` can be narrowed to `private`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
+ });
+
+ try {
+ waitForConditionInVM7.await();
+ } finally {
+ }
Review comment:
Delete the `try` and `finally`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2097,152 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived()
+ throws InterruptedException {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ logger
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ vm7.invoke(() -> {
+ DistributionMessageObserver.setInstance(new BlockingDestroyRegionObserver());
+ });
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ logger.info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ logger.info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ logger.info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ logger.info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ logger.info("Start all the senders.");
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 =
+ vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM5 =
+ vm5.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> startSenderwithCleanQueuesInVM6 =
+ vm6.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation<Void> waitForConditionInVM7 = vm7.invokeAsync(() -> {
+ BlockingDestroyRegionObserver observer =
+ (BlockingDestroyRegionObserver) DistributionMessageObserver.getInstance();
+ observer.startedBlocking.await(1, TimeUnit.MINUTES);
Review comment:
You should always use `GeodeAwaitility.getTimeOut` for any await type calls. The timeout was created and set to 5 minutes because lower values (even 2 minutes) results in intermittent flaky failures in the current environments (AWS or GCP):
```
import static org.apache.geode.test.awaitility.GeodeAwaitility.getTimeout;
import static java.util.concurrent.TimeUnit.MINUTES;
```
```
observer.startedBlocking.await(getTimeout().toMinutes(), MINUTES);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #7035: GEODE-9768: wait for destroying of region to finish, before creating …
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #7035:
URL: https://github.com/apache/geode/pull/7035#discussion_r737910918
##########
File path: geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java
##########
@@ -489,7 +489,21 @@ public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
boolean isAccessor = (userPR.getLocalMaxMemory() == 0);
final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());
- prQ = (PartitionedRegion) cache.getRegion(prQName);
+
+ prQ = (PartitionedRegion) cache.getRegion(prQName, true);
+ if (prQ != null && prQ.isDestroyed()) {
+ PartitionedRegion oldPrQ = prQ;
+ while (oldPrQ == prQ) {
+ try {
+ Thread.sleep(50);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ logger.info("wait for destroy to finish");
Review comment:
I just realized this is `info` level which is turned on by default. The wording looks like it should be a `debug` statement.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2088,210 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived() {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ LogWriterUtils.getLogWriter()
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ LogWriterUtils.getLogWriter().info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ LogWriterUtils.getLogWriter().info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ LogWriterUtils.getLogWriter().info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ LogWriterUtils.getLogWriter().info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ LogWriterUtils.getLogWriter().info("Start all the senders.");
+
+ AsyncInvocation inv1 = vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
Review comment:
You should type these AsyncInvocations with Void: `AsyncInvocation<Void>`
And I would give them names that describe "what" and "where":
```
AsyncInvocation<Void> startSenderwithCleanQueuesInVM4 = vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
```
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2088,210 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived() {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ LogWriterUtils.getLogWriter()
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ LogWriterUtils.getLogWriter().info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ LogWriterUtils.getLogWriter().info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ LogWriterUtils.getLogWriter().info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ LogWriterUtils.getLogWriter().info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ LogWriterUtils.getLogWriter().info("Start all the senders.");
+
+ AsyncInvocation inv1 = vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation inv2 = vm5.invokeAsync(() -> {
+ try {
+ Thread.sleep(200);
Review comment:
Please use `CountDownLatch` instead of `Thread.sleep`. Any uses of sleep are really just going to introduce more flaky tests that fail intermittently in precheckin runs.
You should be able to find lots of examples of uses of CountDownLatch within various distributed tests.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2088,210 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived() {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ LogWriterUtils.getLogWriter()
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
+
+ // create PR on remote site
+ vm2.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+ vm3.invoke(
+ () -> WANTestBase.createPartitionedRegion(getTestMethodName(), null, 1, 100, isOffHeap()));
+
+ // create PR on local site
+ vm4.invoke(createPartitionedRegionRunnable());
+ vm5.invoke(createPartitionedRegionRunnable());
+ vm6.invoke(createPartitionedRegionRunnable());
+ vm7.invoke(createPartitionedRegionRunnable());
+
+ // start the senders on local site
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ // wait for senders to become running
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ LogWriterUtils.getLogWriter().info("All senders are running.");
+
+ // start puts in region on local site
+ vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
+ LogWriterUtils.getLogWriter().info("Completed puts in the region");
+
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ LogWriterUtils.getLogWriter().info("Check that no events are propagated to remote site");
+
+ // --------------------close and rebuild local site
+ // -------------------------------------------------
+ // stop the senders
+
+ vm4.invoke(() -> WANTestBase.stopSender("ln"));
+ vm5.invoke(() -> WANTestBase.stopSender("ln"));
+ vm6.invoke(() -> WANTestBase.stopSender("ln"));
+ vm7.invoke(() -> WANTestBase.stopSender("ln"));
+
+ LogWriterUtils.getLogWriter().info("Stopped all the senders.");
+
+ // wait for senders to stop
+ vm4.invoke(waitForSenderNonRunnable());
+ vm5.invoke(waitForSenderNonRunnable());
+ vm6.invoke(waitForSenderNonRunnable());
+ vm7.invoke(waitForSenderNonRunnable());
+
+
+ // create receiver on remote site
+ createReceiverInVMs(vm2, vm3);
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+ vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 0));
+
+ LogWriterUtils.getLogWriter().info("Start all the senders.");
+
+ AsyncInvocation inv1 = vm4.invokeAsync(() -> startSenderwithCleanQueues("ln"));
+ AsyncInvocation inv2 = vm5.invokeAsync(() -> {
+ try {
+ Thread.sleep(200);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ startSenderwithCleanQueues("ln");
+ });
+ AsyncInvocation inv3 = vm6.invokeAsync(() -> {
+ try {
+ Thread.sleep(250);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ startSenderwithCleanQueues("ln");
+ });
+ AsyncInvocation inv4 = vm7.invokeAsync(() -> {
+ try {
+ Thread.sleep(300);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ startSenderwithCleanQueues("ln");
+ });
+
+ try {
+ inv1.join();
+ inv2.join();
+ inv3.join();
+ inv4.join();
+ } catch (InterruptedException e) {
+ e.printStackTrace();
+ fail();
+ }
Review comment:
It's much better to add `throws InterruptedException` to the method and let Java propagate it out. When the exception reaches JUnit, JUnit will provide a much better stack trace and failure message than what this does.
You should also just use `await()`:
```
inv1.await();
inv2.await();
inv3.await();
inv4.await();
```
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -1435,6 +1435,21 @@ public static void waitForSenderRunningState(String senderId) {
}
}
+
+ public static void waitForSenderNonRunningState(String senderId) {
+ final IgnoredException exln = IgnoredException.addIgnoredException("Could not connect");
+ try {
Review comment:
If you want to minimize the scope of the `IgnoredException`, you might want to use try-with-resource syntax:
```
try (IgnoredException ie = IgnoredException.addIgnoredException("Could not connect")) {
...
}
```
`IgnoredException` implements `AutoCloseable` by invoking `remove` for you.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java
##########
@@ -2084,6 +2088,210 @@ public void testPersistentPRWithGatewaySenderPersistenceEnabled_RestartAndStopSe
}
+ /**
+ * Enable persistence for GatewaySender. Pause the sender and do some puts in local region. Stop
+ * GatewaySender.
+ * Then start GatewaySender with clean-queues option. Check if the remote site receives all the
+ * events.
+ */
+ @Test
+ public void testpersistentWanGateway_restartSenderWithCleanQueuesDelayed_expectNoEventsReceived() {
+ // create locator on local site
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ // create locator on remote site
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ // create cache in remote site
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ // create cache in local site
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ // create senders with disk store
+ String diskStore1 = (String) vm4.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore2 = (String) vm5.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore3 = (String) vm6.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+ String diskStore4 = (String) vm7.invoke(() -> WANTestBase.createSenderWithDiskStore("ln", 2,
+ true, 100, 10, false, true, null, null, true));
+
+ LogWriterUtils.getLogWriter()
+ .info("The DS are: " + diskStore1 + "," + diskStore2 + "," + diskStore3 + "," + diskStore4);
Review comment:
`LogWriterUtils` is deprecated in favor of using a Log4j2 logger:
```
private static final Logger logger = LogService.getLogger();
logger.info("The disk stores are: {}, {}, {}, {}", diskStore1, diskStore2, diskStore3, diskStore4);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org