@Test
  public void testPutGetListDeleteSequence() throws Exception {
    int blobsPerContainer = 5;

    List<NamedBlobRecord> records = new ArrayList<>();
    for (Account account : accountService.getAllAccounts()) {
      for (Container container : account.getAllContainers()) {
        for (int i = 0; i < blobsPerContainer; i++) {
          String blobId = getBlobId(account, container);
          String blobName = "name/" + i + "/more path segments--";
          long expirationTime =
              i % 2 == 0 ? Utils.Infinite_Time : System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1);
          long blobSize = 20;
          NamedBlobRecord record =
              new NamedBlobRecord(account.getName(), container.getName(), blobName, blobId, expirationTime, 0,
                  blobSize);
          namedBlobDb.put(record).get();
          records.add(record);
        }
      }
    }

    // get records just inserted
    time.setCurrentMilliseconds(System.currentTimeMillis());
    for (NamedBlobRecord record : records) {
      NamedBlobRecord recordFromStore =
          namedBlobDb.get(record.getAccountName(), record.getContainerName(), record.getBlobName()).get();
      assertEquals("Record does not match expectations.", record, recordFromStore);
    }

    // list records in each container
    time.setCurrentMilliseconds(System.currentTimeMillis());
    for (Account account : accountService.getAllAccounts()) {
      for (Container container : account.getAllContainers()) {
        Page<NamedBlobRecord> page = namedBlobDb.list(account.getName(), container.getName(), "name", null, null).get();
        assertNull("No continuation token expected", page.getNextPageToken());
        assertEquals("Unexpected number of blobs in container", blobsPerContainer, page.getEntries().size());
      }
    }

    // list all records in each container.
    time.setCurrentMilliseconds(System.currentTimeMillis());
    for (Account account : accountService.getAllAccounts()) {
      for (Container container : account.getAllContainers()) {
        //page with no token
        Page<NamedBlobRecord> page = namedBlobDb.list(account.getName(), container.getName(), null, null, null).get();
        assertNull("No continuation token expected", page.getNextPageToken());
        assertEquals("Unexpected number of blobs in container", blobsPerContainer, page.getEntries().size());
        //page with token
        Page<NamedBlobRecord> pageWithToken =
            namedBlobDb.list(account.getName(), container.getName(), null, "name/4", null).get();
        assertEquals("Unexpected number of blobs in container", blobsPerContainer / 5,
            pageWithToken.getEntries().size());
        //page with maxKeys
        Page<NamedBlobRecord> pageWithMaxKey =
            namedBlobDb.list(account.getName(), container.getName(), null, null, 1).get();
        assertEquals("Unexpected number of blobs in container", blobsPerContainer / 5,
            pageWithMaxKey.getEntries().size());
        // Verify that blob size and modified ts is returned for all blobs
        for (NamedBlobRecord record : pageWithToken.getEntries()) {
          assertEquals("Mismatch in blob size", 20, record.getBlobSize());
          assertNotEquals("Modified timestamp must be present", 0L, record.getModifiedTimeMs());
        }
        for (NamedBlobRecord record : pageWithMaxKey.getEntries()) {
          assertEquals("Mismatch in blob size", 20, record.getBlobSize());
          assertNotEquals("Modified timestamp must be present", 0L, record.getModifiedTimeMs());
        }
      }
    }

    // check that puts to the same keys fail.
    time.setCurrentMilliseconds(System.currentTimeMillis());
    for (Account account : accountService.getAllAccounts()) {
      for (Container container : account.getAllContainers()) {
        for (int i = 0; i < blobsPerContainer; i++) {
          String blobId = getBlobId(account, container);
          String blobName = "name/" + i + "/more path segments--";
          NamedBlobRecord record =
              new NamedBlobRecord(account.getName(), container.getName(), blobName, blobId, Utils.Infinite_Time);
          checkErrorCode(() -> namedBlobDb.put(record), RestServiceErrorCode.Conflict);
        }
      }
    }

    // delete the records and check that they cannot be fetched with a get call.
    for (NamedBlobRecord record : records) {
      time.setCurrentMilliseconds(System.currentTimeMillis());
      DeleteResult deleteResult =
          namedBlobDb.delete(record.getAccountName(), record.getContainerName(), record.getBlobName()).get();
      assertEquals("Expect exactly one blob version to be deleted", 1, deleteResult.getBlobVersions().size());
      assertEquals("Unexpected deleted ID", record.getBlobId(), deleteResult.getBlobVersions().get(0).getBlobId());
      assertFalse("Unexpected alreadyDeleted value", deleteResult.getBlobVersions().get(0).isAlreadyDeleted());
      time.setCurrentMilliseconds(System.currentTimeMillis());
      checkErrorCode(() -> namedBlobDb.get(record.getAccountName(), record.getContainerName(), record.getBlobName()),
          enableHardDelete ? RestServiceErrorCode.NotFound : RestServiceErrorCode.Deleted);
      if (!enableHardDelete) {
        NamedBlobRecord recordFromStore =
            namedBlobDb.get(record.getAccountName(), record.getContainerName(), record.getBlobName(),
                GetOption.Include_Deleted_Blobs, false).get();
        checkRecordsEqual(record, recordFromStore);

        recordFromStore = namedBlobDb.get(record.getAccountName(), record.getContainerName(), record.getBlobName(),
            GetOption.Include_All, false).get();
        checkRecordsEqual(record, recordFromStore);
      }
    }

    if (!enableHardDelete) {
      // deletes should be idempotent and additional delete calls should succeed
      time.setCurrentMilliseconds(System.currentTimeMillis());
      for (NamedBlobRecord record : records) {
        DeleteResult deleteResult =
            namedBlobDb.delete(record.getAccountName(), record.getContainerName(), record.getBlobName()).get();
        assertEquals("Expect exactly one blob version to be deleted", 1, deleteResult.getBlobVersions().size());
        assertEquals("Unexpected deleted ID", record.getBlobId(), deleteResult.getBlobVersions().get(0).getBlobId());
        assertTrue("Unexpected alreadyDeleted value", deleteResult.getBlobVersions().get(0).isAlreadyDeleted());
      }
    }

    // delete and get for non existent blobs should return not found.
    time.setCurrentMilliseconds(System.currentTimeMillis());
    for (NamedBlobRecord record : records) {
      String nonExistentName = record.getBlobName() + "-other";
      checkErrorCode(() -> namedBlobDb.get(record.getAccountName(), record.getContainerName(), nonExistentName),
          RestServiceErrorCode.NotFound);
      checkErrorCode(() -> namedBlobDb.delete(record.getAccountName(), record.getContainerName(), nonExistentName),
          RestServiceErrorCode.NotFound);
    }

    records.clear();
    // should be able to put new records again after deletion
    time.setCurrentMilliseconds(System.currentTimeMillis());
    for (Account account : accountService.getAllAccounts()) {
      for (Container container : account.getAllContainers()) {
        for (int i = 0; i < blobsPerContainer; i++) {
          String blobId = getBlobId(account, container);
          String blobName = "name/" + i + "/more path segments--";
          long expirationTime =
              i % 2 == 1 ? Utils.Infinite_Time : System.currentTimeMillis() + TimeUnit.HOURS.toMillis(1);
          NamedBlobRecord record =
              new NamedBlobRecord(account.getName(), container.getName(), blobName, blobId, expirationTime);
          namedBlobDb.put(record).get();
          records.add(record);
        }
      }
    }
  }
Tests sequences of puts, gets, lists, and deletes across multiple containers.
@throws Exception
ambry-named-mysql/src/integration-test/java/com/github/ambry/named/MySqlNamedBlobDbIntegrationTest.java