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 262dc08 GEODE-10094: Fix NPE when populating index in partitioned region. (#7410) 262dc08 is described below commit 262dc08901cfe36c95c833f0b4deb317f1546e97 Author: Eric Shu <e...@pivotal.io> AuthorDate: Wed Mar 2 13:27:19 2022 -0800 GEODE-10094: Fix NPE when populating index in partitioned region. (#7410) * IndexUtils.getIndexManager can return null if bucket region is destroyed due to cache is closing. Make sure CancelException is thrown in this case. --- .../geode/internal/cache/PartitionedRegion.java | 9 ++- .../internal/cache/PartitionedRegionTest.java | 86 ++++++++++++++++++++++ 2 files changed, 93 insertions(+), 2 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 540ed07..2dc2f87 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 @@ -8581,7 +8581,8 @@ public class PartitionedRegion extends LocalRegion } } - private boolean populateEmptyIndexes(Set<Index> indexes, + @VisibleForTesting + boolean populateEmptyIndexes(Set<Index> indexes, HashMap<String, Exception> exceptionsMap) { boolean throwException = false; if (getDataStore() != null && indexes.size() > 0) { @@ -8594,6 +8595,9 @@ public class PartitionedRegion extends LocalRegion continue; } IndexManager bucketIndexManager = IndexUtils.getIndexManager(cache, bucket, true); + if (bucketIndexManager == null) { + cache.getCancelCriterion().checkCancelInProgress(); + } Set<Index> bucketIndexes = getBucketIndexesForPRIndexes(bucket, indexes); try { bucketIndexManager.populateIndexes(bucketIndexes); @@ -8606,7 +8610,8 @@ public class PartitionedRegion extends LocalRegion return throwException; } - private Set<Index> getBucketIndexesForPRIndexes(Region bucket, Set<Index> indexes) { + @VisibleForTesting + Set<Index> getBucketIndexesForPRIndexes(Region bucket, Set<Index> indexes) { Set<Index> bucketIndexes = new HashSet<>(); for (Index ind : indexes) { bucketIndexes.addAll(((PartitionedIndex) ind).getBucketIndexes(bucket)); 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 110c5e0..67fdf9f 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 @@ -42,9 +42,12 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; import junitparams.Parameters; import junitparams.naming.TestCaseName; +import org.jetbrains.annotations.NotNull; import org.junit.Before; import org.junit.Rule; import org.junit.Test; @@ -55,6 +58,7 @@ import org.mockito.junit.MockitoRule; import org.apache.geode.CancelCriterion; import org.apache.geode.Statistics; import org.apache.geode.cache.AttributesFactory; +import org.apache.geode.cache.CacheClosedException; import org.apache.geode.cache.CacheLoader; import org.apache.geode.cache.CacheWriter; import org.apache.geode.cache.Operation; @@ -65,6 +69,9 @@ import org.apache.geode.cache.RegionShortcut; import org.apache.geode.cache.TransactionDataRebalancedException; import org.apache.geode.cache.TransactionException; import org.apache.geode.cache.asyncqueue.AsyncEventQueue; +import org.apache.geode.cache.query.Index; +import org.apache.geode.cache.query.MultiIndexCreationException; +import org.apache.geode.cache.query.internal.index.IndexManager; import org.apache.geode.cache.wan.GatewaySender; import org.apache.geode.distributed.DistributedLockService; import org.apache.geode.distributed.DistributedSystemDisconnectedException; @@ -637,6 +644,85 @@ public class PartitionedRegionTest { assertThat(partitionedRegion.isRegionCreateNotified()).isTrue(); } + @Test + public void populateEmptyIndexesThrowsIfBucketRegionDestroyedDueToCacheClose() { + PartitionedRegion spyPartitionedRegion = spy(partitionedRegion); + BucketRegion bucketRegion = mock(BucketRegion.class); + when(bucketRegion.isDestroyed()).thenReturn(true); + + Set<Index> indexes = setupIndexes(); + ConcurrentMap<Integer, BucketRegion> map = setupBuckets(bucketRegion); + setupDataStore(spyPartitionedRegion, map); + CacheClosedException cacheClosedException = new CacheClosedException(); + setupCancelCriterion(cacheClosedException); + + assertThatThrownBy(() -> spyPartitionedRegion.populateEmptyIndexes(indexes, new HashMap<>())) + .isEqualTo(cacheClosedException); + } + + @NotNull + private Set<Index> setupIndexes() { + Set<Index> indexes = new HashSet<>(); + Index index = mock(Index.class); + indexes.add(index); + return indexes; + } + + @NotNull + private ConcurrentMap<Integer, BucketRegion> setupBuckets(BucketRegion bucketRegion) { + ConcurrentMap<Integer, BucketRegion> map = new ConcurrentHashMap<>(); + map.put(1, bucketRegion); + return map; + } + + private void setupDataStore(PartitionedRegion spyPartitionedRegion, + ConcurrentMap<Integer, BucketRegion> map) { + PartitionedRegionDataStore dataStore = mock(PartitionedRegionDataStore.class); + doReturn(dataStore).when(spyPartitionedRegion).getDataStore(); + when(dataStore.getAllLocalBuckets()).thenReturn(map.entrySet()); + } + + private void setupCancelCriterion(CacheClosedException cacheClosedException) { + CancelCriterion cancelCriterion = mock(CancelCriterion.class); + when(cache.getCancelCriterion()).thenReturn(cancelCriterion); + doThrow(cacheClosedException).when(cancelCriterion).checkCancelInProgress(); + } + + @Test + public void populateEmptyIndexesReturnsFalseIfIndexManagerPopulateIndexesSuccessfully() { + PartitionedRegion spyPartitionedRegion = spy(partitionedRegion); + IndexManager indexManager = mock(IndexManager.class); + BucketRegion bucketRegion = mock(BucketRegion.class); + when(bucketRegion.getIndexManager()).thenReturn(indexManager); + + Set<Index> indexes = setupIndexes(); + ConcurrentMap<Integer, BucketRegion> map = setupBuckets(bucketRegion); + setupDataStore(spyPartitionedRegion, map); + doReturn(indexes).when(spyPartitionedRegion).getBucketIndexesForPRIndexes(bucketRegion, + indexes); + + assertThat(spyPartitionedRegion.populateEmptyIndexes(indexes, new HashMap<>())).isFalse(); + } + + @Test + public void populateEmptyIndexesReturnsTrueIfIndexManagerPopulateIndexesThrows() + throws Exception { + PartitionedRegion spyPartitionedRegion = spy(partitionedRegion); + IndexManager indexManager = mock(IndexManager.class); + BucketRegion bucketRegion = mock(BucketRegion.class); + when(bucketRegion.getIndexManager()).thenReturn(indexManager); + MultiIndexCreationException exception = mock(MultiIndexCreationException.class); + + Set<Index> indexes = setupIndexes(); + ConcurrentMap<Integer, BucketRegion> map = setupBuckets(bucketRegion); + setupDataStore(spyPartitionedRegion, map); + doReturn(indexes).when(spyPartitionedRegion).getBucketIndexesForPRIndexes(bucketRegion, + indexes); + doThrow(exception).when(indexManager).populateIndexes(indexes); + + assertThat(spyPartitionedRegion.populateEmptyIndexes(indexes, new HashMap<>())).isTrue(); + } + private static <K> Set<K> asSet(K... values) { Set<K> set = new HashSet<>(); Collections.addAll(set, values);