test_class.declarator: DEFAULT_CACHE = "default-cache"
test_class.modifier: private static final
test_class.original_string: private static final String DEFAULT_CACHE = "default-cache";
test_class.type: String
test_class.var_name: DEFAULT_CACHE
test_class.declarator: PARTITION_ID = 123
test_class.modifier: private static final
test_class.original_string: private static final int PARTITION_ID = 123;
test_class.type: int
test_class.var_name: PARTITION_ID
test_class.declarator: SESSION_ID = 123
test_class.modifier: private static final
test_class.original_string: private static final long SESSION_ID = 123;
test_class.type: long
test_class.var_name: SESSION_ID
test_class.declarator: MAX_RECHECK_ATTEMPTS = 3
test_class.modifier: private static final
test_class.original_string: private static final int MAX_RECHECK_ATTEMPTS = 3;
test_class.type: int
test_class.var_name: MAX_RECHECK_ATTEMPTS
test_class.file: modules/core/src/test/java/org/apache/ignite/internal/processors/cache/checker/processor/PartitionReconciliationProcessorTest.java
test_class.identifier: PartitionReconciliationProcessorTest
test_case.body: @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_case.class_method_signature: PartitionReconciliationProcessorTest.testBatchHasElementsRecheckAndNextBatchShouldSchedule()
test_case.full_signature: @Test public void testBatchHasElementsRecheckAndNextBatchShouldSchedule()
test_case.identifier: testBatchHasElementsRecheckAndNextBatchShouldSchedule
test_case.invocations: create put execute whereResult addTask randomUUID schedule verify times any schedule verify times any eq eq
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testBatchHasElementsRecheckAndNextBatchShouldSchedule()
focal_class.declarator: SESSION_CHANGE_MSG = "Reconciliation session has changed."
focal_class.modifier: public static final
focal_class.original_string: public static final String SESSION_CHANGE_MSG = "Reconciliation session has changed.";
focal_class.type: String
focal_class.var_name: SESSION_CHANGE_MSG
focal_class.declarator: TOPOLOGY_CHANGE_MSG = "Topology has changed. Partition reconciliation task was stopped."
focal_class.modifier: public static final
focal_class.original_string: public static final String TOPOLOGY_CHANGE_MSG = "Topology has changed. Partition reconciliation task was stopped.";
focal_class.type: String
focal_class.var_name: TOPOLOGY_CHANGE_MSG
focal_class.declarator: WORK_PROGRESS_MSG = "Partition reconciliation task [sesId=%s, total=%s, remaining=%s]"
focal_class.modifier: public static final
focal_class.original_string: public static final String WORK_PROGRESS_MSG = "Partition reconciliation task [sesId=%s, total=%s, remaining=%s]";
focal_class.type: String
focal_class.var_name: WORK_PROGRESS_MSG
focal_class.declarator: START_EXECUTION_MSG = "Partition reconciliation has started [repair=%s, repairAlg=%s, " +
          "fastCheck=%s, batchSize=%s, recheckAttempts=%s, parallelismLevel=%s, caches=%s]"
focal_class.modifier: public static final
focal_class.original_string: public static final String START_EXECUTION_MSG = "Partition reconciliation has started [repair=%s, repairAlg=%s, " +
          "fastCheck=%s, batchSize=%s, recheckAttempts=%s, parallelismLevel=%s, caches=%s]";
focal_class.type: String
focal_class.var_name: START_EXECUTION_MSG
focal_class.declarator: ERROR_REASON = "Reason [msg=%s, exception=%s]"
focal_class.modifier: public static final
focal_class.original_string: public static final String ERROR_REASON = "Reason [msg=%s, exception=%s]";
focal_class.type: String
focal_class.var_name: ERROR_REASON
focal_class.declarator: workProgressPrintInterval = getLong("WORK_PROGRESS_PRINT_INTERVAL", 1000 * 60 * 3)
focal_class.modifier: private final
focal_class.original_string: private final long workProgressPrintInterval = getLong("WORK_PROGRESS_PRINT_INTERVAL", 1000 * 60 * 3);
focal_class.type: long
focal_class.var_name: workProgressPrintInterval
focal_class.declarator: recheckDelay
focal_class.modifier: private final
focal_class.original_string: private final int recheckDelay;
focal_class.type: int
focal_class.var_name: recheckDelay
focal_class.declarator: caches
focal_class.modifier: private final
focal_class.original_string: private final Collection<String> caches;
focal_class.type: Collection<String>
focal_class.var_name: caches
focal_class.declarator: repair
focal_class.modifier: private final
focal_class.original_string: private final boolean repair;
focal_class.type: boolean
focal_class.var_name: repair
focal_class.declarator: partsToValidate
focal_class.modifier: private final
focal_class.original_string: private final Map<Integer, Set<Integer>> partsToValidate;
focal_class.type: Map<Integer, Set<Integer>>
focal_class.var_name: partsToValidate
focal_class.declarator: batchSize
focal_class.modifier: private final
focal_class.original_string: private final int batchSize;
focal_class.type: int
focal_class.var_name: batchSize
focal_class.declarator: recheckAttempts
focal_class.modifier: private final
focal_class.original_string: private final int recheckAttempts;
focal_class.type: int
focal_class.var_name: recheckAttempts
focal_class.declarator: repairAlg
focal_class.modifier: private final
focal_class.original_string: private final RepairAlgorithm repairAlg;
focal_class.type: RepairAlgorithm
focal_class.var_name: repairAlg
focal_class.declarator: workloadTracker = new WorkloadTracker()
focal_class.modifier: private final
focal_class.original_string: private final WorkloadTracker workloadTracker = new WorkloadTracker();
focal_class.type: WorkloadTracker
focal_class.var_name: workloadTracker
focal_class.declarator: collector
focal_class.modifier: final
focal_class.original_string: final ReconciliationResultCollector collector;
focal_class.type: ReconciliationResultCollector
focal_class.var_name: collector
focal_class.file: modules/core/src/main/java/org/apache/ignite/internal/processors/cache/checker/processor/PartitionReconciliationProcessor.java
focal_class.identifier: PartitionReconciliationProcessor
focal_class.class_method_signature: PartitionReconciliationProcessor.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
      )
