You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by es...@apache.org on 2017/12/20 19:35:51 UTC
[geode] branch develop updated: GEODE-551: make sure accessor gets
critical by putting into its local region (#1183)
This is an automated email from the ASF dual-hosted git repository.
eshu11 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 ed6bac5 GEODE-551: make sure accessor gets critical by putting into its local region (#1183)
ed6bac5 is described below
commit ed6bac5bc5cd73094e2b9073ff47e31e4e3503bd
Author: pivotal-eshu <es...@pivotal.io>
AuthorDate: Wed Dec 20 11:35:47 2017 -0800
GEODE-551: make sure accessor gets critical by putting into its local region (#1183)
* GEODE-551: Make sure accessor gets critical by putting into local region instead of putting into parititioned region.
---
.../MemoryThresholdsOffHeapDUnitTest.java | 118 +++++++++++----------
1 file changed, 60 insertions(+), 58 deletions(-)
diff --git a/geode-core/src/test/java/org/apache/geode/cache/management/MemoryThresholdsOffHeapDUnitTest.java b/geode-core/src/test/java/org/apache/geode/cache/management/MemoryThresholdsOffHeapDUnitTest.java
index a3ad9e1..2fd803c 100644
--- a/geode-core/src/test/java/org/apache/geode/cache/management/MemoryThresholdsOffHeapDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/cache/management/MemoryThresholdsOffHeapDUnitTest.java
@@ -950,7 +950,6 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
* Test that a Partitioned Region loader invocation is rejected if the VM with the bucket is in a
* critical state.
*/
- @Category(FlakyTest.class) // GEODE-551: waitForCriterion, memory sensitive
@Test
public void testPRLoadRejection() throws Exception {
final Host host = Host.getHost(0);
@@ -960,24 +959,18 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
// Make sure the desired VMs will have a fresh DS. TODO: convert these from AsyncInvocation to
// invoke
- AsyncInvocation d0 = accessor.invokeAsync(() -> disconnectFromDS());
- AsyncInvocation d1 = ds1.invokeAsync(() -> disconnectFromDS());
- d0.join();
- assertFalse(d0.exceptionOccurred());
- d1.join();
- assertFalse(d1.exceptionOccurred());
- CacheSerializableRunnable establishConnectivity =
- new CacheSerializableRunnable("establishcConnectivity") {
- @Override
- public void run2() throws CacheException {
- getSystem();
- }
- };
- ds1.invoke(establishConnectivity);
- accessor.invoke(establishConnectivity);
+ accessor.invoke(() -> disconnectFromDS());
+ ds1.invoke(() -> disconnectFromDS());
- ds1.invoke(createPR(rName, false));
- accessor.invoke(createPR(rName, true));
+ ds1.invoke("establishcConnectivity", () -> {
+ getSystem();
+ createPR(rName, false);
+ });
+
+ accessor.invoke("establishcConnectivity", () -> {
+ getSystem();
+ createPR(rName, true);
+ });
final AtomicInteger expectedInvocations = new AtomicInteger(0);
@@ -1081,6 +1074,8 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
new SerializableCallable("Set safe state on datastore, assert local load behavior") {
public Object call() throws Exception {
final PartitionedRegion r = (PartitionedRegion) getCache().getRegion(rName);
+ final OffHeapMemoryMonitor ohmm =
+ ((InternalResourceManager) getCache().getResourceManager()).getOffHeapMonitor();
r.destroy("oh3");
WaitCriterion wc = new WaitCriterion() {
@@ -1089,7 +1084,7 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
}
public boolean done() {
- return !r.memoryThresholdReached.get();
+ return !ohmm.getState().isCritical();
}
};
Wait.waitForCriterion(wc, 30 * 1000, 10, true);
@@ -1116,8 +1111,13 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
assertTrue(r.containsKey(k));
assertEquals(k.toString(), r.get(k, expectedInvocations9)); // no load
- // Go critical in accessor
- r.put("oh3", new byte[157287]);
+ // Go critical in accessor by creating entries in local node
+ String localRegionName = "localRegionName";
+ AttributesFactory<Integer, String> af = getLocalRegionAttributesFactory();
+ final LocalRegion localRegion =
+ (LocalRegion) getCache().createRegion(localRegionName, af.create());
+ localRegion.put("oh1", new byte[838860]);
+ localRegion.put("oh3", new byte[157287]);
WaitCriterion wc = new WaitCriterion() {
public String description() {
@@ -1125,9 +1125,10 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
}
public boolean done() {
- return r.memoryThresholdReached.get();
+ return ohmm.getState().isCritical();
}
};
+ Wait.waitForCriterion(wc, 30 * 1000, 10, true);
k = new Integer(5);
Integer expectedInvocations10 = new Integer(expectedInvocations.incrementAndGet());
@@ -1136,7 +1137,7 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
assertEquals(k.toString(), r.get(k, expectedInvocations10)); // no load
// Clean up critical state
- r.destroy("oh3");
+ localRegion.destroy("oh3");
wc = new WaitCriterion() {
public String description() {
return "verify critical state";
@@ -1146,6 +1147,8 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
return !ohmm.getState().isCritical();
}
};
+ Wait.waitForCriterion(wc, 30 * 1000, 10, true);
+
return expectedInvocations10;
}
});
@@ -1154,40 +1157,34 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
ds1.invoke(removeExpectedException);
}
- private CacheSerializableRunnable createPR(final String rName, final boolean accessor) {
- return new CacheSerializableRunnable("create PR accessor") {
- @Override
- public void run2() throws CacheException {
- // Assert some level of connectivity
- getSystem(getOffHeapProperties());
- InternalResourceManager irm = (InternalResourceManager) getCache().getResourceManager();
- irm.setCriticalOffHeapPercentage(90f);
- AttributesFactory<Integer, String> af = new AttributesFactory<Integer, String>();
- if (!accessor) {
- af.setCacheLoader(new CacheLoader<Integer, String>() {
- final AtomicInteger numLoaderInvocations = new AtomicInteger(0);
-
- public String load(LoaderHelper<Integer, String> helper) throws CacheLoaderException {
- Integer expectedInvocations = (Integer) helper.getArgument();
- final int actualInvocations = this.numLoaderInvocations.getAndIncrement();
- if (expectedInvocations.intValue() != actualInvocations) {
- throw new CacheLoaderException("Expected " + expectedInvocations
- + " invocations, actual is " + actualInvocations);
- }
- return helper.getKey().toString();
- }
-
- public void close() {}
- });
+ private void createPR(final String rName, final boolean accessor) {
+ getSystem(getOffHeapProperties());
+ InternalResourceManager irm = (InternalResourceManager) getCache().getResourceManager();
+ irm.setCriticalOffHeapPercentage(90f);
+ AttributesFactory<Integer, String> af = new AttributesFactory<Integer, String>();
+ if (!accessor) {
+ af.setCacheLoader(new CacheLoader<Integer, String>() {
+ final AtomicInteger numLoaderInvocations = new AtomicInteger(0);
- af.setPartitionAttributes(new PartitionAttributesFactory().create());
- } else {
- af.setPartitionAttributes(new PartitionAttributesFactory().setLocalMaxMemory(0).create());
+ public String load(LoaderHelper<Integer, String> helper) throws CacheLoaderException {
+ Integer expectedInvocations = (Integer) helper.getArgument();
+ final int actualInvocations = this.numLoaderInvocations.getAndIncrement();
+ if (expectedInvocations.intValue() != actualInvocations) {
+ throw new CacheLoaderException("Expected " + expectedInvocations
+ + " invocations, actual is " + actualInvocations + " for key " + helper.getKey());
+ }
+ return helper.getKey().toString();
}
- af.setOffHeap(true);
- getCache().createRegion(rName, af.create());
- }
- };
+
+ public void close() {}
+ });
+
+ af.setPartitionAttributes(new PartitionAttributesFactory().create());
+ } else {
+ af.setPartitionAttributes(new PartitionAttributesFactory().setLocalMaxMemory(0).create());
+ }
+ af.setOffHeap(true);
+ getCache().createRegion(rName, af.create());
}
/**
@@ -1209,9 +1206,7 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
InternalResourceManager irm = (InternalResourceManager) getCache().getResourceManager();
final OffHeapMemoryMonitor ohmm = irm.getOffHeapMonitor();
irm.setCriticalOffHeapPercentage(90f);
- AttributesFactory<Integer, String> af = new AttributesFactory<Integer, String>();
- af.setScope(Scope.LOCAL);
- af.setOffHeap(true);
+ AttributesFactory<Integer, String> af = getLocalRegionAttributesFactory();
final AtomicInteger numLoaderInvocations = new AtomicInteger(0);
af.setCacheLoader(new CacheLoader<Integer, String>() {
public String load(LoaderHelper<Integer, String> helper) throws CacheLoaderException {
@@ -1297,6 +1292,13 @@ public class MemoryThresholdsOffHeapDUnitTest extends ClientServerTestCase {
});
}
+ private AttributesFactory<Integer, String> getLocalRegionAttributesFactory() {
+ AttributesFactory<Integer, String> af = new AttributesFactory<Integer, String>();
+ af.setScope(Scope.LOCAL);
+ af.setOffHeap(true);
+ return af;
+ }
+
/**
* Create a list of integers consisting of the ranges defined by the provided argument e.g..
* createRanges(1, 4, 10, 12) means create ranges 1 through 4 and 10 through 12 and should yield
--
To stop receiving notification emails like this one, please contact
['"commits@geode.apache.org" <co...@geode.apache.org>'].