class 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;

            // 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;
    }

    public  Differ(JobConfiguration config,
                  DiffJob.Params params,
                  int perExecutorRateLimit,
                  DiffJob.Split split,
                  TokenHelper tokenHelper,
                  ClusterProvider sourceProvider,
                  ClusterProvider targetProvider,
                  ClusterProvider metadataProvider,
                  DiffJob.TrackerProvider trackerProvider);

}