focal_class.full_signature: @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
      )
focal_class.identifier: PartitionReconciliationProcessor
focal_class.modifiers: @SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType") public
focal_class.parameters: (
          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
      )
focal_class.signature:  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
      )
focal_class.class_method_signature: PartitionReconciliationProcessor.execute()
focal_class.full_signature: public ExecutionResult<ReconciliationAffectedEntries> execute()
focal_class.identifier: execute
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: ExecutionResult<ReconciliationAffectedEntries>
focal_class.signature: ExecutionResult<ReconciliationAffectedEntries> execute()
focal_class.class_method_signature: PartitionReconciliationProcessor.collector()
focal_class.full_signature: public ReconciliationResultCollector collector()
focal_class.identifier: collector
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: ReconciliationResultCollector
focal_class.signature: ReconciliationResultCollector collector()
focal_class.class_method_signature: PartitionReconciliationProcessor.partitions(String name)
focal_class.full_signature: private int[] partitions(String name)
focal_class.identifier: partitions
focal_class.modifiers: private
focal_class.parameters: (String name)
focal_class.return: int[]
focal_class.signature: int[] partitions(String name)
focal_class.class_method_signature: PartitionReconciliationProcessor.handle(Batch workload)
focal_class.full_signature: private void handle(Batch workload)
focal_class.identifier: handle
focal_class.modifiers: private
focal_class.parameters: (Batch workload)
focal_class.return: void
focal_class.signature: void handle(Batch workload)
focal_class.class_method_signature: PartitionReconciliationProcessor.handle(Recheck workload)
focal_class.full_signature: private void handle(Recheck workload)
focal_class.identifier: handle
focal_class.modifiers: private
focal_class.parameters: (Recheck workload)
focal_class.return: void
focal_class.signature: void handle(Recheck workload)
focal_class.class_method_signature: PartitionReconciliationProcessor.handle(Repair workload)
focal_class.full_signature: private void handle(Repair workload)
focal_class.identifier: handle
focal_class.modifiers: private
focal_class.parameters: (Repair workload)
focal_class.return: void
focal_class.signature: void handle(Repair workload)
focal_class.class_method_signature: PartitionReconciliationProcessor.repair(
          long sesId,
          UUID workloadChainId,
          String cacheName,
          int partId,
          Map<KeyCacheObject, Map<UUID, GridCacheVersion>> notResolvingConflicts,
          Map<KeyCacheObject, Map<UUID, VersionedValue>> actualKeys,
          int repairAttempts
      )
focal_class.full_signature: private Repair repair(
          long sesId,
          UUID workloadChainId,
          String cacheName,
          int partId,
          Map<KeyCacheObject, Map<UUID, GridCacheVersion>> notResolvingConflicts,
          Map<KeyCacheObject, Map<UUID, VersionedValue>> actualKeys,
          int repairAttempts
      )
focal_class.identifier: repair
focal_class.modifiers: private
focal_class.parameters: (
          long sesId,
          UUID workloadChainId,
          String cacheName,
          int partId,
          Map<KeyCacheObject, Map<UUID, GridCacheVersion>> notResolvingConflicts,
          Map<KeyCacheObject, Map<UUID, VersionedValue>> actualKeys,
          int repairAttempts
      )
focal_class.return: Repair
focal_class.signature: Repair repair(
          long sesId,
          UUID workloadChainId,
          String cacheName,
          int partId,
          Map<KeyCacheObject, Map<UUID, GridCacheVersion>> notResolvingConflicts,
          Map<KeyCacheObject, Map<UUID, VersionedValue>> actualKeys,
          int repairAttempts
      )
focal_class.superclass: extends AbstractPipelineProcessor
focal_method.body: 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()));
          }
      }
focal_method.class_method_signature: PartitionReconciliationProcessor.execute()
focal_method.full_signature: public ExecutionResult<ReconciliationAffectedEntries> execute()
focal_method.identifier: execute
focal_method.invocations: isInfoEnabled info format cachex expiry context warning partitions randomUUID addTrackingChain schedule isEmpty hasLiveHandlers topologyChanged isSessionExpired isInterrupted get isEmpty sleep currentTimeMillis isInfoEnabled info format totalChains remaningChains takeTask handle handle handle error result waitWorkFinish warning result format getMessage getClass error result format getMessage getClass
focal_method.modifiers: public
focal_method.parameters: ()
focal_method.return: ExecutionResult<ReconciliationAffectedEntries>
focal_method.signature: ExecutionResult<ReconciliationAffectedEntries> execute()