HDDS-11665. Minor optimizations on the write path (apache#7407)

(cherry picked from commit 5663971)

Author: duongkame

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientRatis.java

@@ -282,7 +282,7 @@ private CompletableFuture<RaftClientReply> sendRequestAsync(
   // gets the minimum log index replicated to all servers
   @Override
   public long getReplicatedMinCommitIndex() {
-    return commitInfoMap.values().parallelStream()
+    return commitInfoMap.values().stream()
         .mapToLong(Long::longValue).min().orElse(0);
   }
 

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/common/ChunkBufferImplWithByteBuffer.java

@@ -25,7 +25,6 @@
 import java.util.List;
 import java.util.NoSuchElementException;
 import java.util.Objects;
-import java.util.UUID;
 import java.util.function.Function;
 
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
@@ -35,7 +34,6 @@
 final class ChunkBufferImplWithByteBuffer implements ChunkBuffer {
   private final ByteBuffer buffer;
   private final UncheckedAutoCloseable underlying;
-  private final UUID identity = UUID.randomUUID();
 
   ChunkBufferImplWithByteBuffer(ByteBuffer buffer) {
     this(buffer, null);
@@ -163,6 +161,6 @@ public int hashCode() {
   @Override
   public String toString() {
     return getClass().getSimpleName() + ":limit=" + buffer.limit()
-        + "@" + identity;
+        + "@" + Integer.toHexString(super.hashCode());
   }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/container/common/helpers/BlockData.java

@@ -254,7 +254,7 @@ public void setChunks(List<ContainerProtos.ChunkInfo> chunks) {
         size = singleChunk.getLen();
       } else {
         chunkList = chunks;
-        size = chunks.parallelStream()
+        size = chunks.stream()
             .mapToLong(ContainerProtos.ChunkInfo::getLen)
             .sum();
       }

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyOutputStream.java

@@ -303,7 +303,7 @@ private int writeToOutputStream(BlockOutputStreamEntry current,
       if (retry) {
         current.writeOnRetry(len);
       } else {
-        waitForRetryHandling(current);
+        current.waitForRetryHandling(retryHandlingCondition);
         current.write(b, off, writeLen);
         offset += writeLen;
       }
@@ -584,7 +584,7 @@ private void handleFlushOrClose(StreamAction op) throws IOException {
               blockOutputStreamEntryPool.getCurrentStreamEntry();
           if (entry != null) {
             // If the current block is to handle retries, wait until all the retries are done.
-            waitForRetryHandling(entry);
+            doInWriteLock(() -> entry.waitForRetryHandling(retryHandlingCondition));
             entry.registerCallReceived();
             try {
               handleStreamAction(entry, op);
@@ -608,10 +608,6 @@ private void handleFlushOrClose(StreamAction op) throws IOException {
     }
   }
 
-  private void waitForRetryHandling(BlockOutputStreamEntry currentEntry) throws InterruptedException {
-    doInWriteLock(() -> currentEntry.waitForRetryHandling(retryHandlingCondition));
-  }
-
   private void handleStreamAction(BlockOutputStreamEntry entry,
                                   StreamAction op) throws IOException {
     Collection<DatanodeDetails> failedServers = entry.getFailedServers();