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 2020/08/25 08:18:51 UTC
[GitHub] [geode] albertogpz opened a new pull request #5476: GEODE-8455: Fix difference between create region with gw sender and a…
albertogpz opened a new pull request #5476:
URL: https://github.com/apache/geode/pull/5476
…lter region with gw sender
Geode behaves differently with respect to transactions when
creating a partitioned region with a parallel gateway sender to
when first the partitioned region is created and then the
parallel gateway sender is added by altering the region.
In the first case (create region with gateway sender), when
sending a transaction for the region, an event is sent
to each server hosting the bucket for the data in the transaction.
In the second case (create region + alter region),
when sending a transaction for the region, an event
is sent to every cache member and not only to those hosting
the bucket for the data in the transaction. This leads to,
in the servers not hosting the bucket for the data,
that the events are stored in the
bucketToTempQueueMap member variable of the
ParallelGatewaySenderQueue.
The behavior in the first case is the right one.
The wrong behavior is provoked by the setting to true of the
requiresNotification variable of the PartitionedRegion
by the distributeUpdatedProfileOnSenderCreation method.
This method should only set this variable to true
in case the sender is serial as it is done when
a partitioned region is initialized.
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] DonalEvans commented on a change in pull request #5476: GEODE-8455: Fix difference between create region with gw sender and a…
Posted by GitBox <gi...@apache.org>.
DonalEvans commented on a change in pull request #5476:
URL: https://github.com/apache/geode/pull/5476#discussion_r477509049
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -3197,6 +3114,18 @@ public static Boolean killSender(String senderId) {
}
}
+ private static AbstractGatewaySender getAbstractGatewaySender(String senderId) {
Review comment:
This method almost entirely duplicates the existing `getGatewaySender()` method, so it can be removed and calls to it replaced with `(AbstractGatewaySender) getGatewaySender(senderId)`
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPropagationDUnitTest.java
##########
@@ -1131,6 +1141,111 @@ public void testParallelPropagationWithFilter_AfterAck() throws Exception {
}
+ /**
+ * Test that, when a parallel gateway sender is added to a partitioned region through attributes
+ * mutator, transaction events are not sent to all region members but only to those who are
+ * hosting the bucket for the event and thus, events are not stored in the bucketToTempQueueMap
+ * member of the ParallelGatewaySenderQueue.
+ * Redundancy = 1 in the partitioned region.
+ *
+ */
+ @Test
+ public void testParallelPropagationTxNotificationsNotSentToAllRegionMembersWhenAddingParallelGatewaySenderThroughAttributesMutator()
+ throws Exception {
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ vm4.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
+ vm5.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
+ vm6.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
+ vm7.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
+
+ vm4.invoke(createPartitionedRegionRedundancy1RunnableNoSenders());
+ vm5.invoke(createPartitionedRegionRedundancy1RunnableNoSenders());
+ vm6.invoke(createPartitionedRegionRedundancy1RunnableNoSenders());
+ vm7.invoke(createPartitionedRegionRedundancy1RunnableNoSenders());
+
+ vm2.invoke(createReceiverPartitionedRegionRedundancy1());
+ vm3.invoke(createReceiverPartitionedRegionRedundancy1());
+
+ vm4.invoke(() -> addSenderThroughAttributesMutator(getTestMethodName() + "_PR", "ln"));
+ vm5.invoke(() -> addSenderThroughAttributesMutator(getTestMethodName() + "_PR", "ln"));
+ vm6.invoke(() -> addSenderThroughAttributesMutator(getTestMethodName() + "_PR", "ln"));
+ vm7.invoke(() -> addSenderThroughAttributesMutator(getTestMethodName() + "_PR", "ln"));
+
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ vm4.invoke(() -> WANTestBase.doTxPuts(getTestMethodName() + "_PR"));
+
+ vm4.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_PR", 3));
+ vm4.invoke(() -> WANTestBase.verifyQueueSize("ln", 3));
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_PR", 0));
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_PR", 0));
+
+ vm4.invoke(() -> WANTestBase.validateEmptyBucketToTempQueueMap("ln"));
+ vm5.invoke(() -> WANTestBase.validateEmptyBucketToTempQueueMap("ln"));
+ vm6.invoke(() -> WANTestBase.validateEmptyBucketToTempQueueMap("ln"));
+ vm7.invoke(() -> WANTestBase.validateEmptyBucketToTempQueueMap("ln"));
+ }
+
+ /**
+ * Test that, when a parallel gateway sender is added to a partitioned region through attributes
+ * mutator, transaction events are not sent to all region members but only to those who are
+ * hosting the bucket for the event and thus, events are not stored in the bucketToTempQueueMap
+ * member of the ParallelGatewaySenderQueue.
+ * No redundancy in the partitioned region.
+ *
+ */
+ @Test
+ public void testParallelPropagationTxNotificationsNotSentToAllRegionMembersWhenAddingParallelGatewaySenderThroughAttributesMutatorNoRedundancy()
+ throws Exception {
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
Review comment:
These casts to `Integer` are redundant and can be removed.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPropagationDUnitTest.java
##########
@@ -190,6 +190,16 @@ protected SerializableRunnableIF createPartitionedRegionRedundancy1Runnable() {
isOffHeap());
}
+ protected SerializableRunnableIF createPartitionedRegionRedundancy1RunnableNoSenders() {
+ return () -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", null, 1, 100,
+ isOffHeap());
+ }
+
+ protected SerializableRunnableIF createPartitionedRegionRedundancy0RunnableNoSenders() {
+ return () -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", null, 0, 100,
+ isOffHeap());
+ }
+
Review comment:
These methods are very similar. Would it be possible to replace them with one method which takes an `int` argument for the desired redundancy level?
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPropagationDUnitTest.java
##########
@@ -1131,6 +1141,111 @@ public void testParallelPropagationWithFilter_AfterAck() throws Exception {
}
+ /**
+ * Test that, when a parallel gateway sender is added to a partitioned region through attributes
+ * mutator, transaction events are not sent to all region members but only to those who are
+ * hosting the bucket for the event and thus, events are not stored in the bucketToTempQueueMap
+ * member of the ParallelGatewaySenderQueue.
+ * Redundancy = 1 in the partitioned region.
+ *
+ */
+ @Test
+ public void testParallelPropagationTxNotificationsNotSentToAllRegionMembersWhenAddingParallelGatewaySenderThroughAttributesMutator()
+ throws Exception {
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
Review comment:
These casts to `Integer` are redundant and can be removed.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -3599,21 +3444,42 @@ public static Long getNumberOfEntriesInVM(final String senderId) {
}
public static void verifyTmpDroppedEventSize(String senderId, int size) {
- Set<GatewaySender> senders = cache.getGatewaySenders();
- GatewaySender sender = null;
- for (GatewaySender s : senders) {
- if (s.getId().equals(senderId)) {
- sender = s;
- break;
- }
- }
+ GatewaySender sender = getGatewaySender(senderId);
AbstractGatewaySender ags = (AbstractGatewaySender) sender;
await().untilAsserted(() -> assertEquals("Expected tmpDroppedEvents size: " + size
+ " but actual size: " + ags.getTmpDroppedEventSize(), size, ags.getTmpDroppedEventSize()));
}
- public static void verifyQueueSize(String senderId, int size) {
+ /**
+ * Checks that the bucketToTempQueueMap for a partitioned region
+ * that holds events for buckets that are not available locally, is empty.
+ */
+ public static void validateEmptyBucketToTempQueueMap(String senderId) {
+ GatewaySender sender = getGatewaySender(senderId);
+
+ int size = 0;
+ Set queues = ((AbstractGatewaySender) sender).getQueues();
Review comment:
This should be `Set<RegionQueue>` to avoid compiler warnings.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -3599,21 +3444,42 @@ public static Long getNumberOfEntriesInVM(final String senderId) {
}
public static void verifyTmpDroppedEventSize(String senderId, int size) {
- Set<GatewaySender> senders = cache.getGatewaySenders();
- GatewaySender sender = null;
- for (GatewaySender s : senders) {
- if (s.getId().equals(senderId)) {
- sender = s;
- break;
- }
- }
+ GatewaySender sender = getGatewaySender(senderId);
AbstractGatewaySender ags = (AbstractGatewaySender) sender;
await().untilAsserted(() -> assertEquals("Expected tmpDroppedEvents size: " + size
+ " but actual size: " + ags.getTmpDroppedEventSize(), size, ags.getTmpDroppedEventSize()));
}
- public static void verifyQueueSize(String senderId, int size) {
+ /**
+ * Checks that the bucketToTempQueueMap for a partitioned region
+ * that holds events for buckets that are not available locally, is empty.
+ */
+ public static void validateEmptyBucketToTempQueueMap(String senderId) {
+ GatewaySender sender = getGatewaySender(senderId);
+
+ int size = 0;
+ Set queues = ((AbstractGatewaySender) sender).getQueues();
+ for (Object queue : queues) {
+ PartitionedRegion region =
+ (PartitionedRegion) ((ConcurrentParallelGatewaySenderQueue) queue).getRegion();
+ int buckets = region.getTotalNumberOfBuckets();
+ for (int bucket = 0; bucket < buckets; bucket++) {
+ BlockingQueue newQueue =
+ ((ConcurrentParallelGatewaySenderQueue) queue).getBucketTmpQueue((int) bucket);
Review comment:
The cast to `int` here is unnecessary.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/WANTestBase.java
##########
@@ -3599,21 +3444,42 @@ public static Long getNumberOfEntriesInVM(final String senderId) {
}
public static void verifyTmpDroppedEventSize(String senderId, int size) {
- Set<GatewaySender> senders = cache.getGatewaySenders();
- GatewaySender sender = null;
- for (GatewaySender s : senders) {
- if (s.getId().equals(senderId)) {
- sender = s;
- break;
- }
- }
+ GatewaySender sender = getGatewaySender(senderId);
AbstractGatewaySender ags = (AbstractGatewaySender) sender;
await().untilAsserted(() -> assertEquals("Expected tmpDroppedEvents size: " + size
+ " but actual size: " + ags.getTmpDroppedEventSize(), size, ags.getTmpDroppedEventSize()));
}
- public static void verifyQueueSize(String senderId, int size) {
+ /**
+ * Checks that the bucketToTempQueueMap for a partitioned region
+ * that holds events for buckets that are not available locally, is empty.
+ */
+ public static void validateEmptyBucketToTempQueueMap(String senderId) {
+ GatewaySender sender = getGatewaySender(senderId);
+
+ int size = 0;
+ Set queues = ((AbstractGatewaySender) sender).getQueues();
+ for (Object queue : queues) {
+ PartitionedRegion region =
+ (PartitionedRegion) ((ConcurrentParallelGatewaySenderQueue) queue).getRegion();
+ int buckets = region.getTotalNumberOfBuckets();
+ for (int bucket = 0; bucket < buckets; bucket++) {
+ BlockingQueue newQueue =
Review comment:
This should be `BlockingQueue<GatewaySenderEventImpl>` to avoid compiler warnings.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPropagationDUnitTest.java
##########
@@ -1131,6 +1141,111 @@ public void testParallelPropagationWithFilter_AfterAck() throws Exception {
}
+ /**
+ * Test that, when a parallel gateway sender is added to a partitioned region through attributes
+ * mutator, transaction events are not sent to all region members but only to those who are
+ * hosting the bucket for the event and thus, events are not stored in the bucketToTempQueueMap
+ * member of the ParallelGatewaySenderQueue.
+ * Redundancy = 1 in the partitioned region.
+ *
+ */
+ @Test
+ public void testParallelPropagationTxNotificationsNotSentToAllRegionMembersWhenAddingParallelGatewaySenderThroughAttributesMutator()
+ throws Exception {
+ Integer lnPort = (Integer) vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
+ Integer nyPort = (Integer) vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));
+
+ createCacheInVMs(nyPort, vm2, vm3);
+
+ createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);
+
+ vm4.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
+ vm5.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
+ vm6.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
+ vm7.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
+
+ vm4.invoke(createPartitionedRegionRedundancy1RunnableNoSenders());
+ vm5.invoke(createPartitionedRegionRedundancy1RunnableNoSenders());
+ vm6.invoke(createPartitionedRegionRedundancy1RunnableNoSenders());
+ vm7.invoke(createPartitionedRegionRedundancy1RunnableNoSenders());
+
+ vm2.invoke(createReceiverPartitionedRegionRedundancy1());
+ vm3.invoke(createReceiverPartitionedRegionRedundancy1());
+
+ vm4.invoke(() -> addSenderThroughAttributesMutator(getTestMethodName() + "_PR", "ln"));
+ vm5.invoke(() -> addSenderThroughAttributesMutator(getTestMethodName() + "_PR", "ln"));
+ vm6.invoke(() -> addSenderThroughAttributesMutator(getTestMethodName() + "_PR", "ln"));
+ vm7.invoke(() -> addSenderThroughAttributesMutator(getTestMethodName() + "_PR", "ln"));
+
+ startSenderInVMs("ln", vm4, vm5, vm6, vm7);
+
+ vm4.invoke(waitForSenderRunnable());
+ vm5.invoke(waitForSenderRunnable());
+ vm6.invoke(waitForSenderRunnable());
+ vm7.invoke(waitForSenderRunnable());
+
+ vm4.invoke(() -> WANTestBase.doTxPuts(getTestMethodName() + "_PR"));
+
+ vm4.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_PR", 3));
+ vm4.invoke(() -> WANTestBase.verifyQueueSize("ln", 3));
+
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_PR", 0));
+ vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_PR", 0));
+
+ vm4.invoke(() -> WANTestBase.validateEmptyBucketToTempQueueMap("ln"));
+ vm5.invoke(() -> WANTestBase.validateEmptyBucketToTempQueueMap("ln"));
+ vm6.invoke(() -> WANTestBase.validateEmptyBucketToTempQueueMap("ln"));
+ vm7.invoke(() -> WANTestBase.validateEmptyBucketToTempQueueMap("ln"));
+ }
+
+ /**
+ * Test that, when a parallel gateway sender is added to a partitioned region through attributes
+ * mutator, transaction events are not sent to all region members but only to those who are
+ * hosting the bucket for the event and thus, events are not stored in the bucketToTempQueueMap
+ * member of the ParallelGatewaySenderQueue.
+ * No redundancy in the partitioned region.
+ *
+ */
+ @Test
+ public void testParallelPropagationTxNotificationsNotSentToAllRegionMembersWhenAddingParallelGatewaySenderThroughAttributesMutatorNoRedundancy()
+ throws Exception {
Review comment:
No exception is thrown from this method, so this can be removed.
##########
File path: geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPropagationDUnitTest.java
##########
@@ -1131,6 +1141,111 @@ public void testParallelPropagationWithFilter_AfterAck() throws Exception {
}
+ /**
+ * Test that, when a parallel gateway sender is added to a partitioned region through attributes
+ * mutator, transaction events are not sent to all region members but only to those who are
+ * hosting the bucket for the event and thus, events are not stored in the bucketToTempQueueMap
+ * member of the ParallelGatewaySenderQueue.
+ * Redundancy = 1 in the partitioned region.
+ *
+ */
+ @Test
+ public void testParallelPropagationTxNotificationsNotSentToAllRegionMembersWhenAddingParallelGatewaySenderThroughAttributesMutator()
+ throws Exception {
Review comment:
No exception is thrown from this method, so this can be removed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] mkevo merged pull request #5476: GEODE-8455: Fix difference between create region with gw sender and a…
Posted by GitBox <gi...@apache.org>.
mkevo merged pull request #5476:
URL: https://github.com/apache/geode/pull/5476
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] albertogpz commented on pull request #5476: GEODE-8455: Fix difference between create region with gw sender and a…
Posted by GitBox <gi...@apache.org>.
albertogpz commented on pull request #5476:
URL: https://github.com/apache/geode/pull/5476#issuecomment-681917316
> A few small changes to clean up the code a bit. I also notice that there are several DUnit test failures associated with this change (more than we would expect just due to the known flakiness of some WAN DUnit tests) so I would prefer not to merge this change until they have been diagnosed and remedied.
>
> Also, while it's not strictly within the scope of this PR, I notice that in `WANTestBase.createSender()` line 1747, the call to `gateway.create(dsName, remoteDsId);` comes before the call to `gateway.setBatchTimeInterval(batchTimeInterval);` which means the second call has no effect on the batch time interval. If fixing this has no effect on existing tests, then it would be okay to include it as part of this PR, I think. If fixing it causes tests to start failing, a separate JIRA ticket should be created to address the changes.
@DonalEvans Thanks a lot for your comments. You have an incredible eye to find every error and identify items to improve.
- I have added a fix for the distributed tests that were failing.
- I have incorporated all the fixes you have suggested in your comments.
- I have also tackled the issue you mentioned regarding `WANTestBase.createSender()` when specifying a `batchTimeInterval`. The way to address it has been to remove the parameter from the method as it had no effect anyway. I think it was me the one who introduced it in the first place when I implemented the "GEODE-7971" and was trying to fix some flaky tests. But obviously, that parameter did not have anything to do to the fix of the test cases.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org