@SuppressWarnings({"rawtypes", "unchecked"})
  protected static Iterator<RecordInfo> resolveComparisonTies(Iterator<RecordInfo> recordInfoIterator,
      HashFunction hashFunction) {
    Map<Object, RecordInfo> deDuplicatedRecordInfo = new HashMap<>();
    while (recordInfoIterator.hasNext()) {
      RecordInfo recordInfo = recordInfoIterator.next();
      Comparable newComparisonValue = recordInfo.getComparisonValue();
      deDuplicatedRecordInfo.compute(HashUtils.hashPrimaryKey(recordInfo.getPrimaryKey(), hashFunction),
          (key, maxComparisonValueRecordInfo) -> {
            if (maxComparisonValueRecordInfo == null) {
              return recordInfo;
            }
            int comparisonResult = newComparisonValue.compareTo(maxComparisonValueRecordInfo.getComparisonValue());
            if (comparisonResult >= 0) {
              return recordInfo;
            }
            return maxComparisonValueRecordInfo;
          });
    }
    return deDuplicatedRecordInfo.values().iterator();
  }
@Test
  public void testResolveComparisonTies() {
    // Build a record info list for testing
    int[] primaryKeys = new int[]{0, 1, 2, 0, 1, 0};
    int[] timestamps = new int[]{0, 0, 0, 0, 0, 0};
    int numRecords = primaryKeys.length;
    List<RecordInfo> recordInfoList = new ArrayList<>();
    for (int docId = 0; docId < numRecords; docId++) {
      recordInfoList.add(new RecordInfo(makePrimaryKey(primaryKeys[docId]), docId, timestamps[docId], false));
    }
    // Resolve comparison ties
    Iterator<RecordInfo> deDuplicatedRecords =
        BasePartitionUpsertMetadataManager.resolveComparisonTies(recordInfoList.iterator(), HashFunction.NONE);
    // Ensure we have only 1 record for each unique primary key
    Map<PrimaryKey, RecordInfo> recordsByPrimaryKeys = new HashMap<>();
    while (deDuplicatedRecords.hasNext()) {
      RecordInfo recordInfo = deDuplicatedRecords.next();
      assertFalse(recordsByPrimaryKeys.containsKey(recordInfo.getPrimaryKey()));
      recordsByPrimaryKeys.put(recordInfo.getPrimaryKey(), recordInfo);
    }
    assertEquals(recordsByPrimaryKeys.size(), 3);
    // Ensure that to resolve ties, we pick the last docId
    assertEquals(recordsByPrimaryKeys.get(makePrimaryKey(0)).getDocId(), 5);
    assertEquals(recordsByPrimaryKeys.get(makePrimaryKey(1)).getDocId(), 4);
    assertEquals(recordsByPrimaryKeys.get(makePrimaryKey(2)).getDocId(), 2);
  }