@Test
  public void addAndGetDatasetVersionTest() throws Exception {
    //Add dataset
    Account testAccount = new ArrayList<>(accountService.getAllAccounts()).get(1);
    Container testContainer = new ArrayList<>(testAccount.getAllContainers()).get(1);
    Dataset.VersionSchema versionSchema = Dataset.VersionSchema.TIMESTAMP;
    Map<String, String> userTags = new HashMap<>();
    String userTagKey = "tagKey";
    userTags.put(userTagKey, "tagValues");
    long datasetTtl = 3600;

    Dataset dataset =
        new DatasetBuilder(testAccount.getName(), testContainer.getName(), DATASET_NAME).setVersionSchema(versionSchema)
            .setRetentionTimeInSeconds(datasetTtl)
            .setUserTags(userTags)
            .build();
    byte[] datasetsUpdateJson = AccountCollectionSerde.serializeDatasetsInJson(dataset);
    List<ByteBuffer> body = new LinkedList<>();
    body.add(ByteBuffer.wrap(datasetsUpdateJson));
    body.add(null);
    JSONObject headers = new JSONObject().put(RestUtils.Headers.TARGET_ACCOUNT_NAME, testAccount.getName())
        .put(RestUtils.Headers.TARGET_CONTAINER_NAME, testContainer.getName());
    RestRequest restRequest =
        createRestRequest(RestMethod.POST, Operations.ACCOUNTS_CONTAINERS_DATASETS, headers, body);
    MockRestResponseChannel restResponseChannel = new MockRestResponseChannel();
    doOperation(restRequest, restResponseChannel);

    // Add dataset version
    Long version = System.currentTimeMillis();
    String blobName = SLASH + DATASET_NAME + SLASH + version;
    String namedBlobPathUri =
        NAMED_BLOB_PREFIX + SLASH + testAccount.getName() + SLASH + testContainer.getName() + blobName;
    int contentLength = 10;
    ByteBuffer content = ByteBuffer.wrap(TestUtils.getRandomBytes(contentLength));
    body = new LinkedList<>();
    body.add(content);
    body.add(null);

    long blobTtl = 7200;
    String serviceId = "test";
    String contentType = "application/octet-stream";
    String ownerId = "owner";
    String userMetadataKey = USER_META_DATA_HEADER_PREFIX + "userMetadata";
    headers = new JSONObject();
    setAmbryHeadersForPut(headers, blobTtl, testContainer.isCacheable(), serviceId, contentType, ownerId, null, null,
        null);
    headers.put(RestUtils.Headers.DATASET_VERSION_QUERY_ENABLED, true);
    headers.put(userMetadataKey, "userMetadataValue");
    restRequest = createRestRequest(RestMethod.PUT, namedBlobPathUri, headers, body);
    restResponseChannel = new MockRestResponseChannel();

    List<? extends PartitionId> partitions = clusterMap.getWritablePartitionIds(null);
    String blobId =
        new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, ClusterMap.UNKNOWN_DATACENTER_ID, testAccount.getId(),
            testContainer.getId(), partitions.get(ThreadLocalRandom.current().nextInt(partitions.size())), false,
            BlobId.BlobDataType.DATACHUNK).getID();

    reset(namedBlobDb);
    NamedBlobRecord namedBlobRecord =
        new NamedBlobRecord(testAccount.getName(), testContainer.getName(), blobName + SLASH + version, blobId,
            Utils.Infinite_Time);
    when(namedBlobDb.put(any(), any(), any())).thenReturn(
        CompletableFuture.completedFuture(new PutResult(namedBlobRecord)));

    // Issue put dataset version request
    doOperation(restRequest, restResponseChannel);
    String blobIdFromResponse = (String) restResponseChannel.getResponseHeaders().get(RestUtils.Headers.LOCATION);
    Map<String, InMemoryRouter.InMemoryBlob> allBlobs = router.getActiveBlobs();
    InMemoryRouter.InMemoryBlob blob = allBlobs.values().iterator().next();
    Map<String, String> userMetadataFromRouter = RestUtils.buildUserMetadata(blob.getUserMetadata());
    BlobProperties expectedBlobProperties = blob.getBlobProperties();
    assertEquals("Unexpected PUT /DatasetVersions response", blobId, blobIdFromResponse);
    assertEquals("Unexpected userMetadata", userTags.get(userTagKey),
        userMetadataFromRouter.get(USER_META_DATA_HEADER_PREFIX + userTagKey));
    assertEquals("Ttl should be the dataset level ttl", (long) dataset.getRetentionTimeInSeconds(),
        expectedBlobProperties.getTimeToLiveInSeconds());

    // Add a dataset version without version specified.
    namedBlobPathUri =
        NAMED_BLOB_PREFIX + SLASH + testAccount.getName() + SLASH + testContainer.getName() + SLASH + DATASET_NAME;
    try {
      restRequest = createRestRequest(RestMethod.PUT, namedBlobPathUri, headers, body);
      restResponseChannel = new MockRestResponseChannel();
      doOperation(restRequest, restResponseChannel);
      fail("Should fail due to version has not been provided.");
    } catch (RestServiceException e) {
      assertEquals("Unexpected error code", RestServiceErrorCode.BadRequest, e.getErrorCode());
    }

    // Prepare the input and mock class
    namedBlobPathUri = NAMED_BLOB_PREFIX + SLASH + testAccount.getName() + SLASH + testContainer.getName() + blobName;
    BlobProperties blobProperties =
        new BlobProperties(0, testAccount.getName(), "owner", "image/gif", false, blobTtl, testAccount.getId(),
            testContainer.getId(), false, null, null, null);
    ReadableStreamChannel byteBufferContent = new ByteBufferReadableStreamChannel(ByteBuffer.allocate(contentLength));
    String blobIdFromRouter =
        router.putBlobWithIdVersion(blobProperties, new byte[0], byteBufferContent, BlobId.BLOB_ID_V6).get();
    reset(namedBlobDb);
    namedBlobRecord = new NamedBlobRecord(testAccount.getName(), testContainer.getName(), blobName + SLASH + version,
        blobIdFromRouter, Utils.Infinite_Time);
    when(namedBlobDb.get(any(), any(), any(), any(), eq(false))).thenReturn(
        CompletableFuture.completedFuture(namedBlobRecord));

    headers = new JSONObject();
    headers.put(RestUtils.Headers.DATASET_VERSION_QUERY_ENABLED, true);
    restRequest = createRestRequest(RestMethod.GET, namedBlobPathUri, headers, null);
    restResponseChannel = new MockRestResponseChannel();

    //Issue get dataset version request
    doOperation(restRequest, restResponseChannel);
    assertEquals("Unexpected GET /DatasetVersions response content length", contentLength,
        restResponseChannel.getResponseBody().length);

    //delete the dataset version
    reset(namedBlobDb);
    when(namedBlobDb.delete(any(), any(), any())).thenReturn(
        CompletableFuture.completedFuture(new DeleteResult(blobIdFromRouter, false)));
    when(namedBlobDb.get(any(), any(), any(), any(), eq(false))).thenReturn(
        CompletableFuture.completedFuture(namedBlobRecord));

    headers = new JSONObject();
    headers.put(RestUtils.Headers.DATASET_VERSION_QUERY_ENABLED, true);
    restRequest = createRestRequest(RestMethod.DELETE, namedBlobPathUri, headers, null);
    verifyDeleteAccepted(restRequest);
    assertEquals("Unexpected number of blobs has been deleted", 1, router.getDeletedBlobs().size());

    //get the deleted dataset version
    headers = new JSONObject();
    headers.put(RestUtils.Headers.DATASET_VERSION_QUERY_ENABLED, true);
    restRequest = createRestRequest(RestMethod.GET, namedBlobPathUri, headers, null);
    verifyOperationFailure(restRequest, RestServiceErrorCode.Deleted);
    assertEquals("Unexpected number of blobs has been deleted", 1, router.getDeletedBlobs().size());

    //Add dataset without user tags
    versionSchema = Dataset.VersionSchema.MONOTONIC;

    dataset = new DatasetBuilder(testAccount.getName(), testContainer.getName(),
        DATASET_NAME_WITHOUT_USER_TAGS).setVersionSchema(versionSchema).setRetentionTimeInSeconds((long) -1).build();
    datasetsUpdateJson = AccountCollectionSerde.serializeDatasetsInJson(dataset);
    body = new LinkedList<>();
    body.add(ByteBuffer.wrap(datasetsUpdateJson));
    body.add(null);
    headers = new JSONObject().put(RestUtils.Headers.TARGET_ACCOUNT_NAME, testAccount.getName())
        .put(RestUtils.Headers.TARGET_CONTAINER_NAME, testContainer.getName());
    restRequest = createRestRequest(RestMethod.POST, Operations.ACCOUNTS_CONTAINERS_DATASETS, headers, body);
    restResponseChannel = new MockRestResponseChannel();
    doOperation(restRequest, restResponseChannel);

    //Add a dataset version
    version = System.currentTimeMillis();
    blobName = DATASET_NAME_WITHOUT_USER_TAGS + SLASH + version;
    namedBlobPathUri =
        NAMED_BLOB_PREFIX + SLASH + testAccount.getName() + SLASH + testContainer.getName() + SLASH + blobName;
    contentLength = 10;
    content = ByteBuffer.wrap(TestUtils.getRandomBytes(contentLength));
    body = new LinkedList<>();
    body.add(content);
    body.add(null);

    blobTtl = 7200;
    serviceId = "test";
    contentType = "application/octet-stream";
    ownerId = "owner";
    headers = new JSONObject();
    setAmbryHeadersForPut(headers, blobTtl, testContainer.isCacheable(), serviceId, contentType, ownerId, null, null,
        null);
    headers.put(RestUtils.Headers.DATASET_VERSION_QUERY_ENABLED, true);
    headers.put(DATASET_VERSION_TTL_ENABLED, true);
    restRequest = createRestRequest(RestMethod.PUT, namedBlobPathUri, headers, body);
    restResponseChannel = new MockRestResponseChannel();

    blobProperties =
        new BlobProperties(0, testAccount.getName(), "owner", "image/gif", false, blobTtl, testAccount.getId(),
            testContainer.getId(), false, null, null, null);
    byteBufferContent = new ByteBufferReadableStreamChannel(ByteBuffer.allocate(contentLength));
    blobIdFromRouter =
        router.putBlobWithIdVersion(blobProperties, new byte[0], byteBufferContent, BlobId.BLOB_ID_V6).get();
    reset(namedBlobDb);
    namedBlobRecord =
        new NamedBlobRecord(testAccount.getName(), testContainer.getName(), blobName, blobIdFromRouter, blobTtl);
    when(namedBlobDb.put(any(), any(), any())).thenReturn(
        CompletableFuture.completedFuture(new PutResult(namedBlobRecord)));
    when(namedBlobDb.delete(namedBlobRecord.getAccountName(), namedBlobRecord.getContainerName(), blobName)).thenReturn(
        CompletableFuture.completedFuture(new DeleteResult(blobIdFromRouter, false)));
    when(namedBlobDb.get(namedBlobRecord.getAccountName(), namedBlobRecord.getContainerName(), blobName, null,
        false)).thenReturn(CompletableFuture.completedFuture(namedBlobRecord));
    when(namedBlobDb.updateBlobTtlAndStateToReady(any())).thenReturn(
        CompletableFuture.completedFuture(new PutResult(namedBlobRecord)));

    // Issue put dataset version request, should not throw null point exception.
    doOperation(restRequest, restResponseChannel);
    assertEquals("Unexpected ttl", blobTtl,
        allBlobs.get(blobIdFromRouter).getBlobProperties().getTimeToLiveInSeconds());

    // Add another dataset version.
    version = System.currentTimeMillis();
    String blobName1 = DATASET_NAME_WITHOUT_USER_TAGS + SLASH + version;
    String namedBlobPathUri1 =
        NAMED_BLOB_PREFIX + SLASH + testAccount.getName() + SLASH + testContainer.getName() + SLASH + blobName1;
    contentLength = 10;
    content = ByteBuffer.wrap(TestUtils.getRandomBytes(contentLength));
    body = new LinkedList<>();
    body.add(content);
    body.add(null);
    blobTtl = 7200;
    serviceId = "test";
    contentType = "application/octet-stream";
    ownerId = "owner";
    headers = new JSONObject();
    setAmbryHeadersForPut(headers, blobTtl, testContainer.isCacheable(), serviceId, contentType, ownerId, null, null,
        null);
    headers.put(RestUtils.Headers.DATASET_VERSION_QUERY_ENABLED, true);
    restRequest = createRestRequest(RestMethod.PUT, namedBlobPathUri1, headers, body);
    restResponseChannel = new MockRestResponseChannel();

    blobProperties =
        new BlobProperties(0, testAccount.getName(), "owner", "image/gif", false, blobTtl, testAccount.getId(),
            testContainer.getId(), false, null, null, null);
    byteBufferContent = new ByteBufferReadableStreamChannel(ByteBuffer.allocate(contentLength));
    String blobIdFromRouter1 =
        router.putBlobWithIdVersion(blobProperties, new byte[0], byteBufferContent, BlobId.BLOB_ID_V6).get();
    NamedBlobRecord namedBlobRecord1 =
        new NamedBlobRecord(testAccount.getName(), testContainer.getName(), blobName1, blobIdFromRouter1,
            Utils.Infinite_Time);
    when(namedBlobDb.put(any(), any(), any())).thenReturn(
        CompletableFuture.completedFuture(new PutResult(namedBlobRecord1)));
    when(namedBlobDb.delete(namedBlobRecord1.getAccountName(), namedBlobRecord1.getContainerName(),
        blobName1)).thenReturn(CompletableFuture.completedFuture(new DeleteResult(blobIdFromRouter1, false)));
    when(namedBlobDb.get(namedBlobRecord1.getAccountName(), namedBlobRecord1.getContainerName(), blobName1, null,
        false)).thenReturn(CompletableFuture.completedFuture(namedBlobRecord1));
    when(namedBlobDb.updateBlobTtlAndStateToReady(any())).thenReturn(
        CompletableFuture.completedFuture(new PutResult(namedBlobRecord1)));

    doOperation(restRequest, restResponseChannel);

    //delete the dataset
    headers = new JSONObject().put(RestUtils.Headers.TARGET_ACCOUNT_NAME, testAccount.getName())
        .put(RestUtils.Headers.TARGET_CONTAINER_NAME, testContainer.getName())
        .put(RestUtils.Headers.TARGET_DATASET_NAME, DATASET_NAME_WITHOUT_USER_TAGS);
    restRequest = createRestRequest(RestMethod.DELETE, Operations.ACCOUNTS_CONTAINERS_DATASETS, headers, null);
    verifyDeleteAccepted(restRequest);
    assertEquals("Unexpected number of blobs has been deleted", 3, router.getDeletedBlobs().size());

    //get the dataset, should be deleted.
    headers = new JSONObject().put(RestUtils.Headers.TARGET_ACCOUNT_NAME, testAccount.getName())
        .put(RestUtils.Headers.TARGET_CONTAINER_NAME, testContainer.getName())
        .put(RestUtils.Headers.TARGET_DATASET_NAME, DATASET_NAME_WITHOUT_USER_TAGS);
    restRequest = createRestRequest(RestMethod.GET, Operations.ACCOUNTS_CONTAINERS_DATASETS, headers, null);
    verifyOperationFailure(restRequest, RestServiceErrorCode.Deleted);

    //get the datset version under the dataset, should be deleted.
    headers = new JSONObject();
    headers.put(RestUtils.Headers.DATASET_VERSION_QUERY_ENABLED, true);
    restRequest = createRestRequest(RestMethod.GET, namedBlobPathUri, headers, null);
    verifyOperationFailure(restRequest, RestServiceErrorCode.Deleted);

    headers = new JSONObject();
    headers.put(RestUtils.Headers.DATASET_VERSION_QUERY_ENABLED, true);
    restRequest = createRestRequest(RestMethod.GET, namedBlobPathUri1, headers, null);
    verifyOperationFailure(restRequest, RestServiceErrorCode.Deleted);
  }
Test add and get dataset version.
@throws Exception
ambry-frontend/src/test/java/com/github/ambry/frontend/FrontendRestRequestServiceTest.java