PartitionedRegion extends LocalRegion implements CacheDistributionAdvisee, QueryExecutor { public Set<String> filterOutNonParallelGatewaySenders(Set<String> senderIds) { Set<String> allParallelSenders = cache.getAllGatewaySenders() .parallelStream() .filter(GatewaySender::isParallel) .map(GatewaySender::getId) .collect(toSet()); Set<String> parallelSenders = new HashSet<>(senderIds); parallelSenders.retainAll(allParallelSenders); return parallelSenders; } PartitionedRegion(String regionName,
      RegionAttributes regionAttributes,
      LocalRegion parentRegion,
      InternalCache cache,
      InternalRegionArguments internalRegionArgs,
      StatisticsClock statisticsClock,
      ColocationLoggerFactory colocationLoggerFactory); void setShutDownAllStatus(int newStatus); ConcurrentMap<String, Integer[]> getPartitionsMap(); void addColocationListener(ColocationListener colocationListener); void removeColocationListener(ColocationListener colocationListener); void clearNetworkHopData(); byte getNetworkHopType(); byte getMetadataVersion(); EvictionController getPREvictionControllerFromDiskInitialization(); @Override boolean remove(Object key, Object value, Object callbackArg); PartitionListener[] getPartitionListeners(); String bucketStringForLogs(int bucketId); static void clearPRIdMap(); void computeWithPrimaryLocked(Object key, Runnable r); boolean isShadowPR(); AbstractGatewaySender getParallelGatewaySender(); Set<String> getParallelGatewaySenderIds(); List<PartitionedRegion> getColocatedByList(); boolean isColocatedBy(); @Override void postCreateRegion(); @Override void initialize(InputStream snapshotInputStream, InternalDistributedMember imageTarget,
      InternalRegionArguments internalRegionArgs); void cacheRequiresNotification(); @Override void addGatewaySenderId(String gatewaySenderId); void updatePRConfigWithNewSetOfAsynchronousEventDispatchers(
      Set<String> asynchronousEventDispatchers); void updatePRConfigWithNewGatewaySenderAfterAssigningBuckets(String aeqId); @Override void removeGatewaySenderId(String gatewaySenderId); @Override void addAsyncEventQueueId(String asyncEventQueueId); @Override void addAsyncEventQueueId(String asyncEventQueueId, boolean isInternal); @Override void removeAsyncEventQueueId(String asyncEventQueueId); @Override void remoteRegionInitialized(CacheProfile profile); void validateParallelAsynchronousEventDispatcherIds(); Set<String> filterOutNonParallelGatewaySenders(Set<String> senderIds); Set<String> filterOutNonParallelAsyncEventQueues(Set<String> queueIds); void validateParallelAsynchronousEventDispatcherIds(
      Set<String> asynchronousEventDispatcherIds); boolean getRequiresNotification(); String getRegionIdentifier(); void checkPROffline(); EntrySnapshot getEntryRemotely(InternalDistributedMember targetNode, Integer bucketId,
      Object key, boolean access, boolean allowTombstones); @Override void becomeLockGrantor(); @Override Region createSubregion(String subregionName, RegionAttributes regionAttributes); @Override Lock getDistributedLock(Object key); @Override synchronized long getLastModifiedTime(); @Override synchronized long getLastAccessedTime(); @Override long getMissCount(); @Override long getHitCount(); @Override void resetCounts(); Region getSubregion(); @Override Lock getRegionDistributedLock(); @Override void loadSnapshot(InputStream inputStream); @Override void localDestroy(Object key, Object aCallbackArgument); @Override void localInvalidate(Object key, Object aCallbackArgument); @Override void localInvalidateRegion(Object aCallbackArgument); @Override Object executeQuery(final DefaultQuery query,
      final ExecutionContext executionContext,
      final Object[] parameters,
      final Set buckets); @Override void saveSnapshot(OutputStream outputStream); @Override void writeToDisk(); @Override void clear(); @Override boolean virtualPut(EntryEventImpl event, boolean ifNew, boolean ifOld,
      Object expectedOldValue, boolean requireOldValue, long lastModified,
      boolean overwriteDestroyed, boolean invokeCallbacks, boolean throwConcurrentModificaiton); @Override void checkIfAboveThreshold(EntryEventImpl entryEvent); boolean isFixedPartitionedRegion(); @Override void postPutAllFireEvents(DistributedPutAllOperation putAllOp,
      VersionedObjectList successfulPuts); @Override void postRemoveAllFireEvents(DistributedRemoveAllOperation removeAllOp,
      VersionedObjectList successfulOps); @Override long postPutAllSend(DistributedPutAllOperation putAllOp,
      VersionedObjectList successfulPuts); @Override long postRemoveAllSend(DistributedRemoveAllOperation op,
      VersionedObjectList successfulOps); VersionedObjectList tryToSendOnePutAllMessage(PutAllPRMessage prMsg,
      InternalDistributedMember currentTarget); VersionedObjectList tryToSendOneRemoveAllMessage(RemoveAllPRMessage prMsg,
      InternalDistributedMember currentTarget); InternalDistributedMember getOrCreateNodeForBucketWrite(int bucketId,
      final RetryTimeKeeper snoozer); InternalDistributedMember getNodeForBucketWrite(int bucketId,
      final RetryTimeKeeper snoozer); @Override Object get(Object key, Object aCallbackArgument, boolean generateCallbacks,
      boolean disableCopyOnRead, boolean preferCD, ClientProxyMembershipID requestingClient,
      EntryEventImpl clientEvent, boolean returnTombstones); InternalDistributedMember getOrCreateNodeForBucketRead(int bucketId); InternalDistributedMember getNodeForBucketRead(int bucketId); boolean putRemotely(final DistributedMember recipient, final EntryEventImpl event,
      boolean ifNew, boolean ifOld, Object expectedOldValue, boolean requireOldValue); InternalDistributedMember createBucket(int bucketId, int size,
      final RetryTimeKeeper snoozer); ResultCollector executeFunction(final Function function,
      final PartitionedRegionFunctionExecutor execution, ResultCollector rc,
      boolean executeOnBucketSet); ResultCollector executeOnBucketSet(final Function function,
      PartitionedRegionFunctionExecutor execution, ResultCollector rc, Set<Integer> bucketSet); Object getFromLocalBucket(int bucketId, final Object key, final Object aCallbackArgument,
      boolean disableCopyOnRead, boolean preferCD, ClientProxyMembershipID requestingClient,
      EntryEventImpl clientEvent, boolean returnTombstones); DistributedMember getMemberOwning(Object key); boolean localCacheContainsKey(Object key); Object localCacheGet(Object key); Set localCacheKeySet(); List<BucketDump> getAllBucketEntries(final int bucketId); Set getBucketKeys(int bucketNum); Set getBucketKeys(int bucketNum, boolean allowTombstones); void fetchEntries(HashMap<Integer, HashSet> bucketKeys, VersionedObjectList values,
      ServerConnection servConn); void fetchEntries(String regex, VersionedObjectList values, ServerConnection servConn); Set fetchAllLocalKeys(Integer id, Set<Integer> failures, String regex); void fetchRemoteEntries(
      HashMap<InternalDistributedMember, HashMap<Integer, HashSet>> nodeToBuckets,
      HashMap<Integer, HashSet> failures, VersionedObjectList values, ServerConnection servConn); void fetchAllRemoteEntries(
      HashMap<InternalDistributedMember, HashSet<Integer>> nodeToBuckets, HashSet<Integer> failures,
      String regex, VersionedObjectList values, ServerConnection servConn); Set<InternalDistributedMember> getAllNodes(); long getLocalSizeForTest(); Object getRemotely(InternalDistributedMember targetNode, int bucketId, final Object key,
      final Object aCallbackArgument, boolean preferCD, ClientProxyMembershipID requestingClient,
      EntryEventImpl clientEvent, boolean returnTombstones); PartitionedRegionDataStore getDataStore(); int generatePRId(InternalDistributedSystem sys); @Override DistributionAdvisor getDistributionAdvisor(); @Override CacheDistributionAdvisor getCacheDistributionAdvisor(); RegionAdvisor getRegionAdvisor(); @Override Profile getProfile(); @Override void fillInProfile(Profile p); void fillInProfile(PartitionProfile profile); static PartitionedRegion getPRFromId(int prid); static void validatePRID(InternalDistributedMember sender, int prId, String regionId); static String dumpPRId(); String dumpAllPartitionedRegions(); int getPRId(); int getTotalNumberOfBuckets(); @Override void basicDestroy(final EntryEventImpl event, final boolean cacheWrite,
      final Object expectedOldValue); void destroyInBucket(final EntryEventImpl event, Object expectedOldValue); boolean isProfileFromSameGroup(ServerBucketProfile profile); Set<String> getLocalServerGroups(); void destroyRemotely(DistributedMember recipient, Integer bucketId, EntryEventImpl event,
      Object expectedOldValue); int getLocalMaxMemory(); int getRedundantCopies(); @Override VersionTag findVersionTagForEvent(EventID eventId); @Override VersionTag findVersionTagForClientBulkOp(EventID eventId); @Override void cleanupFailedInitialization(); @Override void basicInvalidate(EntryEventImpl event); void invalidateRemotely(DistributedMember recipient, Integer bucketId,
      EntryEventImpl event); @Override Set entrySet(boolean recursive); Set<Region.Entry> entrySet(Set<Integer> bucketIds); @Override Set keys(); @Override Set keySet(boolean allowTombstones); Set keySet(Set<Integer> bucketSet); Set keysWithoutCreatesForTests(); @Override Collection values(); @Override boolean containsValue(final Object value); @Override boolean containsKey(Object key); boolean containsKeyRemotely(InternalDistributedMember targetNode, Integer bucketId,
      Object key); @Override boolean containsValueForKey(Object key); boolean containsValueForKeyRemotely(InternalDistributedMember targetNode, Integer bucketId,
      Object key); PartitionedRegionStats getPrStats(); @Override int getRegionSize(); @Override int getLocalSize(); int entryCount(boolean localOnly); @Override int entryCount(Set<Integer> buckets, boolean estimate); BucketLock getBucketLock(int bucketID); @Override String toString(); RegionLock getRegionLock(); static RegionLock getRegionLock(String regionIdentifier, InternalCache cache); RecoveryLock getRecoveryLock(); Node getNode(); @Override LoaderHelper createLoaderHelper(Object key, Object callbackArgument,
      boolean netSearchAllowed, boolean netLoadAllowed, SearchLoadAndWriteProcessor searcher); PRHARedundancyProvider getRedundancyProvider(); void checkClosed(); void checkClosedOrDestroyed(); void setHaveCacheLoader(); void checkForColocatedChildren(); @Override void destroyRegion(Object aCallbackArgument); void addMissingColocatedRegionLogger(String childName); void addMissingColocatedRegionLogger(PartitionedRegion childRegion); List<String> getMissingColocatedChildren(); void destroyParallelGatewaySenderRegion(Operation op, boolean cacheWrite, boolean lock,
      boolean callbackEvents); @Override void localDestroyRegion(Object aCallbackArgument); @Override void dispatchListenerEvent(EnumListenerEvent op, InternalCacheEvent event); @Override void cacheWriteBeforePut(EntryEventImpl event, Set netWriteRecipients,
      CacheWriter localWriter, boolean requireOldValue, Object expectedOldValue); @Override boolean cacheWriteBeforeDestroy(EntryEventImpl event, Object expectedOldValue); void dumpAllBuckets(boolean distribute); @Override void dumpBackingMap(); void validateAllBuckets(); void sendDumpB2NRegionForBucket(int bucketId); void dumpB2NForBucket(int bId); void sendDumpAllPartitionedRegions(); void dumpSelfEntryFromAllPartitionedRegions(); List getLocalBucketsListTestOnly(); List getLocalPrimaryBucketsListTestOnly(); @Override DistributionAdvisee getParentAdvisee(); String getBucketName(int bucketId); boolean isDataStore(); void getKeysWithRegEx(String regex, boolean allowTombstones, SetCollector collector); void getKeysWithList(List keyList, boolean allowTombstones, SetCollector collector); @Override boolean generateEventID(); @Override boolean shouldNotifyBridgeClients(); boolean isIndexed(); Map getIndex(); PartitionedIndex getIndex(String indexName); Collection getIndexes(); Index createIndex(boolean remotelyOriginated, IndexType indexType, String indexName,
      String indexedExpression, String fromClause, String imports, boolean loadEntries); Index createIndex(boolean remotelyOriginated, IndexType indexType, String indexName,
      String indexedExpression, String fromClause, String imports, boolean loadEntries,
      boolean sendMessage); List<Index> createIndexes(boolean remotelyOriginated,
      HashSet<IndexCreationData> indexDefinitions); void sendIndexCreationMsg(InternalDistributedMember idM); int removeIndexes(boolean remotelyOriginated); int removeIndex(Index ind, boolean remotelyOriginated); int removeIndex(String indexName); @Override Object getValueInVM(Object key); @Override Object getValueOnDisk(Object key); @Override Object getValueOnDiskOrBuffer(Object key); List<Object[]> getBucketOwnersForValidation(int bucketId); InternalDistributedMember getBucketPrimary(int bucketId); void waitOnBucketMetadataInitialization(); int getRetryTimeout(); long getBirthTime(); PartitionResolver getPartitionResolver(); String getColocatedWith(); @Override void handleInterestEvent(InterestRegistrationEvent event); @Override AttributesMutator getAttributesMutator(); @Override ExpirationAttributes setRegionTimeToLive(ExpirationAttributes timeToLive); @Override ExpirationAttributes setRegionIdleTimeout(ExpirationAttributes idleTimeout); @Override ExpirationAttributes setEntryTimeToLive(ExpirationAttributes timeToLive); @Override CustomExpiry setCustomEntryTimeToLive(CustomExpiry custom); @Override ExpirationAttributes setEntryIdleTimeout(ExpirationAttributes idleTimeout); @Override CustomExpiry setCustomEntryIdleTimeout(CustomExpiry custom); DiskRegionStats getDiskRegionStats(); @Override void removeCriticalMember(DistributedMember member); PartitionedRegion getColocatedWithRegion(); List<BucketRegion> getSortedBuckets(); @Override LocalRegion getDataRegionForRead(final KeyInfo keyInfo); @Override LocalRegion getDataRegionForWrite(KeyInfo keyInfo); @Override DistributedMember getOwnerForKey(KeyInfo key); @Override KeyInfo getKeyInfo(Object key); @Override KeyInfo getKeyInfo(Object key, Object callbackArg); @Override KeyInfo getKeyInfo(Object key, Object value, Object callbackArg); List<FixedPartitionAttributesImpl> getFixedPartitionAttributesImpl(); List<FixedPartitionAttributesImpl> getPrimaryFixedPartitionAttributes_TestsOnly(); List<FixedPartitionAttributesImpl> getSecondaryFixedPartitionAttributes_TestsOnly(); BucketRegion getBucketRegion(Object key); BucketRegion getBucketRegion(Object key, Object value); int getPerEntryLRUOverhead(); @Override boolean hasSeenEvent(EntryEventImpl ev); void enableConflation(boolean conflation); boolean isConflationEnabled(); @Override CachePerfStats getRegionPerfStats(); void updateEntryVersionInBucket(EntryEventImpl event); void shadowPRWaitForBucketRecovery(); @Override EntryExpiryTask getEntryExpiryTask(Object key); Logger getLogger(); @VisibleForTesting SenderIdMonitor getSenderIdMonitor(); @Override boolean isRegionCreateNotified(); @Override void setRegionCreateNotified(boolean notified); @Immutable
  static final Random RANDOM; static final int NETWORK_HOP_NONE; static final int NETWORK_HOP_TO_DIFFERENT_GROUP; static final String DATA_MOVED_BY_REBALANCE; @MutableForTesting
  static boolean BEFORE_CALCULATE_STARTING_BUCKET_FLAG; final PartitionedRegionStats prStats; static final int DEFAULT_RETRY_ITERATIONS; volatile boolean isClosed; volatile boolean isLocallyDestroyed; static final int RUNNING_MODE; static final int PRIMARY_BUCKETS_LOCKED; static final int DISK_STORE_FLUSHED; static final int OFFLINE_EQUAL_PERSISTED; static final String BUCKET_ID_SEPARATOR; static final String RETRY_TIMEOUT_PROPERTY; static final String BUCKET_NAME_SEPARATOR; }
@Test public void filterOutNonParallelGatewaySendersShouldReturnCorrectly() { GatewaySender parallelSender = mock(GatewaySender.class); GatewaySender anotherParallelSender = mock(GatewaySender.class); GatewaySender serialSender = mock(GatewaySender.class); when(parallelSender.isParallel()) .thenReturn(true); when(parallelSender.getId()) .thenReturn("parallel"); when(anotherParallelSender.isParallel()) .thenReturn(true); when(anotherParallelSender.getId()) .thenReturn("anotherParallel"); when(serialSender.isParallel()) .thenReturn(false); when(cache.getAllGatewaySenders()) .thenReturn(asSet(parallelSender, anotherParallelSender, serialSender)); assertThat(partitionedRegion.filterOutNonParallelGatewaySenders(asSet("serial"))) .isEmpty(); assertThat(partitionedRegion.filterOutNonParallelGatewaySenders(asSet("unknownSender"))) .isEmpty(); assertThat(partitionedRegion.filterOutNonParallelGatewaySenders(asSet("parallel", "serial"))) .containsExactly("parallel"); assertThat(partitionedRegion .filterOutNonParallelGatewaySenders(asSet("parallel", "serial", "anotherParallel"))) .containsExactly("parallel", "anotherParallel"); }