test_class.file: spark-job/src/test/java/org/apache/cassandra/diff/DifferTest.java
test_class.identifier: DifferTest
test_case.body: @Test
      public void filterTaskStatusForTables() {
          // according to the journal:
          // * t1 is already completed
          // * t2 is started and has reported some progress, but has not completed
          // * t3 has not reported any progress
          DiffJob.Split split = new DiffJob.Split(1, 1, BigInteger.ONE, BigInteger.TEN);
          Iterable<KeyspaceTablePair> tables = Lists.newArrayList(ksTbl("t1"), ksTbl("t2"), ksTbl("t3"));
          Function<KeyspaceTablePair, DiffJob.TaskStatus> journal = (keyspaceTable) -> {
              switch (keyspaceTable.table) {
                  case "t1":
                      return new DiffJob.TaskStatus(split.end, RangeStats.withValues(6, 6, 6, 6, 6, 6, 6, 6, 6));
                  case "t2":
                      return new DiffJob.TaskStatus(BigInteger.valueOf(5L), RangeStats.withValues(5, 5, 5, 5, 5, 5, 5, 5, 5));
                  case "t3":
                      return DiffJob.TaskStatus.EMPTY;
                  default:
                      throw new AssertionError();
              }
          };
  
          Map<KeyspaceTablePair, DiffJob.TaskStatus> filtered = Differ.filterTables(tables, split, journal, false);
          assertEquals(2, filtered.keySet().size());
          assertEquals(RangeStats.withValues(5, 5, 5, 5, 5, 5, 5, 5, 5), filtered.get(ksTbl("t2")).stats);
          assertEquals(BigInteger.valueOf(5L), filtered.get(ksTbl("t2")).lastToken);
          assertEquals(RangeStats.newStats(), filtered.get(ksTbl("t3")).stats);
          assertNull(filtered.get(ksTbl("t3")).lastToken);
  
          // if re-running (part of) a job because of failures or problematic partitions, we want to
          // ignore the status of completed tasks and re-run them anyway as only specified tokens will
          // be processed - so t1 should be included now
          filtered = Differ.filterTables(tables, split, journal, true);
          assertEquals(3, filtered.keySet().size());
          assertEquals(RangeStats.withValues(6, 6, 6, 6, 6, 6, 6, 6, 6), filtered.get(ksTbl("t1")).stats);
          assertEquals(split.end, filtered.get(ksTbl("t1")).lastToken);
          assertEquals(RangeStats.withValues(5, 5, 5, 5, 5, 5, 5, 5, 5), filtered.get(ksTbl("t2")).stats);
          assertEquals(BigInteger.valueOf(5L), filtered.get(ksTbl("t2")).lastToken);
          assertEquals(RangeStats.newStats(), filtered.get(ksTbl("t3")).stats);
          assertNull(filtered.get(ksTbl("t3")).lastToken);
      }
