private void fillChunksAndAssertSuccess() throws Exception {
    VerifiableProperties vProps = getNonBlockingRouterProperties();
    MockClusterMap mockClusterMap = new MockClusterMap();
    RouterConfig routerConfig = new RouterConfig(vProps);
    routerMetrics = new NonBlockingRouterMetrics(mockClusterMap, routerConfig);
    ResponseHandler responseHandler = new ResponseHandler(mockClusterMap);
    short accountId = Utils.getRandomShort(random);
    short containerId = Utils.getRandomShort(random);
    BlobProperties putBlobProperties =
        new BlobProperties(blobSize, "serviceId", "memberId", "contentType", false, Utils.Infinite_Time, accountId,
            containerId, testEncryption, null, null, null);
    Random random = new Random();
    byte[] putUserMetadata = new byte[10];
    random.nextBytes(putUserMetadata);
    putContent = new byte[blobSize];
    random.nextBytes(putContent);
    final ReadableStreamChannel putChannel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(putContent));
    FutureResult<String> futureResult = new FutureResult<String>();
    MockTime time = new MockTime();
    MockNetworkClientFactory networkClientFactory = new MockNetworkClientFactory(vProps, null, 0, 0, 0, null, time);
    if (testEncryption) {
      kms = new MockKeyManagementService(new KMSConfig(vProps),
          TestUtils.getRandomKey(SingleKeyManagementServiceTest.DEFAULT_KEY_SIZE_CHARS));
      cryptoService = new MockCryptoService(new CryptoServiceConfig(vProps));
      cryptoJobHandler = new CryptoJobHandler(CryptoJobHandlerTest.DEFAULT_THREAD_COUNT);
    }
    MockRouterCallback routerCallback =
        new MockRouterCallback(networkClientFactory.getNetworkClient(), Collections.EMPTY_LIST);
    PutOperation op =
        PutOperation.forUpload(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(),
            new InMemAccountService(true, false), putUserMetadata, putChannel, PutBlobOptions.DEFAULT, futureResult,
            null, routerCallback, null, kms, cryptoService, cryptoJobHandler, time, putBlobProperties,
            MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback,
            new CompressionService(routerConfig.getCompressionConfig(), routerMetrics.compressionMetrics));
    op.startOperation();
    numChunks = RouterUtils.getNumChunksForBlobAndChunkSize(blobSize, chunkSize);
    compositeBuffers = new ByteBuf[numChunks];
    compositeEncryptionKeys = new ByteBuffer[numChunks];
    compositeBlobIds = new BlobId[numChunks];
    final AtomicReference<Exception> operationException = new AtomicReference<Exception>(null);

    int chunksLeftToBeFilled = numChunks;
    do {
      if (testEncryption) {
        int chunksPerBatch = Math.min(routerConfig.routerMaxInMemPutChunks, chunksLeftToBeFilled);
        CountDownLatch onPollLatch = new CountDownLatch(chunksPerBatch);
        routerCallback.setOnPollLatch(onPollLatch);
        op.fillChunks();
        Assert.assertTrue("Latch should have been zeroed out", onPollLatch.await(1000, TimeUnit.MILLISECONDS));
        chunksLeftToBeFilled -= chunksPerBatch;
      } else {
        op.fillChunks();
      }
      // All existing chunks must have been filled if no work was done in the last call,
      // since the channel is ByteBuffer based.
      for (PutOperation.PutChunk putChunk : op.putChunks) {
        if (putChunk.isFree()) {
          continue;
        }
        Assert.assertEquals("Chunk should be ready.", PutOperation.ChunkState.Ready, putChunk.getState());
        ByteBuf buf = putChunk.buf.retainedDuplicate();
        totalSizeWritten += buf.readableBytes();
        compositeBuffers[putChunk.getChunkIndex()] = buf;
        if (testEncryption) {
          compositeEncryptionKeys[putChunk.getChunkIndex()] = putChunk.encryptedPerBlobKey.duplicate();
          compositeBlobIds[putChunk.getChunkIndex()] = putChunk.chunkBlobId;
        }
        putChunk.clear();
      }
    } while (!op.isChunkFillingDone());

    if (!testEncryption) {
      Assert.assertEquals("total size written out should match the blob size", blobSize, totalSizeWritten);
    }
    // for encrypted path, size will be implicitly tested via assertDataIdentity

    Exception exception = operationException.get();
    if (exception != null) {
      throw exception;
    }
    assertDataIdentity(mockClusterMap);
  }
Create a {@link PutOperation} and pass in a channel with the blobSize set by the caller; and test the chunk
filling flow for puts.
Note that this test is for the chunk filling flow, not for the ChunkFiller thread (which never gets exercised,
as we do not even instantiate the {@link PutManager})
ambry-router/src/test/java/com/github/ambry/router/ChunkFillTest.java