@Test
  public void testCleanupBlobsPipeline() throws Exception {
    Calendar calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
    calendar.add(Calendar.DATE, -config.staleDataRetentionDays);
    long staleCutoffTime = calendar.getTimeInMillis();
    long staleCutoffTimePlusOneMillisecond = staleCutoffTime + 1;

    Account account = accountService.getAllAccounts().iterator().next();
    Container container = account.getAllContainers().iterator().next();

    int staleCount = 10;
    int needCleanupCount = 0;
    List<NamedBlobRecord> staleRecords = new ArrayList<>();
    List<NamedBlobRecord> records = new ArrayList<>();

    // Create stale named blob records
    for (int i = 0; i < staleCount; i++) {
      String blobId = getBlobId(account, container);
      String blobName = "stale/" + i + "/more path segments--";
      long expirationTime = i % 2 == 0 ? Utils.Infinite_Time : staleCutoffTimePlusOneMillisecond;
      NamedBlobRecord record =
          new NamedBlobRecord(account.getName(), container.getName(), blobName, blobId, expirationTime);

      NamedBlobState blob_state =
          expirationTime == Utils.Infinite_Time ? NamedBlobState.IN_PROGRESS : NamedBlobState.READY;

      time.setCurrentMilliseconds(staleCutoffTime);
      namedBlobDb.put(record, blob_state, true).get();

      time.setCurrentMilliseconds(staleCutoffTimePlusOneMillisecond);
      String blobId2 = getBlobId(account, container);
      NamedBlobRecord record2 =
          new NamedBlobRecord(account.getName(), container.getName(), blobName, blobId2, expirationTime);
      namedBlobDb.put(record2, NamedBlobState.READY, true).get();

      if (expirationTime == Utils.Infinite_Time) {
        needCleanupCount += 1;
      }
      staleRecords.add(record);
      records.add(record2);
    }

    Thread.sleep(100);

    // Confirm the pullStaleBlobs indeed pulled out the stale blob cases
    Set<String> staleInputSet = staleRecords.stream()
        .filter((r) -> r.getExpirationTimeMs() == Utils.Infinite_Time)
        .map((r) -> String.join("|", r.getBlobName(), r.getBlobId()))
        .collect(Collectors.toSet());

    List<StaleNamedBlob> staleNamedBlobs = namedBlobDb.pullStaleBlobs().get();
    Set<String> staleResultSet = staleNamedBlobs.stream()
        .map((s) -> String.join("|", s.getBlobName(), s.getBlobId()))
        .collect(Collectors.toSet());
    assertEquals("Stale records count does not match!", needCleanupCount, staleNamedBlobs.size());
    assertEquals("Stale records pulled out does not meet expectation", staleInputSet, staleResultSet);

    // Confirm pullStaleBlobs return empty list after cleanupStaleData is called
    Integer cleanedUpStaleCount = namedBlobDb.cleanupStaleData(staleNamedBlobs).get();
    List<StaleNamedBlob> staleNamedBlobsNew = namedBlobDb.pullStaleBlobs().get();

    assertEquals("Cleaned Stale records count does not match!", needCleanupCount, cleanedUpStaleCount.intValue());
    assertTrue("Still pulled out stale blobs after cleanup!", staleNamedBlobsNew.isEmpty());

    // Verify we can still pull out the valid named blobs after cleanup
    for (NamedBlobRecord record : records) {
      if (record.getExpirationTimeMs() == Utils.Infinite_Time) {
        NamedBlobRecord recordFromDb =
            namedBlobDb.get(record.getAccountName(), record.getContainerName(), record.getBlobName()).get();
        assertEquals("After stale cleanup, the record does not match expectation.", record, recordFromDb);
      }
    }

    time.setCurrentMilliseconds(System.currentTimeMillis());
  }
Test behavior with blob cleanup main pipeline
ambry-named-mysql/src/integration-test/java/com/github/ambry/named/MySqlNamedBlobDbIntegrationTest.java