@Override
  public Future<Long> write(ByteBuf src, Callback<Long> callback) {
    long writeProcessingStartTime = System.currentTimeMillis();
    if (!responseMetadataWriteInitiated.get()) {
      maybeWriteResponseMetadata(responseMetadata, new ResponseMetadataWriteListener());
    }
    if (finalResponseMetadata == null) {
      // If finalResponseMetadata is still null, it indicates channel becomes inactive.
      if (ctx.channel().isActive()) {
        logger.warn("Channel should be inactive status. {}", ctx.channel());
        nettyMetrics.channelStatusInconsistentCount.inc();
      } else {
        logger.debug("Scheduling a chunk cleanup on channel {} because response channel is closed.", ctx.channel());
        writeFuture.addListener(new CleanupCallback(new ClosedChannelException()));
      }
      FutureResult<Long> future = new FutureResult<Long>();
      future.done(0L, new ClosedChannelException());
      if (callback != null) {
        callback.onCompletion(0L, new ClosedChannelException());
      }
      return future;
    }
    Chunk chunk = new Chunk(src, callback);
    logger.debug("Netty Response Chunk size: {}", src.readableBytes());
    chunksToWrite.add(chunk);
    if (HttpUtil.isContentLengthSet(finalResponseMetadata) && totalBytesReceived.get() > HttpUtil.getContentLength(
        finalResponseMetadata)) {
      Exception exception = new IllegalStateException(
          "Size of provided content [" + totalBytesReceived.get() + "] is greater than Content-Length set ["
              + HttpUtil.getContentLength(finalResponseMetadata) + "]");
      if (!writeFuture.isDone()) {
        writeFuture.setFailure(exception);
      }
      writeFuture.addListener(new CleanupCallback(exception));
    } else if (!isOpen()) {
      // the isOpen() check is not before addition to the queue because chunks need to be acknowledged in the order
      // they were received. If we don't add it to the queue and clean up, chunks may be acknowledged out of order.
      logger.debug("Scheduling a chunk cleanup on channel {} because response channel is closed", ctx.channel());
      writeFuture.addListener(new CleanupCallback(new ClosedChannelException()));
    } else if (finalResponseMetadata instanceof FullHttpResponse) {
      logger.debug("Scheduling a chunk cleanup on channel {} because Content-Length is 0", ctx.channel());
      Exception exception = null;
      // this is only allowed to be a 0 sized buffer.
      if (src.readableBytes() > 0) {
        exception = new IllegalStateException("Provided non zero size content after setting Content-Length to 0");
        if (!writeFuture.isDone()) {
          writeFuture.setFailure(exception);
        }
      }
      writeFuture.addListener(new CleanupCallback(exception));
    } else {
      chunkedWriteHandler.resumeTransfer();
    }

    long writeProcessingTime = System.currentTimeMillis() - writeProcessingStartTime;
    nettyMetrics.writeProcessingTimeInMs.update(writeProcessingTime);
    if (request != null) {
      request.getMetricsTracker().nioMetricsTracker.addToResponseProcessingTime(writeProcessingTime);
    }
    return chunk.future;
  }
Create an instance of NettyResponseChannel that will use {@code ctx} to return responses.
@param ctx the {@link ChannelHandlerContext} to use.
@param nettyMetrics the {@link NettyMetrics} instance to use.
@param performanceConfig the configuration object to use for performance evaluation.
@param nettyConfig the configuration object to use for netty related config.
ambry-rest/src/main/java/com/github/ambry/rest/NettyResponseChannel.java