@Test public void testShouldReturnKeysByBatches() throws Exception { IgniteEx node = startGrids(3); node.cluster().active(true); AffinityTopologyVersion ver = lastTopologyVersion(node); IgniteCache<Object, Object> cache = node.cache(DEFAULT_CACHE_NAME); List<Integer> keys = new ArrayList<>(); int elements = 0; for (int i = 0; i < 100_000; i++) { if (node.affinity(DEFAULT_CACHE_NAME).partition(i) == FIRST_PARTITION) { cache.put(i, i); keys.add(i); elements++; } } for (Integer key : keys) { corruptDataEntry( node.cachex(DEFAULT_CACHE_NAME).context(), key ); } Collection<ClusterNode> nodes = node.affinity(DEFAULT_CACHE_NAME) .mapPartitionToPrimaryAndBackups(FIRST_PARTITION); int batchSize = (elements + 1) / 2; Set<KeyCacheObject> fetched = new HashSet<>(); T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>> firstBatch = node.compute(group(node, nodes)).execute( CollectPartitionKeysByBatchTask.class, new PartitionBatchRequest(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), DEFAULT_CACHE_NAME, FIRST_PARTITION, batchSize, null, ver) ).result(); fetched.addAll(firstBatch.get2().keySet()); KeyCacheObject firstMaxKey = firstBatch.get1(); T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>> secondBatch = node.compute(group(node, nodes)).execute( CollectPartitionKeysByBatchTask.class, new PartitionBatchRequest(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), DEFAULT_CACHE_NAME, FIRST_PARTITION, batchSize, firstMaxKey, ver) ).result(); KeyCacheObject secondMaxKey = secondBatch.get1(); fetched.addAll(secondBatch.get2().keySet()); T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>> thirdBatch = node.compute(group(node, nodes)).execute( CollectPartitionKeysByBatchTask.class, new PartitionBatchRequest(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), DEFAULT_CACHE_NAME, FIRST_PARTITION, batchSize, secondMaxKey, ver) ).result(); assertNull(thirdBatch.get1()); assertEquals(0, thirdBatch.get2().size()); assertEquals(elements, fetched.size()); }
@Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) { ComputeJobResultPolicy superRes = super.result(res, rcvd); if (superRes == ComputeJobResultPolicy.FAILOVER) { superRes = ComputeJobResultPolicy.WAIT; log.warning("CollectPartitionEntryHashesJob failed on node " + "[consistentId=" + res.getNode().consistentId() + "]", res.getException()); } return superRes; }
CollectPartitionKeysByBatchTask extends ComputeTaskAdapter<PartitionBatchRequest, ExecutionResult<T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>>>> { @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) { ComputeJobResultPolicy superRes = super.result(res, rcvd); if (superRes == ComputeJobResultPolicy.FAILOVER) { superRes = ComputeJobResultPolicy.WAIT; log.warning("CollectPartitionEntryHashesJob failed on node " + "[consistentId=" + res.getNode().consistentId() + "]", res.getException()); } return superRes; } }
CollectPartitionKeysByBatchTask extends ComputeTaskAdapter<PartitionBatchRequest, ExecutionResult<T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>>>> { @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) { ComputeJobResultPolicy superRes = super.result(res, rcvd); if (superRes == ComputeJobResultPolicy.FAILOVER) { superRes = ComputeJobResultPolicy.WAIT; log.warning("CollectPartitionEntryHashesJob failed on node " + "[consistentId=" + res.getNode().consistentId() + "]", res.getException()); } return superRes; }  }
CollectPartitionKeysByBatchTask extends ComputeTaskAdapter<PartitionBatchRequest, ExecutionResult<T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>>>> { @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) { ComputeJobResultPolicy superRes = super.result(res, rcvd); if (superRes == ComputeJobResultPolicy.FAILOVER) { superRes = ComputeJobResultPolicy.WAIT; log.warning("CollectPartitionEntryHashesJob failed on node " + "[consistentId=" + res.getNode().consistentId() + "]", res.getException()); } return superRes; }  @NotNull @Override Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
        PartitionBatchRequest partBatch); @Override ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd); @Override @Nullable ExecutionResult<T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>>> reduce(
        List<ComputeJobResult> results); }
CollectPartitionKeysByBatchTask extends ComputeTaskAdapter<PartitionBatchRequest, ExecutionResult<T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>>>> { @Override public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd) { ComputeJobResultPolicy superRes = super.result(res, rcvd); if (superRes == ComputeJobResultPolicy.FAILOVER) { superRes = ComputeJobResultPolicy.WAIT; log.warning("CollectPartitionEntryHashesJob failed on node " + "[consistentId=" + res.getNode().consistentId() + "]", res.getException()); } return superRes; }  @NotNull @Override Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
        PartitionBatchRequest partBatch); @Override ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd); @Override @Nullable ExecutionResult<T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>>> reduce(
        List<ComputeJobResult> results);  }