private void completed(Channel channel) {
    for (Iterator<Callback> iter = waitingAckQueue.iterator(); iter.hasNext();) {
      Callback c = iter.next();
      // if the current unfinished replicas does not contain us then it means that we have already
      // acked this one, let's iterate to find the one we have not acked yet.
      if (c.unfinishedReplicas.remove(channel.id())) {
        long current = EnvironmentEdgeManager.currentTime();
        streamSlowMonitor.checkProcessTimeAndSpeed(datanodeInfoMap.get(channel), c.packetDataLen,
          current - c.flushTimestamp, c.lastAckTimestamp, c.unfinishedReplicas.size());
        c.lastAckTimestamp = current;
        if (c.unfinishedReplicas.isEmpty()) {
          // we need to remove first before complete the future. It is possible that after we
          // complete the future the upper layer will call close immediately before we remove the
          // entry from waitingAckQueue and lead to an IllegalStateException. And also set the
          // ackedBlockLength first otherwise we may use a wrong length to commit the block. This
          // may lead to multiple remove and assign but is OK. The semantic of iter.remove is
          // removing the entry returned by calling previous next, so if the entry has already been
          // removed then it is a no-op, and for the assign, the values are the same so no problem.
          iter.remove();
          ackedBlockLength = c.ackedLength;
          // the future.complete check is to confirm that we are the only one who grabbed the work,
          // otherwise just give up and return.
          if (c.future.complete(c.ackedLength)) {
            // also wake up flush requests which have the same length.
            while (iter.hasNext()) {
              Callback maybeDummyCb = iter.next();
              if (maybeDummyCb.ackedLength == c.ackedLength) {
                iter.remove();
                maybeDummyCb.future.complete(c.ackedLength);
              } else {
                break;
              }
            }
          }
        }
        return;
      }
    }
  }
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