[
https://issues.apache.org/jira/browse/GEODE-7670?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17148192#comment-17148192
]
ASF GitHub Bot commented on GEODE-7670:
---------------------------------------
agingade commented on a change in pull request #4848:
URL: https://github.com/apache/geode/pull/4848#discussion_r447308423
##########
File path:
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionClearWithConcurrentOperationsDUnitTest.java
##########
@@ -0,0 +1,680 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
contributor license
+ * agreements. See the NOTICE file distributed with this work for additional
information regarding
+ * copyright ownership. The ASF licenses this file to You under the Apache
License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance with the
License. You may obtain a
+ * copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express
+ * or implied. See the License for the specific language governing permissions
and limitations under
+ * the License.
+ */
+package org.apache.geode.internal.cache;
+
+import static org.apache.geode.internal.util.ArrayUtils.asList;
+import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
+import static org.apache.geode.test.dunit.VM.getVM;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.IntStream;
+
+import junitparams.JUnitParamsRunner;
+import junitparams.Parameters;
+import junitparams.naming.TestCaseName;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.apache.geode.ForcedDisconnectException;
+import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheWriter;
+import org.apache.geode.cache.CacheWriterException;
+import org.apache.geode.cache.PartitionAttributes;
+import org.apache.geode.cache.PartitionAttributesFactory;
+import org.apache.geode.cache.PartitionedRegionPartialClearException;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionEvent;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.cache.partition.PartitionRegionHelper;
+import org.apache.geode.cache.util.CacheWriterAdapter;
+import org.apache.geode.distributed.DistributedSystemDisconnectedException;
+import org.apache.geode.distributed.internal.DMStats;
+import org.apache.geode.distributed.internal.InternalDistributedSystem;
+import
org.apache.geode.distributed.internal.membership.api.MembershipManagerHelper;
+import org.apache.geode.test.dunit.AsyncInvocation;
+import org.apache.geode.test.dunit.VM;
+import org.apache.geode.test.dunit.rules.CacheRule;
+import org.apache.geode.test.dunit.rules.DistributedRule;
+
+/**
+ * Tests to verify that {@link PartitionedRegion#clear()} operation can be
executed multiple times
+ * on the same region while other cache operations are being executed
concurrently and members are
+ * added or removed.
+ */
+@RunWith(JUnitParamsRunner.class)
+public class PartitionedRegionClearWithConcurrentOperationsDUnitTest
implements Serializable {
+ private static final Integer BUCKETS = 13;
+ private static final String REGION_NAME = "PartitionedRegion";
+ private static final String TEST_CASE_NAME =
+ "[{index}] {method}(Coordinator:{0}, RegionType:{1})";
+
+ @Rule
+ public DistributedRule distributedRule = new DistributedRule(3);
+
+ @Rule
+ public CacheRule cacheRule = CacheRule.builder().createCacheInAll().build();
+
+ private VM accessor, server1, server2;
+
+ private enum TestVM {
+ ACCESSOR(0), SERVER1(1), SERVER2(2);
+
+ final int vmNumber;
+
+ TestVM(int vmNumber) {
+ this.vmNumber = vmNumber;
+ }
+ }
+
+ @SuppressWarnings("unused")
+ static RegionShortcut[] regionTypes() {
+ return new RegionShortcut[] {
+ RegionShortcut.PARTITION,
+ RegionShortcut.PARTITION_REDUNDANT
+ };
+ }
+
+ @SuppressWarnings("unused")
+ static TestVM[] vms() {
+ return new TestVM[] {
+ TestVM.SERVER1, TestVM.ACCESSOR
+ };
+ }
+
+ @SuppressWarnings("unused")
+ static Object[] vmAndRegionTypes() {
+ ArrayList<Object[]> parameters = new ArrayList<>();
+ RegionShortcut[] regionShortcuts = regionTypes();
+
+ Arrays.stream(regionShortcuts).forEach(regionShortcut -> {
+ parameters.add(new Object[] {TestVM.SERVER1, regionShortcut});
+ parameters.add(new Object[] {TestVM.ACCESSOR, regionShortcut});
+ });
+
+ return parameters.toArray();
+ }
+
+ @Before
+ public void setUp() throws Exception {
+ server1 = getVM(TestVM.SERVER1.vmNumber);
+ server2 = getVM(TestVM.SERVER2.vmNumber);
+ accessor = getVM(TestVM.ACCESSOR.vmNumber);
+ }
+
+ private void initAccessor(RegionShortcut regionShortcut) {
+ @SuppressWarnings("rawtypes")
+ PartitionAttributes attributes = new PartitionAttributesFactory<String,
String>()
+ .setTotalNumBuckets(BUCKETS)
+ .setLocalMaxMemory(0)
+ .create();
+
+ cacheRule.getCache().createRegionFactory(regionShortcut)
+ .setPartitionAttributes(attributes)
+ .create(REGION_NAME);
+
+ }
+
+ private void initDataStore(RegionShortcut regionShortcut) {
+ @SuppressWarnings("rawtypes")
+ PartitionAttributes attributes = new PartitionAttributesFactory<String,
String>()
+ .setTotalNumBuckets(BUCKETS)
+ .create();
+
+ cacheRule.getCache().createRegionFactory(regionShortcut)
+ .setPartitionAttributes(attributes)
+ .create(REGION_NAME);
+ }
+
+ private void parametrizedSetup(RegionShortcut regionShortcut) {
+ server1.invoke(() -> initDataStore(regionShortcut));
+ server2.invoke(() -> initDataStore(regionShortcut));
+ accessor.invoke(() -> initAccessor(regionShortcut));
+ }
+
+ private void waitForSilence() {
+ DMStats dmStats =
cacheRule.getSystem().getDistributionManager().getStats();
+ PartitionedRegion region = (PartitionedRegion)
cacheRule.getCache().getRegion(REGION_NAME);
+ PartitionedRegionStats partitionedRegionStats = region.getPrStats();
+
+ await().untilAsserted(() -> {
+ assertThat(dmStats.getReplyWaitsInProgress()).isEqualTo(0);
+
assertThat(partitionedRegionStats.getVolunteeringInProgress()).isEqualTo(0);
+
assertThat(partitionedRegionStats.getBucketCreatesInProgress()).isEqualTo(0);
+
assertThat(partitionedRegionStats.getPrimaryTransfersInProgress()).isEqualTo(0);
+
assertThat(partitionedRegionStats.getRebalanceBucketCreatesInProgress()).isEqualTo(0);
+
assertThat(partitionedRegionStats.getRebalancePrimaryTransfersInProgress()).isEqualTo(0);
+ });
+ }
+
+ /**
+ * Populates the region and verifies the data on the selected VMs.
+ */
+ private void populateRegion(VM feeder, int entryCount, List<VM> vms) {
+ feeder.invoke(() -> {
+ Region<String, String> region =
cacheRule.getCache().getRegion(REGION_NAME);
+ IntStream.range(0, entryCount).forEach(i ->
region.put(String.valueOf(i), "Value_" + i));
+ });
+
+ vms.forEach(vm -> vm.invoke(() -> {
+ waitForSilence();
+ Region<String, String> region =
cacheRule.getCache().getRegion(REGION_NAME);
+
+ IntStream.range(0, entryCount)
+ .forEach(i ->
assertThat(region.get(String.valueOf(i))).isEqualTo("Value_" + i));
+ }));
+ }
+
+ /**
+ * Asserts that the region is empty on requested VMs.
+ */
+ private void assertRegionIsEmpty(List<VM> vms) {
+ vms.forEach(vm -> vm.invoke(() -> {
+ waitForSilence();
+ PartitionedRegion region = (PartitionedRegion)
cacheRule.getCache().getRegion(REGION_NAME);
+
+ assertThat(region.getLocalSize()).isEqualTo(0);
+ }));
+ }
+
+ /**
+ * Asserts that the region data is consistent across buckets.
+ */
+ private void assertRegionBucketsConsistency() throws ForceReattemptException
{
+ List<BucketDump> bucketDumps;
+ PartitionedRegion region = (PartitionedRegion)
cacheRule.getCache().getRegion(REGION_NAME);
+ // Redundant copies + 1 primary.
+ int expectedCopies = region.getRedundantCopies() + 1;
+
+ for (int bucketId = 0; bucketId < BUCKETS; bucketId++) {
+ bucketDumps = region.getAllBucketEntries(bucketId);
+ assertThat(bucketDumps.size())
+ .as("Bucket " + bucketId + " should have " + expectedCopies + "
copies, but has "
+ + bucketDumps.size())
+ .isEqualTo(expectedCopies);
+
+ // Check that all copies of the bucket have the same data.
+ if (bucketDumps.size() > 1) {
+ BucketDump firstDump = bucketDumps.get(0);
+
+ for (int j = 1; j < bucketDumps.size(); j++) {
+ BucketDump otherDump = bucketDumps.get(j);
+
+ assertThat(otherDump.getRvv())
Review comment:
This validation needs to be changed....The RVV's are maintained as
local-rvv (local to the node/member) and array of remote-rvvs (all other
members). The RVVs has to be taken for corresponding member from both dumps and
compared.
----------------------------------------------------------------
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:
us...@infra.apache.org
> Partitioned Region clear operations can occur during concurrent data
> operations
> -------------------------------------------------------------------------------
>
> Key: GEODE-7670
> URL: https://issues.apache.org/jira/browse/GEODE-7670
> Project: Geode
> Issue Type: Sub-task
> Components: regions
> Reporter: Nabarun Nag
> Assignee: Juan Ramos
> Priority: Major
> Labels: GeodeCommons
> Time Spent: 5.5h
> Remaining Estimate: 0h
>
> Clear operations are successful when concurrent read/write operations occur.
> Ensure there are test coverage for this use case and modify the code needed
> to enable this.
> Acceptance :
> * Passing DUnit tests where clear operations are successful on partitioned
> region withÂ
> * concurrent puts (writes) and clear op
> * concurrent gets (reads) and clear op
> * Test coverage to when a member departs in this scenario
> * Test coverage to when a member restarts in this scenario
> * Unit tests with complete code coverage for the newly written code.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
