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 2018/08/13 22:17:04 UTC
[geode] branch develop updated: GEODE-5541: A create followed by a
destroy of the same key will not invoke cache listener on remote nodes.
(#2282)
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 6f8de00 GEODE-5541: A create followed by a destroy of the same key will not invoke cache listener on remote nodes. (#2282)
6f8de00 is described below
commit 6f8de005b0026b3e365cb577d07119e0d53c6ed4
Author: pivotal-eshu <es...@pivotal.io>
AuthorDate: Mon Aug 13 15:16:59 2018 -0700
GEODE-5541: A create followed by a destroy of the same key will not invoke cache listener on remote nodes. (#2282)
* Both far side and near side will check if need to add destroy event in pending callbacks in LocalRegion to avoid invoke a no op destroy event.
---
...tPartitionedRegionWithTransactionDUnitTest.java | 61 +++++++++++
.../apache/geode/internal/cache/LocalRegion.java | 23 +++-
.../apache/geode/internal/cache/TXEntryState.java | 13 +--
.../geode/internal/cache/LocalRegionTest.java | 118 +++++++++++++++++++++
.../geode/internal/cache/RegionCommitTest.java | 4 +-
5 files changed, 206 insertions(+), 13 deletions(-)
diff --git a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentPartitionedRegionWithTransactionDUnitTest.java b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentPartitionedRegionWithTransactionDUnitTest.java
index 9460e63..b30474f 100644
--- a/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentPartitionedRegionWithTransactionDUnitTest.java
+++ b/geode-core/src/distributedTest/java/org/apache/geode/internal/cache/partitioned/PersistentPartitionedRegionWithTransactionDUnitTest.java
@@ -15,13 +15,19 @@
package org.apache.geode.internal.cache.partitioned;
import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.apache.geode.cache.Cache;
+import org.apache.geode.cache.CacheListener;
import org.apache.geode.cache.CacheTransactionManager;
import org.apache.geode.cache.Region;
+import org.apache.geode.cache.util.CacheListenerAdapter;
import org.apache.geode.internal.cache.DiskRegion;
import org.apache.geode.internal.cache.DiskStoreImpl;
import org.apache.geode.internal.cache.PartitionedRegion;
@@ -207,4 +213,59 @@ public class PersistentPartitionedRegionWithTransactionDUnitTest
}
});
}
+
+ @Test
+ public void NoDestroyInvocationIfCreateEntryAndDestroyItInTransaction() throws Throwable {
+ Host host = Host.getHost(0);
+ VM vm0 = host.getVM(0);
+ VM vm1 = host.getVM(1);
+ VM vm2 = host.getVM(2);
+
+ int redundancy = 2;
+
+ vm0.invoke(() -> getCacheSetAlwaysFireLocalListeners());
+ vm1.invoke(() -> getCacheSetAlwaysFireLocalListeners());
+ vm2.invoke(() -> getCacheSetAlwaysFireLocalListeners());
+
+ createPR(vm0, redundancy);
+ createPR(vm1, redundancy);
+ createPR(vm2, redundancy);
+
+ vm0.invoke(() -> addListener());
+ vm1.invoke(() -> addListener());
+ vm2.invoke(() -> addListener());
+
+ vm0.invoke(() -> {
+ Cache cache = getCache();
+ TXManagerImpl txManager = (TXManagerImpl) cache.getCacheTransactionManager();
+ Region region = cache.getRegion(getPartitionedRegionName());
+ txManager.begin();
+ region.create(1, "toBeDestroyed");
+ region.destroy(1);
+ txManager.commit();
+ });
+
+ vm0.invoke(() -> verifyNoDestroyInvocation());
+ vm1.invoke(() -> verifyNoDestroyInvocation());
+ vm2.invoke(() -> verifyNoDestroyInvocation());
+ }
+
+ private void getCacheSetAlwaysFireLocalListeners() {
+ System.setProperty("gemfire.BucketRegion.alwaysFireLocalListeners", "true");
+ getCache();
+ }
+
+ private void addListener() {
+ Cache cache = getCache();
+ Region region = cache.getRegion(getPartitionedRegionName());
+ CacheListener listener = spy(new CacheListenerAdapter() {});
+ region.getAttributesMutator().addCacheListener(listener);
+ }
+
+ private void verifyNoDestroyInvocation() throws Exception {
+ Cache cache = getCache();
+ Region region = cache.getRegion(getPartitionedRegionName());
+ CacheListener listener = region.getAttributes().getCacheListeners()[0];
+ verify(listener, never()).afterDestroy(any());
+ }
}
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
index 5e2a491..a3609c3 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
@@ -6896,7 +6896,10 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
try {
final boolean inTokenMode = needTokensForGII || needRIDestroyToken;
- this.entries.txApplyDestroy(key, rmtOrigin, event, inTokenMode, needRIDestroyToken, op,
+ if (!needsPendingCallbacksForDestroy(key, txEntryState)) {
+ pendingCallbacks = new ArrayList<>();
+ }
+ getRegionMap().txApplyDestroy(key, rmtOrigin, event, inTokenMode, needRIDestroyToken, op,
eventId, aCallbackArgument, pendingCallbacks, filterRoutingInfo, bridgeContext,
isOriginRemote, txEntryState, versionTag, tailKey);
} finally {
@@ -6906,6 +6909,24 @@ public class LocalRegion extends AbstractRegion implements LoaderHelperFactory,
}
}
+ boolean needsPendingCallbacksForDestroy(Object key, TXEntryState txEntryState) {
+ if (isProxy()) {
+ return true;
+ }
+ if (txEntryState != null) {
+ // nearside on tx host
+ // a destroy on destroyed or removed token, to local committed state, becomes a noop
+ // since nothing needed to be done locally.
+ // We don't want to actually do the destroy since we told the
+ // transaction listener that no destroy was done.
+ Object originalValue = txEntryState.getOriginalValue();
+ return originalValue != null && !Token.isRemoved(originalValue);
+ }
+ // farside without TXEntryState
+ RegionEntry regionEntry = basicGetEntry(key);
+ return regionEntry != null && !regionEntry.isDestroyedOrRemoved();
+ }
+
/**
* Called by lower levels, while still holding the write sync lock, and the low level has
* completed its part of the basic destroy
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/TXEntryState.java b/geode-core/src/main/java/org/apache/geode/internal/cache/TXEntryState.java
index 72a1b2d..d95877d 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/TXEntryState.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/TXEntryState.java
@@ -19,8 +19,6 @@ import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.TX
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Set;
import org.apache.logging.log4j.Logger;
@@ -1554,21 +1552,14 @@ public class TXEntryState implements Releasable {
}
private void txApplyDestroyLocally(InternalRegion r, Object key, TXState txState) {
- boolean invokeCallbacks = isOpDestroyEvent(r);
- List<EntryEventImpl> pendingCallbacks =
- invokeCallbacks ? txState.getPendingCallbacks() : new ArrayList<EntryEventImpl>();
try {
r.txApplyDestroy(key, txState.getTransactionId(), null, false/* inTokenMode */,
- getDestroyOperation(), getNearSideEventId(txState), callBackArgument, pendingCallbacks,
+ getDestroyOperation(), getNearSideEventId(txState), callBackArgument,
+ txState.getPendingCallbacks(),
getFilterRoutingInfo(), txState.bridgeContext, false, this, null, -1);
} catch (RegionDestroyedException ignore) {
} catch (EntryDestroyedException ignore) {
}
- // if !isOpDestroyEvent then
- // this destroy, to local committed state, becomes a noop
- // since nothing needed to be done locally.
- // We don't want to actually do the destroy since we told the
- // transaction listener that no destroy was done.
}
private void txApplyInvalidateLocally(InternalRegion r, Object key, Object newValue,
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/LocalRegionTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/LocalRegionTest.java
new file mode 100644
index 0000000..72cc213
--- /dev/null
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/LocalRegionTest.java
@@ -0,0 +1,118 @@
+/*
+ * 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.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.Test;
+
+
+public class LocalRegionTest {
+
+ private final Object key = new Object();
+
+ @Test
+ public void proxyRegionNeedsPendingCallbacksForDestroy() {
+ TXEntryState txEntryState = mock(TXEntryState.class);
+ LocalRegion region = mock(LocalRegion.class);
+ when(region.isProxy()).thenReturn(true);
+ when(region.needsPendingCallbacksForDestroy(key, txEntryState)).thenCallRealMethod();
+
+ assertThat(region.needsPendingCallbacksForDestroy(key, txEntryState)).isTrue();
+ }
+
+ @Test
+ public void destroyNullOriginalValueOnTransactionHostDoesNotNeedPendingCallbacksForDestroy() {
+ TXEntryState txEntryState = mock(TXEntryState.class);
+ LocalRegion region = mock(LocalRegion.class);
+ when(region.isProxy()).thenReturn(false);
+ when(txEntryState.getOriginalValue()).thenReturn(null);
+ when(region.needsPendingCallbacksForDestroy(key, txEntryState)).thenCallRealMethod();
+
+ assertThat(region.needsPendingCallbacksForDestroy(key, txEntryState)).isFalse();
+ }
+
+ @Test
+ public void destroyDestroyedTokenOnTransactionHostDoesNotNeedPendingCallbacksForDestroy() {
+ TXEntryState txEntryState = mock(TXEntryState.class);
+ LocalRegion region = mock(LocalRegion.class);
+ when(region.isProxy()).thenReturn(false);
+ when(txEntryState.getOriginalValue()).thenReturn(Token.DESTROYED);
+ when(region.needsPendingCallbacksForDestroy(key, txEntryState)).thenCallRealMethod();
+
+ assertThat(region.needsPendingCallbacksForDestroy(key, txEntryState)).isFalse();
+ }
+
+ @Test
+ public void destroyTombstoneOnTransactionHostDoesNotNeedPendingCallbacksForDestroy() {
+ TXEntryState txEntryState = mock(TXEntryState.class);
+ LocalRegion region = mock(LocalRegion.class);
+ when(region.isProxy()).thenReturn(false);
+ when(txEntryState.getOriginalValue()).thenReturn(Token.TOMBSTONE);
+ when(region.needsPendingCallbacksForDestroy(key, txEntryState)).thenCallRealMethod();
+
+ assertThat(region.needsPendingCallbacksForDestroy(key, txEntryState)).isFalse();
+ }
+
+ @Test
+ public void destroyNotATokenOnTransactionHostNeedsPendingCallbacksForDestroy() {
+ TXEntryState txEntryState = mock(TXEntryState.class);
+ LocalRegion region = mock(LocalRegion.class);
+ when(region.isProxy()).thenReturn(false);
+ when(txEntryState.getOriginalValue()).thenReturn(new Token.NotAToken());
+ when(region.needsPendingCallbacksForDestroy(key, txEntryState)).thenCallRealMethod();
+
+ assertThat(region.needsPendingCallbacksForDestroy(key, txEntryState)).isTrue();
+ }
+
+ @Test
+ public void destroyNonExistingRegionEntryOnRemoteHostDoesNotNeedPendingCallbacksForDestroy() {
+ LocalRegion region = mock(LocalRegion.class);
+ when(region.basicGetEntry(key)).thenReturn(null);
+ when(region.isProxy()).thenReturn(false);
+ when(region.needsPendingCallbacksForDestroy(key, null)).thenCallRealMethod();
+
+ assertThat(region.needsPendingCallbacksForDestroy(key, null)).isFalse();
+ }
+
+ @Test
+ public void destroyRemovedTokenOnRemoteHostDoesNotNeedPendingCallbacksForDestroy() {
+ LocalRegion region = mock(LocalRegion.class);
+ RegionEntry regionEntry = mock(RegionEntry.class);
+ when(region.isProxy()).thenReturn(false);
+ when(region.isProxy()).thenReturn(false);
+ when(region.basicGetEntry(key)).thenReturn(regionEntry);
+ when(regionEntry.isDestroyedOrRemoved()).thenReturn(true);
+ when(region.needsPendingCallbacksForDestroy(key, null)).thenCallRealMethod();
+
+ assertThat(region.needsPendingCallbacksForDestroy(key, null)).isFalse();
+ }
+
+ @Test
+ public void destroyAValueOnRemoteHostNeedsPendingCallbacksForDestroy() {
+ LocalRegion region = mock(LocalRegion.class);
+ RegionEntry regionEntry = mock(RegionEntry.class);
+ when(region.isProxy()).thenReturn(false);
+ when(region.isProxy()).thenReturn(false);
+ when(region.basicGetEntry(key)).thenReturn(regionEntry);
+ when(regionEntry.isDestroyedOrRemoved()).thenReturn(false);
+ when(region.needsPendingCallbacksForDestroy(key, null)).thenCallRealMethod();
+
+ assertThat(region.needsPendingCallbacksForDestroy(key, null)).isTrue();
+ }
+}
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/RegionCommitTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/RegionCommitTest.java
index 193faf2..39218ad 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/RegionCommitTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/RegionCommitTest.java
@@ -31,6 +31,8 @@ public class RegionCommitTest {
private String path;
private LocalRegion region;
private TXCommitMessage txCommitMessage;
+ private RegionCommit regionCommit;
+ private final Object key = new Object();
@Before
public void setUp() {
@@ -41,6 +43,7 @@ public class RegionCommitTest {
dm = mock(ClusterDistributionManager.class);
region = mock(LocalRegion.class);
txCommitMessage = mock(TXCommitMessage.class);
+ regionCommit = new RegionCommit(txCommitMessage);
when(dm.getCache()).thenReturn(cache);
when(cache.getRegionByPath(path)).thenReturn(region);
@@ -49,7 +52,6 @@ public class RegionCommitTest {
@Test
public void getsRegionFromCacheFromDM() {
- RegionCommit regionCommit = new RegionCommit(txCommitMessage);
assertThat(regionCommit.getRegionByPath(dm, path)).isEqualTo(region);
}
}