@Override
  public Future<Long> write(ByteBuffer src, Callback<Long> callback) {
    // still need to copy the buffer since the writer may decide to reuse the buffer after the callback is called.
    return writeInternal(() -> {
      ByteBuf copy = src.isDirect() ? ByteBufAllocator.DEFAULT.directBuffer(src.remaining())
          : PooledByteBufAllocator.DEFAULT.heapBuffer(src.remaining());
      copy.writeBytes(src);
      src.position(src.limit());
      return copy;
    }, callback);
  }
@inheritDoc

This method requires copying the buffer because {@link ByteBuffer}s do not have the concept of a reference count,
so the other side is free to reuse the buffer after the callback has been called (as opposed to when the ref count
reaches 0).
ambry-commons/src/main/java/com/github/ambry/commons/RetainingAsyncWritableChannel.java