PartitionedRegionQueryEvaluator extends StreamingPartitionOperation { public SelectResults queryBuckets(final TestHook th) throws QueryException, InterruptedException { final boolean isDebugEnabled = logger.isDebugEnabled(); if (Thread.interrupted()) { throw new InterruptedException(); } if (isDebugEnabled) { logger.debug("PRQE query :{}", this.query.getQueryString()); } Assert.assertTrue(!(this.bucketsToQuery == null || this.bucketsToQuery.isEmpty()), "bucket set is empty."); this.node2bucketIds = buildNodeToBucketMap(); Assert.assertTrue(!this.node2bucketIds.isEmpty(), " There are no data stores hosting any of the buckets."); boolean needsRetry = true; int retry = 0; while (needsRetry && retry < MAX_PR_QUERY_RETRIES) { needsRetry = executeQueryOnRemoteAndLocalNodes(th); if (th != null) { th.hook(1); } if (needsRetry) { if (query.isQueryWithFunctionContext()) { if (isDebugEnabled) { logger.debug("No of retry attempts are: {}", retry); } break; } Map b2n = buildNodeToBucketMapForBuckets(calculateRetryBuckets()); if (th != null) { th.hook(2); } this.node2bucketIds = b2n; if (isDebugEnabled) { logger.debug("PR Query retry: {} total: {}", retry, this.pr.getCachePerfStats().getPRQueryRetries()); } this.pr.getCachePerfStats().incPRQueryRetries(); retry++; waitBeforeRetry(); } if (th != null) { th.hook(3); } } if (needsRetry) { String msg = "Failed to query all the partitioned region " + "dataset (buckets) after " + retry + " attempts."; if (isDebugEnabled) { logger.debug("{} Unable to query some of the buckets from the set :{}", msg, this.calculateRetryBuckets()); } throw new QueryException(msg); } return addResultsToResultSet(); } PartitionedRegionQueryEvaluator(final InternalDistributedSystem sys,
      final PartitionedRegion pr,
      final DefaultQuery query,
      final ExecutionContext executionContext,
      final Object[] parameters,
      final SelectResults cumulativeResults,
      final Set<Integer> bucketsToQuery); boolean executeQueryOnRemoteAndLocalNodes(final TestHook th); SelectResults queryBuckets(final TestHook th); Map getResultsPerMember();  }
@Test public void testLocalQueryReturnsResultsToPartitionedQueryEvaluator() throws Exception { List resultsForMember1 = createResultObjects("1", "2", "3"); PartitionedQueryScenario scenario = new PartitionedQueryScenario(localNode, allNodes, noFailingMembers, createFakeBucketMap(), new ProcessDataFaker() { @Override public void processData(PartitionedRegionQueryEvaluator prqe) { } @Override @SuppressWarnings("unchecked") public void executeQueryLocally(Collection resultsCollector) { resultsCollector.add(resultsForMember1); } }); Set<Integer> allBucketsToQuery = scenario.getAllBucketsToQuery(); Queue<PartitionedQueryScenario> scenarios = createScenariosQueue(scenario); dataStore.setScenarios(scenarios); PartitionedRegionQueryEvaluator prqe = new ExtendedPartitionedRegionQueryEvaluator(system, pr, query, mock(ExecutionContext.class), null, new LinkedResultSet(), allBucketsToQuery, scenarios); Collection results = prqe.queryBuckets(null).asList(); assertThat(results).isNotNull(); assertThat(results.size()).isEqualTo(resultsForMember1.size()); results.removeAll(resultsForMember1); assertThat(results.isEmpty()).isTrue(); }