private void flushBuffer(CompletableFuture<Long> future, ByteBuf dataBuf,
    long nextPacketOffsetInBlock, boolean syncBlock) {
    int dataLen = dataBuf.readableBytes();
    int chunkLen = summer.getBytesPerChecksum();
    int trailingPartialChunkLen = dataLen % chunkLen;
    int numChecks = dataLen / chunkLen + (trailingPartialChunkLen != 0 ? 1 : 0);
    int checksumLen = numChecks * summer.getChecksumSize();
    ByteBuf checksumBuf = alloc.directBuffer(checksumLen);
    summer.calculateChunkedSums(dataBuf.nioBuffer(), checksumBuf.nioBuffer(0, checksumLen));
    checksumBuf.writerIndex(checksumLen);
    PacketHeader header = new PacketHeader(4 + checksumLen + dataLen, nextPacketOffsetInBlock,
      nextPacketSeqno, false, dataLen, syncBlock);
    int headerLen = header.getSerializedSize();
    ByteBuf headerBuf = alloc.buffer(headerLen);
    header.putInBuffer(headerBuf.nioBuffer(0, headerLen));
    headerBuf.writerIndex(headerLen);
    Callback c =
      new Callback(future, nextPacketOffsetInBlock + dataLen, datanodeInfoMap.keySet(), dataLen);
    waitingAckQueue.addLast(c);
    // recheck again after we pushed the callback to queue
    if (state != State.STREAMING && waitingAckQueue.peekFirst() == c) {
      future.completeExceptionally(new IOException("stream already broken"));
      // it's the one we have just pushed or just a no-op
      waitingAckQueue.removeFirst();

      checksumBuf.release();
      headerBuf.release();

      // This method takes ownership of the dataBuf, so we need release it before returning.
      dataBuf.release();
      return;
    }
    // TODO: we should perhaps measure time taken per DN here;
    // we could collect statistics per DN, and/or exclude bad nodes in createOutput.
    datanodeInfoMap.keySet().forEach(ch -> {
      safeWrite(ch, headerBuf.retainedDuplicate());
      safeWrite(ch, checksumBuf.retainedDuplicate());
      safeWriteAndFlush(ch, dataBuf.retainedDuplicate());
    });
    checksumBuf.release();
    headerBuf.release();
    dataBuf.release();
    nextPacketSeqno++;
  }
An asynchronous HDFS output stream implementation which fans out data to datanode and only
supports writing file with only one block.
<p>
Use the createOutput method in {@link FanOutOneBlockAsyncDFSOutputHelper} to create. The main
usage of this class is implementing WAL, so we only expose a little HDFS configurations in the
method. And we place it here under io package because we want to make it independent of WAL
implementation thus easier to move it to HDFS project finally.
<p>
Note that, although we support pipelined flush, i.e, write new data and then flush before the
previous flush succeeds, the implementation is not thread safe, so you should not call its
methods concurrently.
<p>
Advantages compare to DFSOutputStream:
<ol>
<li>The fan out mechanism. This will reduce the latency.</li>
<li>Fail-fast when connection to datanode error. The WAL implementation could open new writer
ASAP.</li>
<li>We could benefit from netty's ByteBuf management mechanism.</li>
</ol>
hbase-asyncfs/src/main/java/org/apache/hadoop/hbase/io/asyncfs/FanOutOneBlockAsyncDFSOutput.java