@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);
}