void fillChunks() {
    try {
      PutChunk chunkToFill;
      while (!isChunkFillingDone()) {
        // Attempt to fill a chunk
        if (channelReadBuf == null) {
          channelReadBuf = chunkFillerChannel.getNextByteBuf(0);
        }
        if (channelReadBuf != null) {
          if (channelReadBuf.readableBytes() > 0 && isChunkAwaitingResolution()) {
            // If first chunk is awaiting resolution, and there are more bytes in the channel prepare it first as a data chunk.
            // Also, this makes the blob a composite blob.
            isSimpleBlob = false;
            maybeResolveAwaitingChunk();
          }
          maybeStopTrackingWaitForChannelDataTime();
          chunkToFill = getChunkToFill();
          if (chunkToFill == null) {
            // channel has data, but no chunks are free to be filled yet.
            maybeStartTrackingWaitForChunkTime();
            break;
          } else {
            // channel has data, and there is a chunk that can be filled.
            maybeStopTrackingWaitForChunkTime();
            bytesFilledSoFar += chunkToFill.fillFrom(channelReadBuf);
            enforceMaxUploadSize();

            if (chunkToFill.isReady() && !chunkToFill.chunkBlobProperties.isEncrypted()) {
              routerCallback.onPollReady();
            }
            if (!channelReadBuf.isReadable()) {
              chunkFillerChannel.resolveOldestChunk(null);
              channelReadBuf = null;
            }
          }
        } else {
          // channel does not have more data yet.
          if (getFreeChunk() != null) {
            // this means there is a chunk available to be filled, but no data in the channel.
            maybeStartTrackingWaitForChannelDataTime();
          }
          break;
        }
      }
      if (chunkFillingCompletedSuccessfully) {
        // If the blob size is less than 4MB or the last chunk size is less than 4MB, than this lastChunk will be
        // the chunk above.
        PutChunk lastChunk = getChunkInState(ChunkState.Building);

        // The last chunk could be the second chunk, if the blob size is less than 8MB. We need to check if any chunk is
        // awaiting resolution, and if so, resolve it first.
        maybeResolveAwaitingChunk();

        if (lastChunk != null) {
          if (chunkCounter != 0 && lastChunk.buf.readableBytes() == 0) {
            logger.trace("{}: The last buffer(s) received from chunkFillerChannel have no data, discarding them.",
                loggingContext);
            lastChunk.releaseBlobContent();
            lastChunk.state = ChunkState.Free;
          } else {
            lastChunk.onFillComplete(true);
            updateChunkFillerWaitTimeMetrics();
          }
          if (lastChunk.isReady()) {
            routerCallback.onPollReady();
          }
        }
      }
      if (operationCompleted) {
        logger.debug("{}: Operation terminated, now try to clean up chunks", loggingContext);
        // Go over the put chunk list and release building, ready and complete chunks. This is because the multi-threading.
        // We have callback in chunk.readInto to release all the ready and complete chunks but that's in different
        // thread. There might a chance when callback runs, the last chunk is still building and then change to complete.
        releaseDataForAllChunks();
      }
    } catch (Exception e) {
      RouterException routerException = e instanceof RouterException ? (RouterException) e
          : new RouterException("PutOperation fillChunks encountered unexpected error", e,
              RouterErrorCode.UnexpectedInternalError);
      PutChunk lastChunk = getChunkInState(ChunkState.Building);
      if (lastChunk != null) {
        // call release blob content, not clear, since clear should only be used in the main thread.
        lastChunk.releaseBlobContent();
      }
      PutChunk resolutionAwaitingChunk = getChunkInState(ChunkState.AwaitingBlobTypeResolution);
      if (resolutionAwaitingChunk != null) {
        resolutionAwaitingChunk.releaseBlobContent();
      }
      routerMetrics.chunkFillerUnexpectedErrorCount.inc();
      setOperationExceptionAndComplete(routerException);
      routerCallback.onPollReady();
    }
  }
This method runs in the context of the ChunkFiller thread. As long as there are chunks available to
be written to, it gets the chunk that is to be filled and keeps filling it with the data from the
chunkFillerChannel, if there is any.
@throws InterruptedException if the call to get a chunk from the chunkFillerChannel is interrupted.
ambry-router/src/main/java/com/github/ambry/router/PutOperation.java