@Test public void testRecheckShouldFinishWithoutActionIfResultEmpty() throws IgniteCheckedException { MockedProcessor processor = MockedProcessor.create(false); Map<KeyCacheObject, Map<UUID, GridCacheVersion>> batchRes = new HashMap<>(); batchRes.put(new KeyCacheObjectImpl(1, null, PARTITION_ID), new HashMap<>()); ExecutionResult<Map<KeyCacheObject, Map<UUID, VersionedValue>>> emptyRes = new ExecutionResult<>(new HashMap<>()); processor.addTask(new Recheck(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), batchRes, DEFAULT_CACHE, PARTITION_ID, 0, 0)) .whereResult(CollectPartitionKeysByRecheckRequestTask.class, emptyRes) .execute(); processor.verify(never()).schedule(any()); processor.verify(never()).schedule(any(), anyInt(), any()); processor.verify(never()).scheduleHighPriority(any()); }
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())); } }
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())); } } }
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")  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
    ); }
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")  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
    ); ExecutionResult<ReconciliationAffectedEntries> execute(); ReconciliationResultCollector collector(); }
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")  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
    ); ExecutionResult<ReconciliationAffectedEntries> execute(); ReconciliationResultCollector collector(); static final String SESSION_CHANGE_MSG; static final String TOPOLOGY_CHANGE_MSG; static final String WORK_PROGRESS_MSG; static final String START_EXECUTION_MSG; static final String ERROR_REASON; }