This is an automated email from the ASF dual-hosted git repository.
boglesby pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/develop by this push:
new 12e17d8 GEODE-5478: Modified to not increment low redundancy bucket
count twice
12e17d8 is described below
commit 12e17d80629bfd4193b3bcd603a75eb9694c8107
Author: Barry Oglesby <[email protected]>
AuthorDate: Fri Jul 27 13:34:05 2018 -0700
GEODE-5478: Modified to not increment low redundancy bucket count twice
Co-authored-by: Darrel Schneider <[email protected]>
---
...edRegionLowBucketRedundancyDistributedTest.java | 208 +++++++++++++++++++++
.../internal/cache/BucketRedundancyTracker.java | 69 ++++---
.../cache/BucketRedundancyTrackerTest.java | 28 ++-
3 files changed, 272 insertions(+), 33 deletions(-)
diff --git
a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionLowBucketRedundancyDistributedTest.java
b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionLowBucketRedundancyDistributedTest.java
new file mode 100644
index 0000000..e87fbd8
--- /dev/null
+++
b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/PartitionedRegionLowBucketRedundancyDistributedTest.java
@@ -0,0 +1,208 @@
+/*
+ * 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.cache.RegionShortcut.PARTITION;
+import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.Serializable;
+import java.util.concurrent.TimeUnit;
+
+import org.awaitility.Awaitility;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import org.apache.geode.cache.PartitionAttributesFactory;
+import org.apache.geode.cache.Region;
+import org.apache.geode.cache.RegionShortcut;
+import org.apache.geode.test.dunit.AsyncInvocation;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.rules.serializable.SerializableTestName;
+
+public class PartitionedRegionLowBucketRedundancyDistributedTest implements
Serializable {
+
+ public String regionName;
+
+ @Rule
+ public ClusterStartupRule startupRule = new ClusterStartupRule();
+
+ @Rule
+ public SerializableTestName testName = new SerializableTestName();
+
+ @Before
+ public void setUp() {
+ regionName = testName.getMethodName() + "_region";
+ }
+
+ @Test
+ public void testTwoServersWithOneRedundantCopy() throws Exception {
+ // Start locator
+ MemberVM locator = startupRule.startLocatorVM(0);
+ int locatorPort = locator.getPort();
+
+ // Start server1 and create region
+ MemberVM server1 = startServerAndCreateRegion(1, locatorPort, PARTITION,
1);
+
+ // Verify lowBucketRedundancyCount == 0 in server1
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+
+ // Do puts in server1
+ server1.getVM().invoke(() -> doPuts(500));
+
+ // Verify lowBucketRedundancyCount == 113 in server1
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(113));
+
+ // Start server2 and create region
+ MemberVM server2 = startServerAndCreateRegion(2, locatorPort, PARTITION,
1);
+
+ // Verify lowBucketRedundancyCount == 0 in both servers
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server2.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+
+ // Stop server2
+ server2.stop(false);
+
+ // Verify lowBucketRedundancyCount == 113 in server1
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(113));
+ }
+
+ @Test
+ public void testThreeServersWithTwoRedundantCopies() throws Exception {
+ // Start locator
+ MemberVM locator = startupRule.startLocatorVM(0);
+ int locatorPort = locator.getPort();
+
+ // Start server1 and create region
+ MemberVM server1 = startServerAndCreateRegion(1, locatorPort, PARTITION,
2);
+
+ // Verify lowBucketRedundancyCount == 0 in server1
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+
+ // Do puts in server1
+ server1.getVM().invoke(() -> doPuts(500));
+
+ // Verify lowBucketRedundancyCount == 113 in server1
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(113));
+
+ // Start server2 and create region
+ MemberVM server2 = startServerAndCreateRegion(2, locatorPort, PARTITION,
2);
+
+ // Verify lowBucketRedundancyCount == 113 in both servers
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(113));
+ server2.getVM().invoke(() -> waitForLowBucketRedundancyCount(113));
+
+ // Start server2 and create region
+ MemberVM server3 = startServerAndCreateRegion(3, locatorPort, PARTITION,
2);
+
+ // Verify lowBucketRedundancyCount == 113 in server1
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server2.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server3.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ }
+
+ @Test
+ public void testFourServersWithPersistentRegionAndOneRedundantCopy() throws
Exception {
+ // Start locator
+ MemberVM locator = startupRule.startLocatorVM(0);
+ int locatorPort = locator.getPort();
+
+ // Start servers and create regions
+ MemberVM server1 = startServerAndCreateRegion(1, locatorPort,
PARTITION_PERSISTENT, 1);
+ MemberVM server2 = startServerAndCreateRegion(2, locatorPort,
PARTITION_PERSISTENT, 1);
+ MemberVM server3 = startServerAndCreateRegion(3, locatorPort,
PARTITION_PERSISTENT, 1);
+ MemberVM server4 = startServerAndCreateRegion(4, locatorPort,
PARTITION_PERSISTENT, 1);
+
+ // Verify lowBucketRedundancyCount == 0 in all servers
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server2.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server3.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server4.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+
+ // Do puts in server1
+ server1.getVM().invoke(() -> doPuts(500));
+
+ // Verify lowBucketRedundancyCount == 0 in all servers
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server2.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server3.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server4.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+
+ // Stop servers 1 and 2
+ server1.stop(false);
+ server2.stop(false);
+
+ server3.getVM().invoke(() -> waitForMembers(1));
+ server4.getVM().invoke(() -> waitForMembers(1));
+
+ // Restart servers 1 and 2
+ server1 = startupRule.startServerVM(1, locatorPort);
+ server2 = startupRule.startServerVM(2, locatorPort);
+
+ // Asynchronously recreate the regions in servers 1 and 2 (since they are
recovering persistent
+ // data)
+ AsyncInvocation recreateRegionInServer1 =
+ server1.getVM().invokeAsync(() -> createRegion(PARTITION_PERSISTENT,
1));
+ AsyncInvocation recreateRegionInServer2 =
+ server2.getVM().invokeAsync(() -> createRegion(PARTITION_PERSISTENT,
1));
+ recreateRegionInServer1.await();
+ recreateRegionInServer2.await();
+
+ // Verify lowBucketRedundancyCount == 0 in all servers
+ server1.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server2.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server3.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ server4.getVM().invoke(() -> waitForLowBucketRedundancyCount(0));
+ }
+
+ private MemberVM startServerAndCreateRegion(int vmId, int locatorPort,
RegionShortcut shortcut,
+ int redundantCopies) {
+ // Start server
+ MemberVM server = startupRule.startServerVM(vmId, locatorPort);
+
+ // Create region
+ server.getVM().invoke(() -> createRegion(shortcut, redundantCopies));
+
+ return server;
+ }
+
+ private void createRegion(RegionShortcut shortcut, int redundantCopies) {
+ PartitionAttributesFactory<?, ?> paf = new PartitionAttributesFactory();
+ paf.setRedundantCopies(redundantCopies);
+ ClusterStartupRule.getCache().createRegionFactory(shortcut)
+ .setPartitionAttributes(paf.create()).create(regionName);
+ }
+
+ private void doPuts(int numPuts) {
+ Region region = ClusterStartupRule.getCache().getRegion(regionName);
+ for (int i = 0; i < numPuts; i++) {
+ region.put("key" + i, "value" + i);
+ }
+ }
+
+ private void waitForLowBucketRedundancyCount(int count) {
+ PartitionedRegion region =
+ (PartitionedRegion)
ClusterStartupRule.getCache().getRegion(regionName);
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).until(
+ () ->
assertThat(region.getPrStats().getLowRedundancyBucketCount()).isEqualTo(count));
+ }
+
+ private void waitForMembers(int count) {
+ Awaitility.await().atMost(30, TimeUnit.SECONDS).until(
+ () ->
assertThat(ClusterStartupRule.getCache().getMembers().size()).isEqualTo(count));
+ }
+}
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRedundancyTracker.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRedundancyTracker.java
index f46ef0c..f7340a0 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRedundancyTracker.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/BucketRedundancyTracker.java
@@ -19,10 +19,11 @@ package org.apache.geode.internal.cache;
* {@link PartitionedRegionRedundancyTracker} of the bucket's status for the
region.
*/
class BucketRedundancyTracker {
- private boolean redundancySatisfied = false;
- private boolean hasAnyCopies = false;
+ // if true decrement allowed; if false increment allowed
+ private boolean noCopiesDecrementOkay = false;
+ // if true decrement allowed; if false increment allowed
+ private boolean lowRedundancyDecrementOkay = false;
private boolean redundancyEverSatisfied = false;
- private boolean hasEverHadCopies = false;
private volatile int currentRedundancy = -1;
private final int targetRedundancy;
private final PartitionedRegionRedundancyTracker regionRedundancyTracker;
@@ -45,14 +46,8 @@ class BucketRedundancyTracker {
* Adjust statistics based on closing a bucket
*/
synchronized void closeBucket() {
- if (!redundancySatisfied) {
- regionRedundancyTracker.decrementLowRedundancyBucketCount();
- redundancySatisfied = true;
- }
- if (hasEverHadCopies && !hasAnyCopies) {
- regionRedundancyTracker.decrementNoCopiesBucketCount();
- hasAnyCopies = true;
- }
+ decrementLowRedundancy();
+ decrementNoCopies();
}
/**
@@ -76,37 +71,53 @@ class BucketRedundancyTracker {
}
private void updateNoCopiesStatistics(int currentBucketHosts) {
- if (hasAnyCopies && currentBucketHosts == 0) {
- hasAnyCopies = false;
+ if (currentBucketHosts == 0) {
+ incrementNoCopies();
+ } else if (currentBucketHosts > 0) {
+ decrementNoCopies();
+ }
+ }
+
+ private void decrementNoCopies() {
+ if (noCopiesDecrementOkay) {
+ noCopiesDecrementOkay = false;
+ regionRedundancyTracker.decrementNoCopiesBucketCount();
+ }
+ }
+
+ private void incrementNoCopies() {
+ if (!noCopiesDecrementOkay) {
+ noCopiesDecrementOkay = true;
regionRedundancyTracker.incrementNoCopiesBucketCount();
- } else if (!hasAnyCopies && currentBucketHosts > 0) {
- if (hasEverHadCopies) {
- regionRedundancyTracker.decrementNoCopiesBucketCount();
- }
- hasEverHadCopies = true;
- hasAnyCopies = true;
}
}
private void updateRedundancyStatistics(int updatedBucketHosts) {
int updatedRedundancy = updatedBucketHosts - 1;
updateCurrentRedundancy(updatedRedundancy);
-
if (updatedRedundancy < targetRedundancy) {
reportUpdatedBucketCount(updatedBucketHosts);
- if (redundancySatisfied) {
- regionRedundancyTracker.incrementLowRedundancyBucketCount();
- redundancySatisfied = false;
- } else if (!hasAnyCopies && updatedRedundancy >= 0) {
- regionRedundancyTracker.incrementLowRedundancyBucketCount();
- }
- } else if (!redundancySatisfied && updatedRedundancy == targetRedundancy) {
- regionRedundancyTracker.decrementLowRedundancyBucketCount();
- redundancySatisfied = true;
+ incrementLowRedundancy();
+ } else if (updatedRedundancy == targetRedundancy) {
+ decrementLowRedundancy();
redundancyEverSatisfied = true;
}
}
+ private void decrementLowRedundancy() {
+ if (lowRedundancyDecrementOkay) {
+ lowRedundancyDecrementOkay = false;
+ regionRedundancyTracker.decrementLowRedundancyBucketCount();
+ }
+ }
+
+ private void incrementLowRedundancy() {
+ if (!lowRedundancyDecrementOkay) {
+ lowRedundancyDecrementOkay = true;
+ regionRedundancyTracker.incrementLowRedundancyBucketCount();
+ }
+ }
+
private void updateCurrentRedundancy(int updatedRedundancy) {
if (updatedRedundancy != currentRedundancy) {
regionRedundancyTracker.setActualRedundancy(updatedRedundancy);
diff --git
a/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRedundancyTrackerTest.java
b/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRedundancyTrackerTest.java
index f6c2102..d48d9c1 100644
---
a/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRedundancyTrackerTest.java
+++
b/geode-core/src/test/java/org/apache/geode/internal/cache/BucketRedundancyTrackerTest.java
@@ -24,7 +24,6 @@ import static org.mockito.Mockito.verify;
import org.junit.Before;
import org.junit.Test;
-
public class BucketRedundancyTrackerTest {
private static final int TARGET_COPIES = 2;
@@ -57,7 +56,7 @@ public class BucketRedundancyTrackerTest {
bucketRedundancyTracker.updateStatistics(TARGET_COPIES);
bucketRedundancyTracker.updateStatistics(TARGET_COPIES - 1);
bucketRedundancyTracker.updateStatistics(TARGET_COPIES);
- verify(regionRedundancyTracker,
times(2)).decrementLowRedundancyBucketCount();
+ verify(regionRedundancyTracker,
times(1)).decrementLowRedundancyBucketCount();
assertEquals(TARGET_COPIES - 1,
bucketRedundancyTracker.getCurrentRedundancy());
}
@@ -66,7 +65,7 @@ public class BucketRedundancyTrackerTest {
bucketRedundancyTracker.updateStatistics(TARGET_COPIES);
bucketRedundancyTracker.updateStatistics(TARGET_COPIES - 1);
bucketRedundancyTracker.closeBucket();
- verify(regionRedundancyTracker,
times(2)).decrementLowRedundancyBucketCount();
+ verify(regionRedundancyTracker,
times(1)).decrementLowRedundancyBucketCount();
assertEquals(0, bucketRedundancyTracker.getCurrentRedundancy());
}
@@ -76,11 +75,26 @@ public class BucketRedundancyTrackerTest {
bucketRedundancyTracker.updateStatistics(TARGET_COPIES - 1);
bucketRedundancyTracker.updateStatistics(0);
bucketRedundancyTracker.closeBucket();
- verify(regionRedundancyTracker,
times(2)).decrementLowRedundancyBucketCount();
+ verify(regionRedundancyTracker,
times(1)).decrementLowRedundancyBucketCount();
assertEquals(-1, bucketRedundancyTracker.getCurrentRedundancy());
}
@Test
+ public void decrementsBucketCountOnIncrementBeforeNoCopies() {
+ bucketRedundancyTracker =
+ new BucketRedundancyTracker(2, regionRedundancyTracker);
+ bucketRedundancyTracker.updateStatistics(3);
+ bucketRedundancyTracker.updateStatistics(2);
+ // Verify incrementLowRedundancyBucketCount is invoked.
+ verify(regionRedundancyTracker,
times(1)).incrementLowRedundancyBucketCount();
+ bucketRedundancyTracker.updateStatistics(1);
+ bucketRedundancyTracker.updateStatistics(2);
+ // Verify incrementLowRedundancyBucketCount is not invoked again when the
count goes 2.
+ verify(regionRedundancyTracker,
times(1)).incrementLowRedundancyBucketCount();
+ assertEquals(1, bucketRedundancyTracker.getCurrentRedundancy());
+ }
+
+ @Test
public void bucketCountNotDecrementedOnClosingBucketThatNeverHadCopies() {
verify(regionRedundancyTracker,
never()).decrementLowRedundancyBucketCount();
assertEquals(-1, bucketRedundancyTracker.getCurrentRedundancy());
@@ -104,8 +118,14 @@ public class BucketRedundancyTrackerTest {
@Test
public void decrementsBucketCountOnHavingAtLeastOneCopyOfBucket() {
bucketRedundancyTracker.updateStatistics(1);
+ // Verify incrementLowRedundancyBucketCount is invoked.
+ verify(regionRedundancyTracker,
times(1)).incrementLowRedundancyBucketCount();
bucketRedundancyTracker.updateStatistics(0);
bucketRedundancyTracker.updateStatistics(1);
+ // Verify incrementLowRedundancyBucketCount is not invoked again when the
count goes to 1.
+ verify(regionRedundancyTracker,
times(1)).incrementLowRedundancyBucketCount();
+ // Verify decrementLowRedundancyBucketCount is not invoked.
+ verify(regionRedundancyTracker,
never()).decrementLowRedundancyBucketCount();
verify(regionRedundancyTracker, times(1)).decrementNoCopiesBucketCount();
assertEquals(0, bucketRedundancyTracker.getCurrentRedundancy());
}
