class PartitionReconciliationProcessor extends AbstractPipelineProcessor {

    public ExecutionResult<ReconciliationAffectedEntries> execute() {
        if (log.isInfoEnabled()) {
            log.info(String.format(
                START_EXECUTION_MSG,
                repair,
                repairAlg,
                partsToValidate != null,
                batchSize,
                recheckAttempts,
                parallelismLevel,
                caches));
        }

        try {
            for (String cache : caches) {
                IgniteInternalCache<Object, Object> cachex = ignite.cachex(cache);

                ExpiryPolicy expPlc = cachex.context().expiry();
                if (expPlc != null && !(expPlc instanceof EternalExpiryPolicy)) {
                    log.warning("The cache '" + cache + "' was skipped because CacheConfiguration#setExpiryPolicyFactory is set.");

                    continue;
                }

                int[] partitions = partitions(cache);

                for (int partId : partitions) {
                    Batch workload = new Batch(sesId, UUID.randomUUID(), cache, partId, null);

                    workloadTracker.addTrackingChain(workload);

                    schedule(workload);
                }
            }

            boolean live = false;
            long lastUpdateTime = 0;

            while (!isEmpty() || (live = hasLiveHandlers())) {
                if (topologyChanged())
                    throw new IgniteException(TOPOLOGY_CHANGE_MSG);

                if (isSessionExpired())
                    throw new IgniteException(SESSION_CHANGE_MSG);

                if (isInterrupted())
                    throw new IgniteException(error.get());

                if (isEmpty() && live) {
                    U.sleep(100);

                    continue;
                }

                long currTimeMillis = System.currentTimeMillis();

                if (currTimeMillis >= lastUpdateTime + workProgressPrintInterval) {
                    if (log.isInfoEnabled()) {
                        log.info(String.format(
                            WORK_PROGRESS_MSG,
                            sesId,
                            workloadTracker.totalChains(),
                            workloadTracker.remaningChains()));
                    }

                    lastUpdateTime = currTimeMillis;
                }

                PipelineWorkload workload = takeTask();

                if (workload instanceof Batch)
                    handle((Batch)workload);
                else if (workload instanceof Recheck)
                    handle((Recheck)workload);
                else if (workload instanceof Repair)
                    handle((Repair)workload);
                else {
                    String err = "Unsupported workload type: " + workload;

                    log.error(err);

                    throw new IgniteException(err);
                }
            }

            return new ExecutionResult<>(collector.result());
        }
        catch (InterruptedException | IgniteException e) {
            String errMsg = "Partition reconciliation was interrupted.";

            waitWorkFinish();

            log.warning(errMsg, e);

            return new ExecutionResult<>(collector.result(), errMsg + ' ' + String.format(ERROR_REASON, e.getMessage(), e.getClass()));
        }
        catch (Exception e) {
            String errMsg = "Unexpected error.";

            log.error(errMsg, e);

            return new ExecutionResult<>(collector.result(), errMsg + ' ' + String.format(ERROR_REASON, e.getMessage(), e.getClass()));
        }
    }

    @SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType") public  PartitionReconciliationProcessor(
        long sesId,
        IgniteEx ignite,
        Collection<String> caches,
        Map<Integer, Set<Integer>> partsToValidate,
        boolean repair,
        RepairAlgorithm repairAlg,
        int parallelismLevel,
        int batchSize,
        int recheckAttempts,
        int recheckDelay,
        boolean compact,
        boolean includeSensitive
    );

}