@Test
  public void testChunkFillerSleepWithBuildingChunk() throws Exception {
    VerifiableProperties vProps = getRouterConfigInVerifiableProperties();
    MockNetworkClient networkClient =
        new MockNetworkClientFactory(vProps, mockSelectorState, MAX_PORTS_PLAIN_TEXT, MAX_PORTS_SSL,
            CHECKOUT_TIMEOUT_MS, mockServerLayout, mockTime).getMockNetworkClient();
    PutManager manager = new PutManager(mockClusterMap, new ResponseHandler(mockClusterMap), notificationSystem,
        new RouterConfig(vProps), new NonBlockingRouterMetrics(mockClusterMap, null),
        new RouterCallback(networkClient, null), "0", kms, cryptoService, cryptoJobHandler, accountService, mockTime,
        MockClusterMap.DEFAULT_PARTITION_CLASS, getNonBlockingRouter());
    BlobProperties blobProperties =
        new BlobProperties(-1, "serviceId", "memberId", "contentType", false, Utils.Infinite_Time,
            Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), false, null, null, null);
    byte[] userMetadata = new byte[10];
    byte[] content = new byte[chunkSize / 4];
    random.nextBytes(content);
    MockReadableStreamChannel channel =
        new MockReadableStreamChannel(chunkSize / 2, false); // make sure we are not sending the entire chunk
    FutureResult<String> future = new FutureResult<>();
    manager.submitPutBlobOperation(blobProperties, userMetadata, channel, PutBlobOptions.DEFAULT, future, null, null);
    channel.write(ByteBuffer.wrap(content));

    // Sleep until
    // Op has a building chunk
    // Chunk Filler is sleeping
    PutOperation op = manager.getPutOperations().iterator().next();
    Assert.assertFalse(op.isOperationComplete());
    PutOperation.PutChunk putChunk = op.putChunks.iterator().next();
    Assert.assertTrue(putChunk.isBuilding());

    manager.forceChunkFillerThreadToSleep();
    Thread chunkFillerThread = TestUtils.getThreadByThisName("ChunkFillerThread");
    Assert.assertTrue("ChunkFillerThread should have gone to WAITING state as there are no active operations",
        waitForThreadState(chunkFillerThread, Thread.State.WAITING));

    channel.beBad();
    channel.write(ByteBuffer.wrap(content));

    Assert.assertTrue(op.isOperationComplete());
    Assert.assertTrue(putChunk.isBuilding());

    manager.poll(new ArrayList<>(), new HashSet<>());
    Assert.assertTrue(putChunk.isDataReleased());
    Assert.assertEquals(0, manager.getPutOperations().size());
    router.currentOperationsCount.incrementAndGet(); // Make sure this static field's value stay the same
    manager.close();
  }
Test when channel is closed while chunk filler thread is not responding, PutManager would release the data chunks.
@throws Exception
ambry-router/src/test/java/com/github/ambry/router/PutManagerTest.java