@Override
    public HttpContent readChunk(ByteBufAllocator allocator) throws Exception {
      long chunkDispenseStartTime = System.currentTimeMillis();
      logger.trace("Servicing request for next chunk on channel {} {}", ctx.channel(), progress.get());
      HttpContent content = null;
      Chunk chunk = chunksToWrite.poll();
      if (chunk != null) {
        chunk.onDequeue();
        progress.addAndGet(chunk.buffer.readableBytes());
        chunksAwaitingCallback.add(chunk);
        // Since netty would release the outbound message after finishing writing, here we have to increase the ref-counter.
        if (chunk.isLast) {
          content = new DefaultLastHttpContent(chunk.buffer.retainedDuplicate());
          sentLastChunk = true;
        } else {
          content = new DefaultHttpContent(chunk.buffer.retainedDuplicate());
        }
      } else if (allChunksWritten() && !sentLastChunk) {
        // Send last chunk for this input
        sentLastChunk = true;
        // When the Transfer-Encoding is Chunked, we don't really know which chunk is the last chunk since the content
        // length information is lost. But here, we know that all the chunks are already put in the queue and there is
        // no chunk polling out of the queue, we send back an empty last content.
        content = LastHttpContent.EMPTY_LAST_CONTENT;
        logger.trace("Last chunk was sent on channel {}", ctx.channel());
      }
      nettyMetrics.chunkDispenseTimeInMs.update(System.currentTimeMillis() - chunkDispenseStartTime);
      return content;
    }
Dispenses the next chunk from {@link #chunksToWrite} if one is available and adds the chunk dispensed to
{@link #chunksAwaitingCallback}.
@param allocator the {@link ByteBufAllocator} to use if required.
@return a chunk of data if one is available. {@code null} otherwise.
ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java