private void flush0(CompletableFuture<Long> future, boolean syncBlock) {
    if (state != State.STREAMING) {
      future.completeExceptionally(new IOException("stream already broken"));
      return;
    }
    int dataLen = buf.readableBytes();
    if (dataLen == trailingPartialChunkLength) {
      // no new data
      long lengthAfterFlush = nextPacketOffsetInBlock + dataLen;
      Callback lastFlush = waitingAckQueue.peekLast();
      if (lastFlush != null) {
        Callback c = new Callback(future, lengthAfterFlush, Collections.emptySet(), dataLen);
        waitingAckQueue.addLast(c);
        // recheck here if we have already removed the previous callback from the queue
        if (waitingAckQueue.peekFirst() == c) {
          // all previous callbacks have been removed
          // notice that this does mean we will always win here because the background thread may
          // have already started to mark the future here as completed in the completed or failed
          // methods but haven't removed it from the queue yet. That's also why the removeFirst
          // call below may be a no-op.
          if (state != State.STREAMING) {
            future.completeExceptionally(new IOException("stream already broken"));
          } else {
            future.complete(lengthAfterFlush);
          }
          // it's the one we have just pushed or just a no-op
          waitingAckQueue.removeFirst();
        }
      } else {
        // we must have acked all the data so the ackedBlockLength must be same with
        // lengthAfterFlush
        future.complete(lengthAfterFlush);
      }
      return;
    }

    if (encryptor != null) {
      ByteBuf encryptBuf = alloc.directBuffer(dataLen);
      buf.readBytes(encryptBuf, trailingPartialChunkLength);
      int toEncryptLength = dataLen - trailingPartialChunkLength;
      try {
        encryptor.encrypt(buf.nioBuffer(trailingPartialChunkLength, toEncryptLength),
          encryptBuf.nioBuffer(trailingPartialChunkLength, toEncryptLength));
      } catch (IOException e) {
        encryptBuf.release();
        future.completeExceptionally(e);
        return;
      }
      encryptBuf.writerIndex(dataLen);
      buf.release();
      buf = encryptBuf;
    }

    if (dataLen > maxDataLen) {
      // We need to write out the data by multiple packets as the max packet allowed is 16M.
      long nextSubPacketOffsetInBlock = nextPacketOffsetInBlock;
      for (int remaining = dataLen;;) {
        if (remaining < maxDataLen) {
          flushBuffer(future, buf.readRetainedSlice(remaining), nextSubPacketOffsetInBlock,
            syncBlock);
          break;
        } else {
          flushBuffer(new CompletableFuture<>(), buf.readRetainedSlice(maxDataLen),
            nextSubPacketOffsetInBlock, syncBlock);
          remaining -= maxDataLen;
          nextSubPacketOffsetInBlock += maxDataLen;
        }
      }
    } else {
      flushBuffer(future, buf.retain(), nextPacketOffsetInBlock, syncBlock);
    }
    trailingPartialChunkLength = dataLen % summer.getBytesPerChecksum();
    ByteBuf newBuf = alloc.directBuffer(sendBufSizePRedictor.guess(dataLen))
      .ensureWritable(trailingPartialChunkLength);
    if (trailingPartialChunkLength != 0) {
      buf.readerIndex(dataLen - trailingPartialChunkLength).readBytes(newBuf,
        trailingPartialChunkLength);
    }
    buf.release();
    this.buf = newBuf;
    nextPacketOffsetInBlock += dataLen - trailingPartialChunkLength;
  }
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