public static Iterator<Row> computeUpdates(
      Iterator<Row> rowIterator, StructType rowType, String[] identifierFields) {
    Iterator<Row> carryoverRemoveIterator = removeCarryovers(rowIterator, rowType);
    ChangelogIterator changelogIterator =
        new ComputeUpdateIterator(carryoverRemoveIterator, rowType, identifierFields);
    return Iterators.filter(changelogIterator, Objects::nonNull);
  }
@Test
  public void testUpdatedRowsWithDuplication() {
    List<Row> rowsWithDuplication =
        Lists.newArrayList(
            // two rows with same identifier fields(id, name)
            new GenericRowWithSchema(new Object[] {1, "a", "data", DELETE, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {1, "a", "data", DELETE, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {1, "a", "new_data", INSERT, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {1, "a", "new_data", INSERT, 0, 0}, null));

    Iterator<Row> iterator =
        ChangelogIterator.computeUpdates(rowsWithDuplication.iterator(), SCHEMA, IDENTIFIER_FIELDS);

    assertThrows(
        "Cannot compute updates because there are multiple rows with the same identifier fields([id, name]). Please make sure the rows are unique.",
        IllegalStateException.class,
        () -> Lists.newArrayList(iterator));

    // still allow extra insert rows
    rowsWithDuplication =
        Lists.newArrayList(
            new GenericRowWithSchema(new Object[] {1, "a", "data", DELETE, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {1, "a", "new_data1", INSERT, 0, 0}, null),
            new GenericRowWithSchema(new Object[] {1, "a", "new_data2", INSERT, 0, 0}, null));

    Iterator<Row> iterator1 =
        ChangelogIterator.computeUpdates(rowsWithDuplication.iterator(), SCHEMA, IDENTIFIER_FIELDS);

    assertEquals(
        "Rows should match.",
        Lists.newArrayList(
            new Object[] {1, "a", "data", UPDATE_BEFORE, 0, 0},
            new Object[] {1, "a", "new_data1", UPDATE_AFTER, 0, 0},
            new Object[] {1, "a", "new_data2", INSERT, 0, 0}),
        rowsToJava(Lists.newArrayList(iterator1)));
  }