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; }
@Test public void testBatchDoesNotHaveElementsNothingSchedule() throws IgniteCheckedException { MockedProcessor processor = MockedProcessor.create(false); ExecutionResult<T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>>> emptyRes = new ExecutionResult<>(new T2<>(null, new HashMap<>())); processor.addTask(new Batch(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), DEFAULT_CACHE, PARTITION_ID, null)) .whereResult(CollectPartitionKeysByBatchTask.class, emptyRes) .execute(); processor.verify(never()).schedule(any()); processor.verify(never()).schedule(any(), anyInt(), any()); processor.verify(never()).scheduleHighPriority(any()); }
@Test public void testBatchHasElementsRecheckAndNextBatchShouldSchedule() throws IgniteCheckedException { MockedProcessor processor = MockedProcessor.create(false); KeyCacheObject nextKey = new KeyCacheObjectImpl(1, null, PARTITION_ID); Map<KeyCacheObject, Map<UUID, GridCacheVersion>> batchRes = new HashMap<>(); batchRes.put(nextKey, new HashMap<>()); ExecutionResult<T2<KeyCacheObject, Map<KeyCacheObject, Map<UUID, GridCacheVersion>>>> emptyRes = new ExecutionResult<>(new T2<>(nextKey, batchRes)); processor.addTask(new Batch(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), DEFAULT_CACHE, PARTITION_ID, null)) .whereResult(CollectPartitionKeysByBatchTask.class, emptyRes) .execute(); processor.verify(times(1)).schedule(any(Batch.class)); processor.verify(times(1)).schedule(any(Recheck.class), eq(10L), eq(SECONDS)); }
@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()); }
@Test public void testRecheckShouldFinishWithoutActionIfConflictWasSolved() throws IgniteCheckedException { UUID nodeId1 = UUID.randomUUID(); UUID nodeId2 = UUID.randomUUID(); MockedProcessor processor = MockedProcessor.create(false, Arrays.asList(nodeId1, nodeId2)); KeyCacheObjectImpl key = new KeyCacheObjectImpl(1, null, PARTITION_ID); GridCacheVersion ver = new GridCacheVersion(1, 0, 0, 0); Map<KeyCacheObject, Map<UUID, GridCacheVersion>> batchRes = new HashMap<>(); Map<UUID, GridCacheVersion> oldKeys = new HashMap<>(); oldKeys.put(nodeId1, ver); oldKeys.put(nodeId2, ver); batchRes.put(key, oldKeys); Map<KeyCacheObject, Map<UUID, VersionedValue>> sameRes = new HashMap<>(); Map<UUID, VersionedValue> actualKey = new HashMap<>(); actualKey.put(nodeId1, new VersionedValue(null, ver, 1, 1)); actualKey.put(nodeId2, new VersionedValue(null, ver, 1, 1)); sameRes.put(key, actualKey); processor.addTask(new Recheck(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), batchRes, DEFAULT_CACHE, PARTITION_ID, 0, 0)) .whereResult(CollectPartitionKeysByRecheckRequestTask.class, new ExecutionResult<>(sameRes)) .execute(); processor.verify(never()).schedule(any()); processor.verify(never()).schedule(any(), anyInt(), any()); processor.verify(never()).scheduleHighPriority(any()); }
@Test public void testRecheckShouldTryAgainIfConflictAndAttemptsExist() throws IgniteCheckedException { MockedProcessor processor = MockedProcessor.create(false); KeyCacheObjectImpl key = new KeyCacheObjectImpl(1, null, PARTITION_ID); UUID nodeId1 = UUID.randomUUID(); UUID nodeId2 = UUID.randomUUID(); GridCacheVersion ver = new GridCacheVersion(1, 0, 0, 0); GridCacheVersion ver2 = new GridCacheVersion(2, 0, 0, 0); Map<KeyCacheObject, Map<UUID, GridCacheVersion>> batchRes = new HashMap<>(); Map<UUID, GridCacheVersion> oldKeys = new HashMap<>(); oldKeys.put(nodeId1, ver); oldKeys.put(nodeId2, ver2); batchRes.put(key, oldKeys); Map<KeyCacheObject, Map<UUID, VersionedValue>> sameRes = new HashMap<>(); Map<UUID, VersionedValue> actualKey = new HashMap<>(); actualKey.put(nodeId1, new VersionedValue(null, ver, 1, 1)); actualKey.put(nodeId2, new VersionedValue(null, ver2, 1, 1)); sameRes.put(key, actualKey); processor.addTask(new Recheck(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), batchRes, DEFAULT_CACHE, PARTITION_ID, 0, 0)) .whereResult(CollectPartitionKeysByRecheckRequestTask.class, new ExecutionResult<>(sameRes)) .execute(); processor.verify(times(1)).schedule(any(Recheck.class), eq(10L), eq(SECONDS)); }
@Test public void testRecheckShouldTryRepairIfAttemptsDoesNotExist() throws IgniteCheckedException { MockedProcessor processor = MockedProcessor.create(true); KeyCacheObjectImpl key = new KeyCacheObjectImpl(1, null, PARTITION_ID); UUID nodeId1 = UUID.randomUUID(); UUID nodeId2 = UUID.randomUUID(); GridCacheVersion ver = new GridCacheVersion(1, 0, 0, 0); GridCacheVersion ver2 = new GridCacheVersion(2, 0, 0, 0); Map<KeyCacheObject, Map<UUID, GridCacheVersion>> batchRes = new HashMap<>(); Map<UUID, GridCacheVersion> oldKeys = new HashMap<>(); oldKeys.put(nodeId1, ver); oldKeys.put(nodeId2, ver2); batchRes.put(key, oldKeys); Map<KeyCacheObject, Map<UUID, VersionedValue>> sameRes = new HashMap<>(); Map<UUID, VersionedValue> actualKey = new HashMap<>(); actualKey.put(nodeId1, new VersionedValue(null, ver, 1, 1)); actualKey.put(nodeId2, new VersionedValue(null, ver2, 1, 1)); sameRes.put(key, actualKey); processor.addTask(new Recheck(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), batchRes, DEFAULT_CACHE, PARTITION_ID, MAX_RECHECK_ATTEMPTS, RepairRequestTask.MAX_REPAIR_ATTEMPTS)) .whereResult(CollectPartitionKeysByRecheckRequestTask.class, new ExecutionResult<>(sameRes)) .execute(); processor.verify(times(1)).scheduleHighPriority(any(Repair.class)); }
@Test public void testThatRepairHaveHigherPriorityThenChecking() throws IgniteCheckedException { List<String> evtHist = new ArrayList<>(); MockedProcessor processor = MockedProcessor.create(true); processor.useRealScheduler = true; processor.registerListener((stage, workload) -> { if (stage.equals(ReconciliationEventListener.WorkLoadStage.READY)) evtHist.add(workload.getClass().getName()); }); KeyCacheObjectImpl key = new KeyCacheObjectImpl(1, null, PARTITION_ID); UUID nodeId1 = UUID.randomUUID(); UUID nodeId2 = UUID.randomUUID(); GridCacheVersion ver = new GridCacheVersion(1, 0, 0, 0); GridCacheVersion ver2 = new GridCacheVersion(2, 0, 0, 0); Map<KeyCacheObject, Map<UUID, GridCacheVersion>> batchRes = new HashMap<>(); Map<UUID, GridCacheVersion> oldKeys = new HashMap<>(); oldKeys.put(nodeId1, ver); oldKeys.put(nodeId2, ver2); batchRes.put(key, oldKeys); Map<KeyCacheObject, Map<UUID, VersionedValue>> sameRes = new HashMap<>(); Map<UUID, VersionedValue> actualKey = new HashMap<>(); actualKey.put(nodeId1, new VersionedValue(null, ver, 1, 1)); actualKey.put(nodeId2, new VersionedValue(null, ver2, 1, 1)); sameRes.put(key, actualKey); processor .addTask(new Recheck(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), batchRes, DEFAULT_CACHE, PARTITION_ID, MAX_RECHECK_ATTEMPTS, RepairRequestTask.MAX_REPAIR_ATTEMPTS)) .addTask(new Recheck(ReconciliationExecutionContext.IGNORE_JOB_PERMITS_SESSION_ID, UUID.randomUUID(), batchRes, DEFAULT_CACHE, PARTITION_ID, MAX_RECHECK_ATTEMPTS, RepairRequestTask.MAX_REPAIR_ATTEMPTS)) .whereResult(CollectPartitionKeysByRecheckRequestTask.class, new ExecutionResult<>(sameRes)) .whereResult(RepairRequestTask.class, new ExecutionResult<>(new RepairResult())) .execute(); assertEquals(RecheckRequest.class.getName(), evtHist.get(0)); assertEquals(RepairRequest.class.getName(), evtHist.get(1)); assertEquals(RecheckRequest.class.getName(), evtHist.get(2)); }