@Test
  public void testListNamedBlobsWithPrefixWithMultipleVersions() throws Exception {
    ListOperationTestParam param = setupTestParamForListOperation(10);
    Account account = param.account;
    Container container = param.container;

    // There are 10 different blob names and they all have a valid base version. After base version, we will
    // 1. Add a new version with deleted_ts = -1, state = ready to first record.
    //    New version is expected in the list operation
    // 2. Add a new version with deleted_ts = now() + 1 day, state = ready to second record
    //    New version is expected in the list operation
    // 3. Add a new version with deleted_ts = -1, state = wip to third record
    //    Base version is expected in the list operation
    // 4. Add a new version with deleted_ts = now() + 1 day, state = wip to fourth record
    //    Base version is expected in the list operation
    // 5. Add a new version with deleted_ts = now() - 1 day, state = ready to fifth record
    //    Blob name is deleted, not expected from list operation
    // 6. Add a new version with deleted_ts = now() - 1 day, state = wip to sixth record
    //    Base version is expected in the list operation
    // 7. Add two new versions to seventh record, one with deleted_ts = -1, state = ready and another with deleted_ts = -1 and state = wip
    //    First new version is expected in the list operation
    // 8. Add two new versions to eighth record, both with deleted_ts = -1, state = wip
    //    Base version is expected in the list operation
    // 9. Delete ninth record
    //    Blob name is deleted, not expected from list operation
    // 10. Add a new version with deleted_ts = -1 and state = ready and then delete this blob mame
    //     Blob name is deleted, not expected from list operation

    List<NamedBlobRecord> expectedRecords = new ArrayList<>();
    List<NamedBlobRecord> removedRecords = new ArrayList<>();
    // we need to sleep for some time and set the current time to mock time instance before put and delete so the timestamp
    // these operations saved to database is the same as the real time.

    Thread.sleep(10);
    time.setCurrentMilliseconds(SystemTime.getInstance().milliseconds());

    // Case 1
    NamedBlobRecord record = param.records.get(0);
    NamedBlobRecord newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            Utils.Infinite_Time, 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.READY, true).get();
    expectedRecords.add(newRecord);

    // Case 2
    record = param.records.get(1);
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            time.milliseconds() + TimeUnit.DAYS.toMillis(1), 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.READY, true).get();
    expectedRecords.add(newRecord);

    // Case 3
    record = param.records.get(2);
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            Utils.Infinite_Time, 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.IN_PROGRESS, true).get();
    expectedRecords.add(record);

    // Case 4
    record = param.records.get(3);
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            time.milliseconds() + TimeUnit.DAYS.toMillis(1), 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.IN_PROGRESS, true).get();
    expectedRecords.add(record);

    // Case 5
    record = param.records.get(4);
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            time.milliseconds() - TimeUnit.DAYS.toMillis(1), 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.READY, true).get();
    removedRecords.add(record);

    // Case 6
    record = param.records.get(5);
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            time.milliseconds() - TimeUnit.DAYS.toMillis(1), 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.IN_PROGRESS, true).get();
    expectedRecords.add(record);

    // Case 7
    record = param.records.get(6);
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            Utils.Infinite_Time, 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.READY, true).get();
    Thread.sleep(10);
    time.setCurrentMilliseconds(SystemTime.getInstance().milliseconds());
    NamedBlobRecord secondNewRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            Utils.Infinite_Time, 0, 1024);
    namedBlobDb.put(secondNewRecord, NamedBlobState.IN_PROGRESS, true).get();
    expectedRecords.add(newRecord);

    // Case 8
    record = param.records.get(7);
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            Utils.Infinite_Time, 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.IN_PROGRESS, true).get();
    Thread.sleep(10);
    time.setCurrentMilliseconds(SystemTime.getInstance().milliseconds());
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            Utils.Infinite_Time, 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.IN_PROGRESS, true).get();
    expectedRecords.add(record);

    // Case 9
    Thread.sleep(10);
    time.setCurrentMilliseconds(SystemTime.getInstance().milliseconds());
    record = param.records.get(8);
    namedBlobDb.delete(account.getName(), container.getName(), record.getBlobName()).get();
    removedRecords.add(record);

    // Case 10
    record = param.records.get(9);
    newRecord =
        new NamedBlobRecord(account.getName(), container.getName(), record.getBlobName(), getBlobId(account, container),
            Utils.Infinite_Time, 0, 1024);
    namedBlobDb.put(newRecord, NamedBlobState.READY, true).get();
    Thread.sleep(10);
    time.setCurrentMilliseconds(SystemTime.getInstance().milliseconds());
    namedBlobDb.delete(account.getName(), container.getName(), record.getBlobName()).get();
    removedRecords.add(record);

    Page<NamedBlobRecord> page =
        namedBlobDb.list(account.getName(), container.getName(), param.blobNamePrefix, null, null).get();
    assertNull(page.getNextPageToken());
    validateListResult(page.getEntries(), expectedRecords);

    for (NamedBlobRecord expectedRecord : expectedRecords) {
      NamedBlobRecord recordFromStore =
          namedBlobDb.get(account.getName(), container.getName(), expectedRecord.getBlobName()).get();
      assertEquals("List result does not match with get for blob_name " + expectedRecord.getBlobName(), expectedRecord,
          recordFromStore);
    }
  }
Test case for list named blobs with prefix, with multiple versions on blob names
@throws Exception
ambry-named-mysql/src/integration-test/java/com/github/ambry/named/MySqlNamedBlobDbListOperationIntegrationTest.java