@Test
  public void testNamedBlobDelete() throws Exception {
    try {
      maxPutChunkSize = PUT_CONTENT_SIZE;
      String deleteServiceId = "delete-service";
      // metadata blob + data chunks.
      final AtomicInteger deletesInitiated = new AtomicInteger();
      final AtomicReference<String> receivedDeleteServiceId = new AtomicReference<>();
      LoggingNotificationSystem deleteTrackingNotificationSystem = new LoggingNotificationSystem() {
        @Override
        public void onBlobDeleted(String blobId, String serviceId, Account account, Container container) {
          deletesInitiated.incrementAndGet();
          receivedDeleteServiceId.set(serviceId);
        }
      };

      final int NUM_VERSIONS = 3;
      final String ACCOUNT_NAME = "accountName";
      final String CONTAINER_NAME = "containerName";
      final String BLOB_NAME = "blobName";
      List<String> blobIds = new ArrayList<>();
      // setup mock idconverter
      IdConverterFactory mockIdConverterFactory = mock(IdConverterFactory.class);
      IdConverter mockIdConverter = mock(IdConverter.class);
      when(mockIdConverterFactory.getIdConverter()).thenReturn(mockIdConverter);
      when(mockIdConverter.convert(any(RestRequest.class), anyString(), any(), any())).thenAnswer(invocation -> {
        FutureResult<String> futureResult = new FutureResult<>();
        RestRequest restRequest = invocation.getArgument(0);
        String blobId = invocation.getArgument(1);
        Callback<String> callback = invocation.getArgument(3);
        String returnValue = null;
        if (restRequest.getRestMethod() == RestMethod.DELETE) {
          // This is coming from a delete request
          returnValue = blobIds.stream().reduce((blobId1, blobId2) -> blobId1 + "," + blobId2).orElse("");
        } else {
          // This is coming from a put request
          returnValue = blobId;
        }
        if (callback != null) {
          callback.onCompletion(returnValue, null);
        }
        futureResult.done(returnValue, null);
        return futureResult;
      });

      Properties props = getNonBlockingRouterProperties(localDcName);
      setRouterWithIdConverterFactory(props, new MockServerLayout(mockClusterMap), deleteTrackingNotificationSystem,
          mockIdConverterFactory);
      // Create 3 versions of the same named blob
      for (int i = 0; i < NUM_VERSIONS; i++) {
        setOperationParams();
        RestRequest request = createNamedBlobRestRequest(RestMethod.PUT, ACCOUNT_NAME, CONTAINER_NAME, BLOB_NAME);
        String blobId =
            router.putBlob(request, putBlobProperties, putUserMetadata, putChannel, new PutBlobOptionsBuilder().build(),
                null, null).get();
        blobIds.add(blobId);
      }
      RestRequest request = createNamedBlobRestRequest(RestMethod.DELETE, ACCOUNT_NAME, CONTAINER_NAME, BLOB_NAME);
      router.deleteBlob(request, null, deleteServiceId, null, null).get();
      long waitStart = SystemTime.getInstance().milliseconds();
      while (router.getBackgroundOperationsCount() != 0
          && SystemTime.getInstance().milliseconds() < waitStart + AWAIT_TIMEOUT_MS) {
        Thread.sleep(1000);
      }
      Assert.assertEquals("All background operations should be complete ", 0, router.getBackgroundOperationsCount());
      Assert.assertEquals("Only the original blob deletion should have been initiated", NUM_VERSIONS,
          deletesInitiated.get());
      Assert.assertEquals("The delete service ID should match the expected value", deleteServiceId,
          receivedDeleteServiceId.get());
      Assert.assertEquals("Get should have been skipped", NUM_VERSIONS, routerMetrics.skippedGetBlobCount.getCount());
      // Now try to get the blob ids, expected not found
      for (String blobId : blobIds) {
        try {
          router.getBlob(blobId, new GetBlobOptionsBuilder().build(), null, null).get();
          fail("Blob id " + blobId + " should be deleted");
        } catch (Exception e) {
          RouterException r = (RouterException) e.getCause();
          Assert.assertEquals(RouterErrorCode.BlobDeleted, r.getErrorCode());
        }
      }
    } finally {
      if (router != null) {
        router.close();
        assertClosed();
        Assert.assertEquals("All operations should have completed", 0, router.getOperationsCount());
      }
    }
  }
Test to ensure when deleting a named blob, all versions are deleted and all blob ids are deleted.
@throws Exception
ambry-router/src/test/java/com/github/ambry/router/NonBlockingRouterTest.java