public static Iterator<Row> removeNetCarryovers(Iterator<Row> rowIterator, StructType rowType) {
    ChangelogIterator changelogIterator = new RemoveNetCarryoverIterator(rowIterator, rowType);
    return Iterators.filter(changelogIterator, Objects::nonNull);
  }
@Test
  public void testRemoveNetCarryovers() {
    List<Row> rowsWithDuplication =
        Lists.newArrayList(
            // this row are different from other rows, it is a net change, should be kept
            new GenericRowWithSchema(new Object[] {0, "d", "data", DELETE, 0, 0}, null),
            // a pair of delete and insert rows, should be removed
            new GenericRowWithSchema(new Object[] {1, "d", "data", DELETE, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {1, "d", "data", INSERT, 0, 0}, null),
            // 2 delete rows and 2 insert rows, should be removed
            new GenericRowWithSchema(new Object[] {1, "d", "data", DELETE, 1, 1}, null),
            new GenericRowWithSchema(new Object[] {1, "d", "data", DELETE, 1, 1}, null),
            new GenericRowWithSchema(new Object[] {1, "d", "data", INSERT, 1, 1}, null),
            new GenericRowWithSchema(new Object[] {1, "d", "data", INSERT, 1, 1}, null),
            // a pair of insert and delete rows across snapshots, should be removed
            new GenericRowWithSchema(new Object[] {1, "d", "data", INSERT, 2, 2}, null),
            new GenericRowWithSchema(new Object[] {1, "d", "data", DELETE, 3, 3}, null),
            // extra insert rows, they are net changes, should be kept
            new GenericRowWithSchema(new Object[] {1, "d", "data", INSERT, 4, 4}, null),
            new GenericRowWithSchema(new Object[] {1, "d", "data", INSERT, 4, 4}, null),
            // different key, net changes, should be kept
            new GenericRowWithSchema(new Object[] {2, "d", "data", DELETE, 4, 4}, null));

    List<Object[]> expectedRows =
        Lists.newArrayList(
            new Object[] {0, "d", "data", DELETE, 0, 0},
            new Object[] {1, "d", "data", INSERT, 4, 4},
            new Object[] {1, "d", "data", INSERT, 4, 4},
            new Object[] {2, "d", "data", DELETE, 4, 4});

    Iterator<Row> iterator =
        ChangelogIterator.removeNetCarryovers(rowsWithDuplication.iterator(), SCHEMA);
    List<Row> result = Lists.newArrayList(iterator);

    assertEquals("Rows should match.", expectedRows, rowsToJava(result));
  }