@Test
  public void testChunkNumAndSizeCalculations() throws Exception {
    chunkSize = 4 * 1024 * 1024;
    // a large blob greater than Integer.MAX_VALUE and not at chunk size boundary.
    final long blobSize = ((long) Integer.MAX_VALUE / chunkSize + 1) * chunkSize + random.nextInt(chunkSize - 1) + 1;
    VerifiableProperties vProps = getNonBlockingRouterProperties();
    MockClusterMap mockClusterMap = new MockClusterMap();
    RouterConfig routerConfig = new RouterConfig(vProps);
    NonBlockingRouterMetrics routerMetrics = new NonBlockingRouterMetrics(mockClusterMap, routerConfig);
    short accountId = Utils.getRandomShort(random);
    short containerId = Utils.getRandomShort(random);
    BlobProperties putBlobProperties =
        new BlobProperties(blobSize, "serviceId", "memberId", "contentType", false, Utils.Infinite_Time, accountId,
            containerId, false, null, null, null);
    Random random = new Random();
    byte[] putUserMetadata = new byte[10];
    random.nextBytes(putUserMetadata);
    final MockReadableStreamChannel putChannel = new MockReadableStreamChannel(blobSize, false);
    FutureResult<String> futureResult = new FutureResult<String>();
    MockTime time = new MockTime();
    MockNetworkClientFactory networkClientFactory = new MockNetworkClientFactory(vProps, null, 0, 0, 0, null, time);
    PutOperation op =
        PutOperation.forUpload(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(),
            new InMemAccountService(true, false), putUserMetadata, putChannel, PutBlobOptions.DEFAULT, futureResult,
            null, new RouterCallback(networkClientFactory.getNetworkClient(), new ArrayList<>()), null, null, null,
            null, new MockTime(), putBlobProperties, MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback,
            new CompressionService(routerConfig.getCompressionConfig(), routerMetrics.compressionMetrics));
    op.startOperation();
    numChunks = RouterUtils.getNumChunksForBlobAndChunkSize(blobSize, chunkSize);
    // largeBlobSize is not a multiple of chunkSize
    int expectedNumChunks = (int) (blobSize / chunkSize + 1);
    Assert.assertEquals("numChunks should be as expected", expectedNumChunks, numChunks);
    int lastChunkSize = (int) (blobSize % chunkSize);
    final AtomicReference<Exception> channelException = new AtomicReference<Exception>(null);
    int chunkIndex = 0;

    // The write to the MockReadableStreamChannel blocks until the data is read as part fo the chunk filling,
    // so create a thread that fills the MockReadableStreamChannel.
    Utils.newThread(new Runnable() {
      @Override
      public void run() {
        try {
          byte[] writeBuf = new byte[chunkSize];
          long written = 0;
          while (written < blobSize) {
            int toWrite = (int) Math.min(chunkSize, blobSize - written);
            putChannel.write(ByteBuffer.wrap(writeBuf, 0, toWrite));
            written += toWrite;
          }
        } catch (Exception e) {
          channelException.set(e);
        }
      }
    }, false).start();

    // Do the chunk filling.
    boolean fillingComplete = false;
    boolean firstChunkChecked = false;
    PutOperation.PutChunk firstPutChunk = null;
    do {
      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) {
        Assert.assertNull("Mock channel write should not have caused an exception", channelException.get());
        if (putChunk.isFree()) {
          continue;
        }
        if (chunkIndex == numChunks - 1) {
          // last chunk may not be Ready as it is dependent on the completion callback to be called.
          Assert.assertTrue("Chunk should be Building or Ready.", putChunk.getState() == PutOperation.ChunkState.Ready
              || putChunk.getState() == PutOperation.ChunkState.Building);
          if (putChunk.getState() == PutOperation.ChunkState.Ready) {
            Assert.assertEquals("Chunk size should be the last chunk size", lastChunkSize,
                putChunk.buf.readableBytes());
            Assert.assertTrue("Chunk Filling should be complete at this time", op.isChunkFillingDone());
            fillingComplete = true;
          }
        } else {
          // if chunkIndex is 0, then the chunk should be Building.
          if (chunkIndex == 0) {
            if (!firstChunkChecked) {
              Assert.assertEquals("Chunk should be Building.", PutOperation.ChunkState.AwaitingBlobTypeResolution, putChunk.getState());
              firstPutChunk = putChunk;
              firstChunkChecked = true;
            }
          } else {
            // if not last chunk, then the chunk should be full and Ready.
            Assert.assertEquals("Chunk should be ready.", PutOperation.ChunkState.Ready, putChunk.getState());
          }
          Assert.assertEquals("Chunk size should be maxChunkSize", chunkSize, putChunk.buf.readableBytes());
          if (chunkIndex > 0 || (firstPutChunk != null && firstPutChunk.getState() == PutOperation.ChunkState.Ready)) {
            firstPutChunk.clear();
            chunkIndex++;
          }
          if (chunkIndex > 0) {
            putChunk.clear();
          }
        }
      }
    } while (!fillingComplete);
  }
Test the calculation of number of chunks and the size of each chunk, using a very large blob size. No content
comparison is done. This test does not consume memory more than chunkSize.
ambry-router/src/test/java/com/github/ambry/router/ChunkFillTest.java