You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@geode.apache.org by up...@apache.org on 2017/05/24 17:21:18 UTC
geode git commit: Starting a test of closing the cache while the AEQ
is enqueuing
Repository: geode
Updated Branches:
refs/heads/feature/close_cache_aeq_enqueue [created] 070494d37
Starting a test of closing the cache while the AEQ is enqueuing
Project: http://git-wip-us.apache.org/repos/asf/geode/repo
Commit: http://git-wip-us.apache.org/repos/asf/geode/commit/070494d3
Tree: http://git-wip-us.apache.org/repos/asf/geode/tree/070494d3
Diff: http://git-wip-us.apache.org/repos/asf/geode/diff/070494d3
Branch: refs/heads/feature/close_cache_aeq_enqueue
Commit: 070494d37d5aff1c3ac8e7fe3a483157dea25927
Parents: 456ee15
Author: Dan Smith <up...@apache.org>
Authored: Wed May 24 10:18:57 2017 -0700
Committer: Dan Smith <up...@apache.org>
Committed: Wed May 24 10:18:57 2017 -0700
----------------------------------------------------------------------
.../asyncqueue/AsyncEventListenerDUnitTest.java | 81 ++++++++++++++++++++
1 file changed, 81 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/geode/blob/070494d3/geode-core/src/test/java/org/apache/geode/internal/cache/wan/asyncqueue/AsyncEventListenerDUnitTest.java
----------------------------------------------------------------------
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/wan/asyncqueue/AsyncEventListenerDUnitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/asyncqueue/AsyncEventListenerDUnitTest.java
index 3dd0550..ac89b48 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/wan/asyncqueue/AsyncEventListenerDUnitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/wan/asyncqueue/AsyncEventListenerDUnitTest.java
@@ -19,6 +19,7 @@ import static org.junit.Assert.*;
import static org.mockito.Matchers.any;
import java.util.Arrays;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -29,6 +30,12 @@ import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
+import org.apache.geode.cache.AttributesFactory;
+import org.apache.geode.cache.CacheClosedException;
+import org.apache.geode.cache.PartitionAttributesFactory;
+import org.apache.geode.cache.wan.GatewayEventFilter;
+import org.apache.geode.cache.wan.GatewayQueueEvent;
+import org.apache.geode.internal.cache.wan.MyAsyncEventListener;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -1674,6 +1681,80 @@ public class AsyncEventListenerDUnitTest extends AsyncEventQueueTestBase {
Awaitility.waitAtMost(10000, TimeUnit.MILLISECONDS).until(() -> getBucketMoved(vm2, "ln"));
}
+ @Test
+ public void testCacheClosedBeforeAEQWrite() {
+ Integer lnPort =
+ (Integer) vm0.invoke(() -> AsyncEventQueueTestBase.createFirstLocatorWithDSId(1));
+
+ vm1.invoke(createCacheRunnable(lnPort));
+ vm2.invoke(createCacheRunnable(lnPort));
+ final DistributedMember member1 =
+ vm1.invoke(() -> cache.getDistributedSystem().getDistributedMember());
+
+ vm1.invoke(() -> {
+ cache.createAsyncEventQueueFactory().addGatewayEventFilter(new GatewayEventFilter() {
+ @Override
+ public boolean beforeEnqueue(final GatewayQueueEvent event) {
+// if (event.getOperation().isDestroy()) {
+ if (event.getOperation().isRemoveAll()) {
+ new Thread(() -> cache.close()).start();
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ //ignore
+ }
+ throw new CacheClosedException();
+ }
+ return true;
+ };
+
+ @Override
+ public boolean beforeTransmit(final GatewayQueueEvent event) {
+ return false;
+ }
+
+ @Override
+ public void afterAcknowledgement(final GatewayQueueEvent event) {
+
+ }
+
+ @Override
+ public void close() {
+
+ }
+ }).create("ln", new MyAsyncEventListener());
+ });
+
+ vm1.invoke(() -> AsyncEventQueueTestBase.createPartitionedRegionWithAsyncEventQueue(
+ getTestMethodName() + "_PR", "ln", isOffHeap()));
+
+ vm2.invoke(() -> {
+ AttributesFactory fact = new AttributesFactory();
+
+ PartitionAttributesFactory pfact = new PartitionAttributesFactory();
+ pfact.setTotalNumBuckets(16);
+ pfact.setLocalMaxMemory(0);
+ fact.setPartitionAttributes(pfact.create());
+ fact.setOffHeap(isOffHeap());
+ Region r = cache.createRegionFactory(fact.create()).addAsyncEventQueueId("ln")
+ .create(getTestMethodName() + "_PR");
+
+ });
+
+ vm1.invoke(() -> AsyncEventQueueTestBase.pauseAsyncEventQueue("ln"));
+// vm2.invoke(() -> AsyncEventQueueTestBase.doPuts(getTestMethodName() + "_PR", 3));
+ vm2.invoke(() -> {
+ Region r = cache.getRegion(Region.SEPARATOR + getTestMethodName() + "_PR");
+ r.put(1, 1);
+ r.put(2, 2);
+ r.removeAll(Collections.singleton(1));
+ r.remove(1);
+ });
+
+ vm1.invoke(() -> Awaitility.await().atMost(1, TimeUnit.MINUTES).until(() -> cache.isClosed()));
+// vm1.invoke(() -> AsyncEventQueueTestBase.resumeAsyncEventQueue("ln"));
+ }
+
private static Set<Object> getKeysSeen(VM vm, String asyncEventQueueId) {
return vm.invoke(() -> {
final BucketMovingAsyncEventListener listener =