CollectPartitionKeysByRecheckRequestTask extends ComputeTaskAdapter<RecheckRequest, ExecutionResult<Map<KeyCacheObject, Map<UUID, VersionedValue>>>> { @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; }  @Override Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid,
        RecheckRequest arg); @Override ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> rcvd); @Override ExecutionResult<Map<KeyCacheObject, Map<UUID, VersionedValue>>> reduce(
        List<ComputeJobResult> results);  }
@Test public void testShouldReturnAllRequiredKeys() throws Exception { IgniteEx node = startGrids(2); node.cluster().active(true); IgniteInternalCache<Object, Object> cache = node.cachex(DEFAULT_CACHE_NAME); cache.put(1, 1); cache.put(2, 2); CacheObjectContext ctx = cache.context().cacheObjectContext(); Collection<ClusterNode> nodes = node.affinity(DEFAULT_CACHE_NAME) .mapPartitionToPrimaryAndBackups(FIRST_PARTITION); List<KeyCacheObject> recheckKeys = new ArrayList<>(); recheckKeys.add(key(1, ctx)); recheckKeys.add(key(2, ctx)); Map<KeyCacheObject, Map<UUID, VersionedValue>> res = node.compute(group(node, nodes)).execute( CollectPartitionKeysByRecheckRequestTask.class, new RecheckRequest(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), recheckKeys, DEFAULT_CACHE_NAME, FIRST_PARTITION, lastTopologyVersion(node)) ).result(); assertEquals(2, res.size()); assertTrue(res.keySet().containsAll(recheckKeys)); }
@Test public void testEmptyKeysSelectEmptyResult() throws Exception { IgniteEx node = startGrids(2); node.cluster().active(true); Collection<ClusterNode> nodes = node.affinity(DEFAULT_CACHE_NAME) .mapPartitionToPrimaryAndBackups(FIRST_PARTITION); List<KeyCacheObject> recheckKeys = new ArrayList<>(); Map<KeyCacheObject, Map<UUID, VersionedValue>> res = node.compute(group(node, nodes)).execute( CollectPartitionKeysByRecheckRequestTask.class, new RecheckRequest(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), recheckKeys, DEFAULT_CACHE_NAME, FIRST_PARTITION, lastTopologyVersion(node)) ).result(); assertTrue(res.isEmpty()); }
@Test public void testRemovedKeyShouldReturnEmptyResult() throws Exception { IgniteEx node = startGrids(2); node.cluster().active(true); Collection<ClusterNode> nodes = node.affinity(DEFAULT_CACHE_NAME) .mapPartitionToPrimaryAndBackups(FIRST_PARTITION); List<KeyCacheObject> recheckKeys = new ArrayList<>(); String keyRow = "test-key"; KeyCacheObjectImpl key = key(keyRow, node.cachex(DEFAULT_CACHE_NAME).context().cacheObjectContext()); recheckKeys.add(key); Map<KeyCacheObject, Map<UUID, VersionedValue>> res = node.compute(group(node, nodes)).execute( CollectPartitionKeysByRecheckRequestTask.class, new RecheckRequest(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), recheckKeys, DEFAULT_CACHE_NAME, FIRST_PARTITION, lastTopologyVersion(node)) ).result(); assertTrue(res.isEmpty()); }