@VisibleForTesting
  public static SegmentSelectionResult processValidDocIdsMetadata(Map<String, String> taskConfigs,
      Map<String, SegmentZKMetadata> completedSegmentsMap,
      Map<String, List<ValidDocIdsMetadataInfo>> validDocIdsMetadataInfoMap) {
    double invalidRecordsThresholdPercent = Double.parseDouble(
        taskConfigs.getOrDefault(UpsertCompactionTask.INVALID_RECORDS_THRESHOLD_PERCENT,
            String.valueOf(DEFAULT_INVALID_RECORDS_THRESHOLD_PERCENT)));
    long invalidRecordsThresholdCount = Long.parseLong(
        taskConfigs.getOrDefault(UpsertCompactionTask.INVALID_RECORDS_THRESHOLD_COUNT,
            String.valueOf(DEFAULT_INVALID_RECORDS_THRESHOLD_COUNT)));
    List<Pair<SegmentZKMetadata, Long>> segmentsForCompaction = new ArrayList<>();
    List<String> segmentsForDeletion = new ArrayList<>();
    for (String segmentName : validDocIdsMetadataInfoMap.keySet()) {
      // check if segment is part of completed segments
      if (!completedSegmentsMap.containsKey(segmentName)) {
        LOGGER.warn("Segment {} is not found in the completed segments list, skipping it for compaction", segmentName);
        continue;
      }
      SegmentZKMetadata segment = completedSegmentsMap.get(segmentName);
      for (ValidDocIdsMetadataInfo validDocIdsMetadata : validDocIdsMetadataInfoMap.get(segmentName)) {
        long totalInvalidDocs = validDocIdsMetadata.getTotalInvalidDocs();

        // Skip segments if the crc from zk metadata and server does not match. They may be being reloaded.
        if (segment.getCrc() != Long.parseLong(validDocIdsMetadata.getSegmentCrc())) {
          LOGGER.warn("CRC mismatch for segment: {}, (segmentZKMetadata={}, validDocIdsMetadata={})", segmentName,
              segment.getCrc(), validDocIdsMetadata.getSegmentCrc());
          continue;
        }
        long totalDocs = validDocIdsMetadata.getTotalDocs();
        double invalidRecordPercent = ((double) totalInvalidDocs / totalDocs) * 100;
        if (totalInvalidDocs == totalDocs) {
          segmentsForDeletion.add(segment.getSegmentName());
        } else if (invalidRecordPercent >= invalidRecordsThresholdPercent
            && totalInvalidDocs >= invalidRecordsThresholdCount) {
          segmentsForCompaction.add(Pair.of(segment, totalInvalidDocs));
        }
        break;
      }
    }
    segmentsForCompaction.sort((o1, o2) -> {
      if (o1.getValue() > o2.getValue()) {
        return -1;
      } else if (o1.getValue().equals(o2.getValue())) {
        return 0;
      }
      return 1;
    });

    return new SegmentSelectionResult(
        segmentsForCompaction.stream().map(Map.Entry::getKey).collect(Collectors.toList()), segmentsForDeletion);
  }
