Differ implements Serializable { @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; 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; } Differ(JobConfiguration config,
                  DiffJob.Params params,
                  int perExecutorRateLimit,
                  DiffJob.Split split,
                  TokenHelper tokenHelper,
                  ClusterProvider sourceProvider,
                  ClusterProvider targetProvider,
                  ClusterProvider metadataProvider,
                  DiffJob.TrackerProvider trackerProvider); Map<KeyspaceTablePair, RangeStats> run(); RangeStats diffTable(final DiffContext context,
                                final BiConsumer<Throwable, BigInteger> partitionErrorReporter,
                                final BiConsumer<MismatchType, BigInteger> mismatchReporter,
                                final BiConsumer<RangeStats, BigInteger> journal); static void shutdown();  }
@Test public void filterTaskStatusForTables() { 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); 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); }