test_case.class_method_signature: DifferTest.filterTaskStatusForTables()
test_case.full_signature: @Test public void filterTaskStatusForTables()
test_case.identifier: filterTaskStatusForTables
test_case.invocations: newArrayList ksTbl ksTbl ksTbl withValues valueOf withValues filterTables assertEquals size keySet assertEquals withValues get ksTbl assertEquals valueOf get ksTbl assertEquals newStats get ksTbl assertNull get ksTbl filterTables assertEquals size keySet assertEquals withValues get ksTbl assertEquals get ksTbl assertEquals withValues get ksTbl assertEquals valueOf get ksTbl assertEquals newStats get ksTbl assertNull get ksTbl
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void filterTaskStatusForTables()
focal_class.declarator: logger = LoggerFactory.getLogger(Differ.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger logger = LoggerFactory.getLogger(Differ.class);
focal_class.type: Logger
focal_class.var_name: logger
focal_class.declarator: metrics = new MetricRegistry()
focal_class.modifier: private static final
focal_class.original_string: private static final MetricRegistry metrics = new MetricRegistry();
focal_class.type: MetricRegistry
focal_class.var_name: metrics
focal_class.declarator: COMPARISON_THREADS = 8
focal_class.modifier: private static final
focal_class.original_string: private static final int COMPARISON_THREADS = 8;
focal_class.type: int
focal_class.var_name: COMPARISON_THREADS
focal_class.declarator: COMPARISON_EXECUTOR = ComparisonExecutor.newExecutor(COMPARISON_THREADS, metrics)
focal_class.modifier: private static final
focal_class.original_string: private static final ComparisonExecutor COMPARISON_EXECUTOR = ComparisonExecutor.newExecutor(COMPARISON_THREADS, metrics);
focal_class.type: ComparisonExecutor
focal_class.var_name: COMPARISON_EXECUTOR
focal_class.declarator: jobId
focal_class.modifier: private final
focal_class.original_string: private final UUID jobId;
focal_class.type: UUID
focal_class.var_name: jobId
focal_class.declarator: split
focal_class.modifier: private final
focal_class.original_string: private final DiffJob.Split split;
focal_class.type: DiffJob.Split
focal_class.var_name: split
focal_class.declarator: tokenHelper
focal_class.modifier: private final
focal_class.original_string: private final TokenHelper tokenHelper;
focal_class.type: TokenHelper
focal_class.var_name: tokenHelper
focal_class.declarator: keyspaceTables
focal_class.modifier: private final
focal_class.original_string: private final List<KeyspaceTablePair> keyspaceTables;
focal_class.type: List<KeyspaceTablePair>
focal_class.var_name: keyspaceTables
focal_class.declarator: rateLimiter
focal_class.modifier: private final
focal_class.original_string: private final RateLimiter rateLimiter;
focal_class.type: RateLimiter
focal_class.var_name: rateLimiter
focal_class.declarator: trackerProvider
focal_class.modifier: private final
focal_class.original_string: private final DiffJob.TrackerProvider trackerProvider;
focal_class.type: DiffJob.TrackerProvider
focal_class.var_name: trackerProvider
focal_class.declarator: reverseReadProbability
focal_class.modifier: private final
focal_class.original_string: private final double reverseReadProbability;
focal_class.type: double
focal_class.var_name: reverseReadProbability
focal_class.declarator: specificTokens
focal_class.modifier: private final
focal_class.original_string: private final SpecificTokens specificTokens;
focal_class.type: SpecificTokens
focal_class.var_name: specificTokens
focal_class.declarator: retryStrategyProvider
focal_class.modifier: private final
focal_class.original_string: private final RetryStrategyProvider retryStrategyProvider;
focal_class.type: RetryStrategyProvider
focal_class.var_name: retryStrategyProvider
focal_class.declarator: srcDiffCluster
focal_class.modifier: private static
focal_class.original_string: private static DiffCluster srcDiffCluster;
focal_class.type: DiffCluster
focal_class.var_name: srcDiffCluster
focal_class.declarator: targetDiffCluster
focal_class.modifier: private static
focal_class.original_string: private static DiffCluster targetDiffCluster;
focal_class.type: DiffCluster
focal_class.var_name: targetDiffCluster
focal_class.declarator: journalSession
focal_class.modifier: private static
focal_class.original_string: private static Session journalSession;
focal_class.type: Session
focal_class.var_name: journalSession
focal_class.file: spark-job/src/main/java/org/apache/cassandra/diff/Differ.java
focal_class.identifier: Differ
focal_class.interfaces: implements Serializable
focal_class.class_method_signature: Differ.Differ(JobConfiguration config,
                    DiffJob.Params params,
                    int perExecutorRateLimit,
                    DiffJob.Split split,
                    TokenHelper tokenHelper,
                    ClusterProvider sourceProvider,
                    ClusterProvider targetProvider,
                    ClusterProvider metadataProvider,
                    DiffJob.TrackerProvider trackerProvider)
focal_class.full_signature: public  Differ(JobConfiguration config,
                    DiffJob.Params params,
                    int perExecutorRateLimit,
                    DiffJob.Split split,
                    TokenHelper tokenHelper,
                    ClusterProvider sourceProvider,
                    ClusterProvider targetProvider,
                    ClusterProvider metadataProvider,
                    DiffJob.TrackerProvider trackerProvider)
focal_class.identifier: Differ
focal_class.modifiers: public
focal_class.parameters: (JobConfiguration config,
                    DiffJob.Params params,
                    int perExecutorRateLimit,
                    DiffJob.Split split,
                    TokenHelper tokenHelper,
                    ClusterProvider sourceProvider,
                    ClusterProvider targetProvider,
                    ClusterProvider metadataProvider,
                    DiffJob.TrackerProvider trackerProvider)
focal_class.signature:  Differ(JobConfiguration config,
                    DiffJob.Params params,
                    int perExecutorRateLimit,
                    DiffJob.Split split,
                    TokenHelper tokenHelper,
                    ClusterProvider sourceProvider,
                    ClusterProvider targetProvider,
                    ClusterProvider metadataProvider,
                    DiffJob.TrackerProvider trackerProvider)
focal_class.class_method_signature: Differ.run()
focal_class.full_signature: public Map<KeyspaceTablePair, RangeStats> run()
focal_class.identifier: run
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Map<KeyspaceTablePair, RangeStats>
focal_class.signature: Map<KeyspaceTablePair, RangeStats> run()
focal_class.class_method_signature: Differ.diffTable(final DiffContext context,
                                  final BiConsumer<Throwable, BigInteger> partitionErrorReporter,
                                  final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                  final BiConsumer<RangeStats, BigInteger> journal)
focal_class.full_signature: public RangeStats diffTable(final DiffContext context,
                                  final BiConsumer<Throwable, BigInteger> partitionErrorReporter,
                                  final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                  final BiConsumer<RangeStats, BigInteger> journal)
focal_class.identifier: diffTable
focal_class.modifiers: public
focal_class.parameters: (final DiffContext context,
                                  final BiConsumer<Throwable, BigInteger> partitionErrorReporter,
                                  final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                  final BiConsumer<RangeStats, BigInteger> journal)
focal_class.return: RangeStats
focal_class.signature: RangeStats diffTable(final DiffContext context,
                                  final BiConsumer<Throwable, BigInteger> partitionErrorReporter,
                                  final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                  final BiConsumer<RangeStats, BigInteger> journal)
focal_class.class_method_signature: Differ.fetchRows(DiffContext context, PartitionKey key, boolean shouldReverse, DiffCluster.Type type)
focal_class.full_signature: private Iterator<Row> fetchRows(DiffContext context, PartitionKey key, boolean shouldReverse, DiffCluster.Type type)
focal_class.identifier: fetchRows
focal_class.modifiers: private
focal_class.parameters: (DiffContext context, PartitionKey key, boolean shouldReverse, DiffCluster.Type type)
focal_class.return: Iterator<Row>
focal_class.signature: Iterator<Row> fetchRows(DiffContext context, PartitionKey key, boolean shouldReverse, DiffCluster.Type type)
focal_class.class_method_signature: Differ.filterTables(Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted)
focal_class.full_signature: @VisibleForTesting static Map<KeyspaceTablePair, DiffJob.TaskStatus> filterTables(Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted)
focal_class.identifier: filterTables
focal_class.modifiers: @VisibleForTesting static
focal_class.parameters: (Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted)
focal_class.return: Map<KeyspaceTablePair, DiffJob.TaskStatus>
focal_class.signature: Map<KeyspaceTablePair, DiffJob.TaskStatus> filterTables(Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted)
focal_class.class_method_signature: Differ.validateTableSpecs(TableSpec source, TableSpec target)
focal_class.full_signature: static void validateTableSpecs(TableSpec source, TableSpec target)
focal_class.identifier: validateTableSpecs
focal_class.modifiers: static
focal_class.parameters: (TableSpec source, TableSpec target)
focal_class.return: void
focal_class.signature: void validateTableSpecs(TableSpec source, TableSpec target)
focal_class.class_method_signature: Differ.validateRange(BigInteger start, BigInteger end, TokenHelper tokens)
focal_class.full_signature: @VisibleForTesting static void validateRange(BigInteger start, BigInteger end, TokenHelper tokens)
focal_class.identifier: validateRange
focal_class.modifiers: @VisibleForTesting static
focal_class.parameters: (BigInteger start, BigInteger end, TokenHelper tokens)
focal_class.return: void
focal_class.signature: void validateRange(BigInteger start, BigInteger end, TokenHelper tokens)
focal_class.class_method_signature: Differ.accumulate(Map<KeyspaceTablePair, RangeStats> stats, Map<KeyspaceTablePair, RangeStats> otherStats)
focal_class.full_signature: @VisibleForTesting static Map<KeyspaceTablePair, RangeStats> accumulate(Map<KeyspaceTablePair, RangeStats> stats, Map<KeyspaceTablePair, RangeStats> otherStats)
focal_class.identifier: accumulate
focal_class.modifiers: @VisibleForTesting static
focal_class.parameters: (Map<KeyspaceTablePair, RangeStats> stats, Map<KeyspaceTablePair, RangeStats> otherStats)
focal_class.return: Map<KeyspaceTablePair, RangeStats>
focal_class.signature: Map<KeyspaceTablePair, RangeStats> accumulate(Map<KeyspaceTablePair, RangeStats> stats, Map<KeyspaceTablePair, RangeStats> otherStats)
focal_class.class_method_signature: Differ.updateMetrics(String prefix, Map<KeyspaceTablePair, RangeStats> statsMap)
focal_class.full_signature: private static void updateMetrics(String prefix, Map<KeyspaceTablePair, RangeStats> statsMap)
focal_class.identifier: updateMetrics
focal_class.modifiers: private static
focal_class.parameters: (String prefix, Map<KeyspaceTablePair, RangeStats> statsMap)
focal_class.return: void
focal_class.signature: void updateMetrics(String prefix, Map<KeyspaceTablePair, RangeStats> statsMap)
focal_class.class_method_signature: Differ.shutdown()
focal_class.full_signature: public static void shutdown()
focal_class.identifier: shutdown
focal_class.modifiers: public static
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void shutdown()
focal_method.body: @VisibleForTesting
      static Map<KeyspaceTablePair, DiffJob.TaskStatus> filterTables(Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted) {
          Map<KeyspaceTablePair, DiffJob.TaskStatus> tablesToProcess = new HashMap<>();
          for (KeyspaceTablePair pair : keyspaceTables) {
              DiffJob.TaskStatus taskStatus = journal.apply(pair);
              RangeStats diffStats = taskStatus.stats;
              BigInteger lastToken = taskStatus.lastToken;
  
              // When we finish processing a split for a given table, we update the task status in journal
              // to set the last seen token to the split's end token, to indicate that the split is complete.
              if (!includeCompleted && lastToken != null && lastToken.equals(split.end)) {
                  logger.info("Found finished table {} for split {}", pair, split);
              }
              else {
                  tablesToProcess.put(pair, diffStats != null
                                            ? taskStatus
                                            : new DiffJob.TaskStatus(taskStatus.lastToken, RangeStats.newStats()));
              }
          }
          return tablesToProcess;
      }
focal_method.class_method_signature: Differ.filterTables(Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted)
focal_method.full_signature: @VisibleForTesting static Map<KeyspaceTablePair, DiffJob.TaskStatus> filterTables(Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted)
focal_method.identifier: filterTables
focal_method.invocations: apply equals info put newStats
focal_method.modifiers: @VisibleForTesting static
focal_method.parameters: (Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted)
focal_method.return: Map<KeyspaceTablePair, DiffJob.TaskStatus>
focal_method.signature: Map<KeyspaceTablePair, DiffJob.TaskStatus> filterTables(Iterable<KeyspaceTablePair> keyspaceTables,
                                                                     DiffJob.Split split,
                                                                     Function<KeyspaceTablePair, DiffJob.TaskStatus> journal,
                                                                     boolean includeCompleted)