@Test
  public void testUnsuccessfulPutDataChunkDelete() throws Exception {
    try {
      // Ensure there are 4 chunks.
      maxPutChunkSize = PUT_CONTENT_SIZE / 4;
      Properties props = getNonBlockingRouterProperties(localDcName);
      VerifiableProperties verifiableProperties = new VerifiableProperties((props));
      RouterConfig routerConfig = new RouterConfig(verifiableProperties);
      MockClusterMap mockClusterMap = new MockClusterMap();
      MockTime mockTime = new MockTime();
      MockServerLayout mockServerLayout = new MockServerLayout(mockClusterMap);
      // Since this test wants to ensure that successfully put data chunks are deleted when the overall put operation
      // fails, it uses a notification system to track the deletions.
      final CountDownLatch deletesDoneLatch = new CountDownLatch(2);
      final Map<String, String> blobsThatAreDeleted = new HashMap<>();
      LoggingNotificationSystem deleteTrackingNotificationSystem = new LoggingNotificationSystem() {
        @Override
        public void onBlobDeleted(String blobId, String serviceId, Account account, Container container) {
          blobsThatAreDeleted.put(blobId, serviceId);
          deletesDoneLatch.countDown();
        }
      };
      router = new NonBlockingRouter(routerConfig, new NonBlockingRouterMetrics(mockClusterMap, routerConfig),
          new MockNetworkClientFactory(verifiableProperties, mockSelectorState, MAX_PORTS_PLAIN_TEXT, MAX_PORTS_SSL,
              CHECKOUT_TIMEOUT_MS, mockServerLayout, mockTime), deleteTrackingNotificationSystem, mockClusterMap, kms,
          cryptoService, cryptoJobHandler, accountService, mockTime, MockClusterMap.DEFAULT_PARTITION_CLASS, null);

      setOperationParams();

      List<DataNodeId> dataNodeIds = mockClusterMap.getDataNodeIds();
      List<ServerErrorCode> serverErrorList = new ArrayList<>();
      // There are 4 chunks for this blob.
      // All put operations make one request to each local server as there are 3 servers overall in the local DC.
      // Set the state of the mock servers so that they return success for the first 2 requests in order to succeed
      // the first two chunks.
      serverErrorList.add(ServerErrorCode.NoError);
      serverErrorList.add(ServerErrorCode.NoError);
      // fail requests for third and fourth data chunks including the slipped put attempts:
      serverErrorList.add(ServerErrorCode.UnknownError);
      serverErrorList.add(ServerErrorCode.UnknownError);
      serverErrorList.add(ServerErrorCode.UnknownError);
      serverErrorList.add(ServerErrorCode.UnknownError);
      // all subsequent requests (no more puts, but there will be deletes) will succeed.
      for (DataNodeId dataNodeId : dataNodeIds) {
        MockServer server = mockServerLayout.getMockServer(dataNodeId.getHostname(), dataNodeId.getPort());
        server.setServerErrors(serverErrorList);
      }

      // Submit the put operation and wait for it to fail.
      try {
        router.putBlob(putBlobProperties, putUserMetadata, putChannel, new PutBlobOptionsBuilder().build()).get();
      } catch (ExecutionException e) {
        Assert.assertEquals(RouterErrorCode.AmbryUnavailable, ((RouterException) e.getCause()).getErrorCode());
      }

      // Now, wait until the deletes of the successfully put blobs are complete.
      Assert.assertTrue("Deletes should not take longer than " + AWAIT_TIMEOUT_MS,
          deletesDoneLatch.await(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS));
      for (Map.Entry<String, String> blobIdAndServiceId : blobsThatAreDeleted.entrySet()) {
        Assert.assertEquals("Unexpected service ID for deleted blob",
            BackgroundDeleteRequest.SERVICE_ID_PREFIX + putBlobProperties.getServiceId(),
            blobIdAndServiceId.getValue());
      }
    } finally {
      if (router != null) {
        router.close();
        assertClosed();
        Assert.assertEquals("All operations should have completed", 0, router.getOperationsCount());
      }
    }
  }
Test that if a composite blob put fails, the successfully put data chunks are deleted.
ambry-router/src/test/java/com/github/ambry/router/NonBlockingRouterTest.java