@Test
  public void testFilter() throws Exception {
    // case 1. all log segments have 100% valid data
    blobStore.logSegmentsNotInJournal = CompactionPolicyTest.generateRandomLogSegmentName(5);
    MockBlobStoreStats blobStoreStats = blobStore.getBlobStoreStats();
    NavigableMap<LogSegmentName, Long> validDataSize = new TreeMap<>();
    Map<LogSegmentName, Long> allDataSize = new HashMap<>();
    long maxDataSize = blobStore.segmentCapacity - blobStore.segmentHeaderSize;
    for (LogSegmentName segmentName : blobStore.logSegmentsNotInJournal) {
      validDataSize.put(segmentName, maxDataSize);
      allDataSize.put(segmentName, maxDataSize);
    }
    blobStoreStats.validDataSizeByLogSegments = validDataSize;
    blobStoreStats.setLogSegmentSizeProvider(new MapLogSegmentSizeProvider(allDataSize));
    CompactionPolicyTest.verifyCompactionDetails(null, blobStore, compactionPolicy);

    // case 2, only leading segments have 100% valid data
    List<LogSegmentName> logSegmentToCompact = new ArrayList<>();
    for (int i = 0; i < blobStore.logSegmentsNotInJournal.size(); i++) {
      LogSegmentName segmentName = blobStore.logSegmentsNotInJournal.get(i);
      if (i < 2) {
        validDataSize.put(segmentName, allDataSize.get(segmentName));
      } else {
        validDataSize.put(segmentName, allDataSize.get(segmentName) / 2);
        logSegmentToCompact.add(segmentName);
      }
    }
    CompactionPolicyTest.verifyCompactionDetails(
        new CompactionDetails(time.milliseconds() - messageRetentionTimeInMs, logSegmentToCompact, null), blobStore,
        compactionPolicy);

    // case 3. only trailing segments have 100% valid data
    logSegmentToCompact = new ArrayList<>();
    for (int i = 0; i < blobStore.logSegmentsNotInJournal.size(); i++) {
      LogSegmentName segmentName = blobStore.logSegmentsNotInJournal.get(i);
      if (i > 2) {
        validDataSize.put(segmentName, allDataSize.get(segmentName));
      } else {
        validDataSize.put(segmentName, allDataSize.get(segmentName) / 2);
        logSegmentToCompact.add(segmentName);
      }
    }
    CompactionPolicyTest.verifyCompactionDetails(
        new CompactionDetails(time.milliseconds() - messageRetentionTimeInMs, logSegmentToCompact, null), blobStore,
        compactionPolicy);

    // case 4: leading and trailing segments both have 100% valid data
    logSegmentToCompact = new ArrayList<>();
    for (int i = 0; i < blobStore.logSegmentsNotInJournal.size(); i++) {
      LogSegmentName segmentName = blobStore.logSegmentsNotInJournal.get(i);
      if (i == 2) {
        validDataSize.put(segmentName, allDataSize.get(segmentName) / 2);
        logSegmentToCompact.add(segmentName);
      } else {
        validDataSize.put(segmentName, allDataSize.get(segmentName));
      }
    }
    CompactionPolicyTest.verifyCompactionDetails(
        new CompactionDetails(time.milliseconds() - messageRetentionTimeInMs, logSegmentToCompact, null), blobStore,
        compactionPolicy);

    // case 5: none of the log segment has 100% valid data
    logSegmentToCompact = new ArrayList<>();
    for (int i = 0; i < blobStore.logSegmentsNotInJournal.size(); i++) {
      LogSegmentName segmentName = blobStore.logSegmentsNotInJournal.get(i);
      validDataSize.put(segmentName, allDataSize.get(segmentName) / 2);
      logSegmentToCompact.add(segmentName);
    }
    CompactionPolicyTest.verifyCompactionDetails(
        new CompactionDetails(time.milliseconds() - messageRetentionTimeInMs, logSegmentToCompact, null), blobStore,
        compactionPolicy);
  }
Test compact all policy when there are log segments who has 100% valid data.
@throws Exception
ambry-store/src/test/java/com/github/ambry/store/CompactAllPolicyTest.java