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 2020/05/05 19:02:17 UTC
[geode] branch feature/GEODE-8073 updated: GEODE-8073: Fix NPE
after FetchKeysMessage failed.
This is an automated email from the ASF dual-hosted git repository.
eshu11 pushed a commit to branch feature/GEODE-8073
in repository https://gitbox.apache.org/repos/asf/geode.git
The following commit(s) were added to refs/heads/feature/GEODE-8073 by this push:
new d61bae6 GEODE-8073: Fix NPE after FetchKeysMessage failed.
d61bae6 is described below
commit d61bae66de48625dfbff45cf159b9fcb54529828
Author: Eric Shu <es...@pivotal.io>
AuthorDate: Tue May 5 12:00:20 2020 -0700
GEODE-8073: Fix NPE after FetchKeysMessage failed.
---
.../geode/internal/cache/PartitionedRegion.java | 73 ++++++++++++++--------
.../internal/cache/PartitionedRegionTest.java | 49 +++++++++++++++
2 files changed, 95 insertions(+), 27 deletions(-)
diff --git a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
index cacab2d..53754e3 100755
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java
@@ -4593,39 +4593,58 @@ public class PartitionedRegion extends LocalRegion
buckets = bucketKeys.keySet();
}
- for (Integer bucket : buckets) {
- Set keys = null;
- if (bucketKeys == null) {
- try {
- FetchKeysResponse fkr = FetchKeysMessage.send(member, this, bucket, true);
- keys = fkr.waitForKeys();
- } catch (ForceReattemptException ignore) {
- failures.add(bucket);
- }
- } else {
- keys = bucketKeys.get(bucket);
+ fetchKeysAndValues(values, servConn, failures, member, bucketKeys, buckets);
+ }
+ return failures;
+ }
+
+ void fetchKeysAndValues(VersionedObjectList values, ServerConnection servConn,
+ Set<Integer> failures, InternalDistributedMember member,
+ HashMap<Integer, HashSet> bucketKeys, Set<Integer> buckets)
+ throws IOException {
+ for (Integer bucket : buckets) {
+ Set keys = null;
+ if (bucketKeys == null) {
+ try {
+ FetchKeysResponse fetchKeysResponse = getFetchKeysResponse(member, bucket);
+ keys = fetchKeysResponse.waitForKeys();
+ } catch (ForceReattemptException ignore) {
+ failures.add(bucket);
}
+ } else {
+ keys = bucketKeys.get(bucket);
+ }
+ if (keys != null) {
+ getValuesForKeys(values, servConn, keys);
+ }
+ }
+ }
- // TODO (ashetkar) Use single Get70 instance for all?
- for (Object key : keys) {
- Get70 command = (Get70) Get70.getCommand();
- Get70.Entry ge = command.getValueAndIsObject(this, key, null, servConn);
+ FetchKeysResponse getFetchKeysResponse(InternalDistributedMember member,
+ Integer bucket)
+ throws ForceReattemptException {
+ return FetchKeysMessage.send(member, this, bucket, true);
+ }
- if (ge.keyNotPresent) {
- values.addObjectPartForAbsentKey(key, ge.value, ge.versionTag);
- } else {
- values.addObjectPart(key, ge.value, ge.isObject, ge.versionTag);
- }
+ void getValuesForKeys(VersionedObjectList values, ServerConnection servConn, Set keys)
+ throws IOException {
+ // TODO (ashetkar) Use single Get70 instance for all?
+ for (Object key : keys) {
+ Get70 command = (Get70) Get70.getCommand();
+ Get70.Entry ge = command.getValueAndIsObject(this, key, null, servConn);
- if (values.size() == BaseCommand.MAXIMUM_CHUNK_SIZE) {
- BaseCommand.sendNewRegisterInterestResponseChunk(this, "keyList", values, false,
- servConn);
- values.clear();
- }
- }
+ if (ge.keyNotPresent) {
+ values.addObjectPartForAbsentKey(key, ge.value, ge.versionTag);
+ } else {
+ values.addObjectPart(key, ge.value, ge.isObject, ge.versionTag);
+ }
+
+ if (values.size() == BaseCommand.MAXIMUM_CHUNK_SIZE) {
+ BaseCommand.sendNewRegisterInterestResponseChunk(this, "keyList", values, false,
+ servConn);
+ values.clear();
}
}
- return failures;
}
/**
diff --git a/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionTest.java b/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionTest.java
index 210ce16..4881979 100644
--- a/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionTest.java
+++ b/geode-core/src/test/java/org/apache/geode/internal/cache/PartitionedRegionTest.java
@@ -25,9 +25,11 @@ import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -69,7 +71,10 @@ import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.control.InternalResourceManager;
+import org.apache.geode.internal.cache.partitioned.FetchKeysMessage;
import org.apache.geode.internal.cache.partitioned.colocation.ColocationLoggerFactory;
+import org.apache.geode.internal.cache.tier.sockets.ServerConnection;
+import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
@RunWith(JUnitParamsRunner.class)
@SuppressWarnings({"deprecation", "unchecked", "unused"})
@@ -573,6 +578,50 @@ public class PartitionedRegionTest {
.hasMessage(PartitionedRegion.DATA_MOVED_BY_REBALANCE).hasCause(exception);
}
+ @Test
+ public void failuresSavedIfFetchKeysThrows() throws Exception {
+ PartitionedRegion spyPartitionedRegion = spy(partitionedRegion);
+
+ VersionedObjectList values = mock(VersionedObjectList.class);
+ ServerConnection serverConnection = mock(ServerConnection.class);
+ Set<Integer> failures = mock(Set.class);
+ InternalDistributedMember member = mock(InternalDistributedMember.class);
+ Set<Integer> buckets = new HashSet<>();
+ buckets.add(1);
+ doThrow(new ForceReattemptException("")).when(spyPartitionedRegion).getFetchKeysResponse(member,
+ 1);
+
+ spyPartitionedRegion.fetchKeysAndValues(values, serverConnection, failures, member, null,
+ buckets);
+
+ verify(failures).add(1);
+ verify(spyPartitionedRegion, never()).getValuesForKeys(values, serverConnection, null);
+ }
+
+ @Test
+ public void fetchKeysAndValuesInvokesGetValuesForKeys() throws Exception {
+ PartitionedRegion spyPartitionedRegion = spy(partitionedRegion);
+
+ VersionedObjectList values = mock(VersionedObjectList.class);
+ ServerConnection serverConnection = mock(ServerConnection.class);
+ Set<Integer> failures = mock(Set.class);
+ InternalDistributedMember member = mock(InternalDistributedMember.class);
+ Set<Integer> buckets = new HashSet<>();
+ buckets.add(1);
+ FetchKeysMessage.FetchKeysResponse fetchKeysResponse =
+ mock(FetchKeysMessage.FetchKeysResponse.class);
+ doReturn(fetchKeysResponse).when(spyPartitionedRegion).getFetchKeysResponse(member, 1);
+ Set keys = mock(Set.class);
+ when(fetchKeysResponse.waitForKeys()).thenReturn(keys);
+ doNothing().when(spyPartitionedRegion).getValuesForKeys(values, serverConnection, keys);
+
+ spyPartitionedRegion.fetchKeysAndValues(values, serverConnection, failures, member, null,
+ buckets);
+
+ verify(failures, never()).add(1);
+ verify(spyPartitionedRegion).getValuesForKeys(values, serverConnection, keys);
+ }
+
private static <K> Set<K> asSet(K... values) {
Set<K> set = new HashSet<>();
Collections.addAll(set, values);