@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; // two batch for all rows
    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());
}