@Test
  public void testProcessValidDocIdsMetadata()
      throws IOException {
    Map<String, String> compactionConfigs = getCompactionConfigs("1", "10");
    String json = "{\"testTable__0\": [{\"totalValidDocs\": 50, \"totalInvalidDocs\": 50, "
        + "\"segmentName\": \"testTable__0\", \"totalDocs\": 100, \"segmentCrc\": \"1000\"}], "
        + "\"testTable__1\": [{\"totalValidDocs\": 0, "
        + "\"totalInvalidDocs\": 10, \"segmentName\": \"testTable__1\", \"totalDocs\": 10, \"segmentCrc\": \"2000\"}]}";

    Map<String, List<ValidDocIdsMetadataInfo>> validDocIdsMetadataInfo =
        JsonUtils.stringToObject(json, new TypeReference<>() {
        });

    // no completed segments scenario, there shouldn't be any segment selected for compaction
    UpsertCompactionTaskGenerator.SegmentSelectionResult segmentSelectionResult =
        UpsertCompactionTaskGenerator.processValidDocIdsMetadata(compactionConfigs, new HashMap<>(),
            validDocIdsMetadataInfo);
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().size(), 0);

    // test with valid crc and thresholds
    segmentSelectionResult =
        UpsertCompactionTaskGenerator.processValidDocIdsMetadata(compactionConfigs, _completedSegmentsMap,
            validDocIdsMetadataInfo);
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().size(), 1);
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().size(), 1);
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().get(0).getSegmentName(),
        _completedSegment.getSegmentName());
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().get(0), _completedSegment2.getSegmentName());

    // test with a higher invalidRecordsThresholdPercent
    compactionConfigs = getCompactionConfigs("60", "10");
    segmentSelectionResult =
        UpsertCompactionTaskGenerator.processValidDocIdsMetadata(compactionConfigs, _completedSegmentsMap,
            validDocIdsMetadataInfo);
    assertTrue(segmentSelectionResult.getSegmentsForCompaction().isEmpty());
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().size(), 1);
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().get(0), _completedSegment2.getSegmentName());

    // test without an invalidRecordsThresholdPercent
    compactionConfigs = getCompactionConfigs("0", "10");
    segmentSelectionResult =
        UpsertCompactionTaskGenerator.processValidDocIdsMetadata(compactionConfigs, _completedSegmentsMap,
            validDocIdsMetadataInfo);
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().size(), 1);
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().size(), 1);
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().get(0).getSegmentName(),
        _completedSegment.getSegmentName());
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().get(0), _completedSegment2.getSegmentName());

    // test without a invalidRecordsThresholdCount
    compactionConfigs = getCompactionConfigs("30", "0");
    segmentSelectionResult =
        UpsertCompactionTaskGenerator.processValidDocIdsMetadata(compactionConfigs, _completedSegmentsMap,
            validDocIdsMetadataInfo);
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().size(), 1);
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().size(), 1);
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().get(0).getSegmentName(),
        _completedSegment.getSegmentName());
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().get(0), _completedSegment2.getSegmentName());

    // Test the case where the completedSegment from api has different crc than segment from zk metadata.
    json = "{\"" + _completedSegment.getSegmentName() + "\": [{\"totalValidDocs\": 50, \"totalInvalidDocs\": 50, "
        + "\"segmentName\": \"" + _completedSegment.getSegmentName() + "\", \"totalDocs\": 100, \"segmentCrc\": "
        + "\"1234567890\"}], \"" + _completedSegment2.getSegmentName() + "\": [{\"totalValidDocs\": 0, "
        + "\"totalInvalidDocs\": 10, \"segmentName\": \"" + _completedSegment2.getSegmentName() + "\", "
        + "\"segmentCrc\": \"" + _completedSegment2.getCrc() + "\", \"totalDocs\": 10}]}";
    validDocIdsMetadataInfo = JsonUtils.stringToObject(json, new TypeReference<>() {
    });
    segmentSelectionResult =
        UpsertCompactionTaskGenerator.processValidDocIdsMetadata(compactionConfigs, _completedSegmentsMap,
            validDocIdsMetadataInfo);

    // completedSegment is supposed to be filtered out
    Assert.assertEquals(segmentSelectionResult.getSegmentsForCompaction().size(), 0);

    // completedSegment2 is still supposed to be deleted
    Assert.assertEquals(segmentSelectionResult.getSegmentsForDeletion().size(), 1);
    assertEquals(segmentSelectionResult.getSegmentsForDeletion().get(0),
        _completedSegment2.getSegmentName());

    // check if both the candidates for compaction are coming in sorted descending order
    json = "{\"" + _completedSegment.getSegmentName() + "\": [{\"totalValidDocs\": 50, \"totalInvalidDocs\": 50, "
        + "\"segmentName\": \"" + _completedSegment.getSegmentName() + "\", \"totalDocs\": 100, \"segmentCrc\": \""
        + _completedSegment.getCrc() + "\"}], \"" + _completedSegment2.getSegmentName() + "\": "
        + "[{\"totalValidDocs\": 10, \"totalInvalidDocs\": 40, \"segmentName\": \""
        + _completedSegment2.getSegmentName() + "\", \"segmentCrc\": \"" + _completedSegment2.getCrc() + "\", "
        + "\"totalDocs\": 50}]}";
    validDocIdsMetadataInfo = JsonUtils.stringToObject(json, new TypeReference<>() {
    });
    compactionConfigs = getCompactionConfigs("30", "0");
    segmentSelectionResult =
        UpsertCompactionTaskGenerator.processValidDocIdsMetadata(compactionConfigs, _completedSegmentsMap,
            validDocIdsMetadataInfo);
    Assert.assertEquals(segmentSelectionResult.getSegmentsForCompaction().size(), 2);
    Assert.assertEquals(segmentSelectionResult.getSegmentsForDeletion().size(), 0);
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().get(0).getSegmentName(),
        _completedSegment.getSegmentName());
    assertEquals(segmentSelectionResult.getSegmentsForCompaction().get(1).getSegmentName(),
        _completedSegment2.getSegmentName());
  }