HDDS-726. Ozone Client should update SCM to move the container out of allocation path in case a write transaction fails. Contributed by Shashikant Banerjee.

Author: bshashikant

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

@@ -214,14 +214,14 @@ public ContainerCommandResponseProto sendCommand(
 
   @Override
   public XceiverClientReply sendCommand(
-      ContainerCommandRequestProto request, List<UUID> excludeDns)
+      ContainerCommandRequestProto request, List<DatanodeDetails> excludeDns)
       throws IOException {
     Preconditions.checkState(HddsUtils.isReadOnly(request));
     return sendCommandWithRetry(request, excludeDns);
   }
 
   private XceiverClientReply sendCommandWithRetry(
-      ContainerCommandRequestProto request, List<UUID> excludeDns)
+      ContainerCommandRequestProto request, List<DatanodeDetails> excludeDns)
       throws IOException {
     ContainerCommandResponseProto responseProto = null;
 
@@ -231,24 +231,24 @@ private XceiverClientReply sendCommandWithRetry(
     // TODO: cache the correct leader info in here, so that any subsequent calls
     // should first go to leader
     List<DatanodeDetails> dns = pipeline.getNodes();
-    DatanodeDetails datanode = null;
     List<DatanodeDetails> healthyDns =
         excludeDns != null ? dns.stream().filter(dnId -> {
-          for (UUID excludeId : excludeDns) {
-            if (dnId.getUuid().equals(excludeId)) {
+          for (DatanodeDetails excludeId : excludeDns) {
+            if (dnId.equals(excludeId)) {
               return false;
             }
           }
           return true;
         }).collect(Collectors.toList()) : dns;
+    XceiverClientReply reply = new XceiverClientReply(null);
     for (DatanodeDetails dn : healthyDns) {
       try {
         LOG.debug("Executing command " + request + " on datanode " + dn);
         // In case the command gets retried on a 2nd datanode,
         // sendCommandAsyncCall will create a new channel and async stub
         // in case these don't exist for the specific datanode.
+        reply.addDatanode(dn);
         responseProto = sendCommandAsync(request, dn).getResponse().get();
-        datanode = dn;
         if (responseProto.getResult() == ContainerProtos.Result.SUCCESS) {
           break;
         }
@@ -264,8 +264,8 @@ private XceiverClientReply sendCommandWithRetry(
     }
 
     if (responseProto != null) {
-      return new XceiverClientReply(
-          CompletableFuture.completedFuture(responseProto), datanode.getUuid());
+      reply.setResponse(CompletableFuture.completedFuture(responseProto));
+      return reply;
     } else {
       throw new IOException(
           "Failed to execute command " + request + " on the pipeline "
@@ -382,11 +382,11 @@ private void reconnect(DatanodeDetails dn, String encodedToken)
   }
 
   @Override
-  public long watchForCommit(long index, long timeout)
+  public XceiverClientReply watchForCommit(long index, long timeout)
       throws InterruptedException, ExecutionException, TimeoutException,
       IOException {
     // there is no notion of watch for commit index in standalone pipeline
-    return 0;
+    return null;
   };
 
   public long getReplicatedMinCommitIndex() {

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

@@ -18,8 +18,8 @@
 
 package org.apache.hadoop.hdds.scm;
 
-import com.google.common.base.Preconditions;
 import org.apache.hadoop.hdds.HddsUtils;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.security.x509.SecurityConfig;
 
@@ -59,6 +59,7 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
 
 /**
  * An abstract implementation of {@link XceiverClientSpi} using Ratis.
@@ -91,7 +92,7 @@ public static XceiverClientRatis newXceiverClientRatis(
   private final GrpcTlsConfig tlsConfig;
 
   // Map to track commit index at every server
-  private final ConcurrentHashMap<String, Long> commitInfoMap;
+  private final ConcurrentHashMap<UUID, Long> commitInfoMap;
 
   // create a separate RaftClient for watchForCommit API
   private RaftClient watchClient;
@@ -118,15 +119,16 @@ private void updateCommitInfosMap(
     // of the servers
     if (commitInfoMap.isEmpty()) {
       commitInfoProtos.forEach(proto -> commitInfoMap
-          .put(proto.getServer().getAddress(), proto.getCommitIndex()));
+          .put(RatisHelper.toDatanodeId(proto.getServer()),
+              proto.getCommitIndex()));
       // In case the commit is happening 2 way, just update the commitIndex
       // for the servers which have been successfully updating the commit
       // indexes. This is important because getReplicatedMinCommitIndex()
       // should always return the min commit index out of the nodes which have
       // been replicating data successfully.
     } else {
       commitInfoProtos.forEach(proto -> commitInfoMap
-          .computeIfPresent(proto.getServer().getAddress(),
+          .computeIfPresent(RatisHelper.toDatanodeId(proto.getServer()),
               (address, index) -> {
                 index = proto.getCommitIndex();
                 return index;
@@ -218,15 +220,23 @@ public long getReplicatedMinCommitIndex() {
     return minIndex.isPresent() ? minIndex.getAsLong() : 0;
   }
 
+  private void addDatanodetoReply(UUID address, XceiverClientReply reply) {
+    DatanodeDetails.Builder builder = DatanodeDetails.newBuilder();
+    builder.setUuid(address.toString());
+    reply.addDatanode(builder.build());
+  }
+
   @Override
-  public long watchForCommit(long index, long timeout)
+  public XceiverClientReply watchForCommit(long index, long timeout)
       throws InterruptedException, ExecutionException, TimeoutException,
       IOException {
     long commitIndex = getReplicatedMinCommitIndex();
+    XceiverClientReply clientReply = new XceiverClientReply(null);
     if (commitIndex >= index) {
       // return the min commit index till which the log has been replicated to
       // all servers
-      return commitIndex;
+      clientReply.setLogIndex(commitIndex);
+      return clientReply;
     }
     LOG.debug("commit index : {} watch timeout : {}", index, timeout);
     // create a new RaftClient instance for watch request
@@ -250,26 +260,30 @@ public long watchForCommit(long index, long timeout)
       // TODO : need to remove the code to create the new RaftClient instance
       // here once the watch request bypassing sliding window in Raft Client
       // gets fixed.
-      watchClient =
-          RatisHelper.newRaftClient(rpcType, getPipeline(), retryPolicy,
+      watchClient = RatisHelper
+          .newRaftClient(rpcType, getPipeline(), retryPolicy,
               maxOutstandingRequests, tlsConfig);
       reply = watchClient
           .sendWatchAsync(index, RaftProtos.ReplicationLevel.MAJORITY_COMMITTED)
           .get(timeout, TimeUnit.MILLISECONDS);
-      Optional<RaftProtos.CommitInfoProto>
-          proto = reply.getCommitInfos().stream().min(Comparator.comparing(
-          RaftProtos.CommitInfoProto :: getCommitIndex));
-      Preconditions.checkState(proto.isPresent());
-      String address = proto.get().getServer().getAddress();
-      // since 3 way commit has failed, the updated map from now on  will
-      // only store entries for those datanodes which have had successful
-      // replication.
-      commitInfoMap.remove(address);
-      LOG.info(
-          "Could not commit " + index + " to all the nodes. Server " + address
-              + " has failed." + " Committed by majority.");
+      List<RaftProtos.CommitInfoProto> commitInfoProtoList =
+          reply.getCommitInfos().stream()
+              .filter(i -> i.getCommitIndex() < index)
+              .collect(Collectors.toList());
+      commitInfoProtoList.parallelStream().forEach(proto -> {
+        UUID address = RatisHelper.toDatanodeId(proto.getServer());
+        addDatanodetoReply(address, clientReply);
+        // since 3 way commit has failed, the updated map from now on  will
+        // only store entries for those datanodes which have had successful
+        // replication.
+        commitInfoMap.remove(address);
+        LOG.info(
+            "Could not commit " + index + " to all the nodes. Server " + address
+                + " has failed." + " Committed by majority.");
+      });
     }
-    return index;
+    clientReply.setLogIndex(index);
+    return clientReply;
   }
 
   /**
@@ -296,17 +310,28 @@ public XceiverClientReply sendCommandAsync(
                 RaftRetryFailureException raftRetryFailureException =
                     reply.getRetryFailureException();
                 if (raftRetryFailureException != null) {
+                  // in case of raft retry failure, the raft client is
+                  // not able to connect to the leader hence the pipeline
+                  // can not be used but this instance of RaftClient will close
+                  // and refreshed again. In case the client cannot connect to
+                   // leader, getClient call will fail.
+
+                  // No need to set the failed Server ID here. Ozone client
+                  // will directly exclude this pipeline in next allocate block
+                  // to SCM as in this case, it is the raft client which is not
+                  // able to connect to leader in the pipeline, though the
+                  // pipeline can still be functional.
                   throw new CompletionException(raftRetryFailureException);
                 }
                 ContainerCommandResponseProto response =
                     ContainerCommandResponseProto
                         .parseFrom(reply.getMessage().getContent());
+                UUID serverId = RatisHelper.toDatanodeId(reply.getReplierId());
                 if (response.getResult() == ContainerProtos.Result.SUCCESS) {
                   updateCommitInfosMap(reply.getCommitInfos());
-                  asyncReply.setLogIndex(reply.getLogIndex());
-                  asyncReply.setDatanode(
-                      RatisHelper.toDatanodeId(reply.getReplierId()));
                 }
+                asyncReply.setLogIndex(reply.getLogIndex());
+                addDatanodetoReply(serverId, asyncReply);
                 return response;
               } catch (InvalidProtocolBufferException e) {
                 throw new CompletionException(e);

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

@@ -42,7 +42,6 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
-import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 
 /**
@@ -290,7 +289,7 @@ private synchronized void readChunkFromContainer() throws IOException {
     XceiverClientReply reply;
     ReadChunkResponseProto readChunkResponse = null;
     final ChunkInfo chunkInfo = chunks.get(chunkIndex);
-    List<UUID> excludeDns = null;
+    List<DatanodeDetails> excludeDns = null;
     ByteString byteString;
     List<DatanodeDetails> dnList = xceiverClient.getPipeline().getNodes();
     while (true) {
@@ -334,7 +333,7 @@ private synchronized void readChunkFromContainer() throws IOException {
         if (excludeDns == null) {
           excludeDns = new ArrayList<>();
         }
-        excludeDns.add(reply.getDatanode());
+        excludeDns.addAll(reply.getDatanodes());
         if (excludeDns.size() == dnList.size()) {
           throw ioe;
         }

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

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.hdds.scm.storage;
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.scm.XceiverClientReply;
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
@@ -41,6 +42,7 @@
 import java.io.OutputStream;
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
+import java.util.Collections;
 import java.util.UUID;
 import java.util.List;
 import java.util.ArrayList;
@@ -102,14 +104,17 @@ public class BlockOutputStream extends OutputStream {
   // by all servers
   private long totalAckDataLength;
 
-  // list to hold up all putBlock futures
-  private List<CompletableFuture<ContainerProtos.ContainerCommandResponseProto>>
-      futureList;
+  // future Map to hold up all putBlock futures
+  private ConcurrentHashMap<Long,
+      CompletableFuture<ContainerProtos.ContainerCommandResponseProto>>
+      futureMap;
   // map containing mapping for putBlock logIndex to to flushedDataLength Map.
   private ConcurrentHashMap<Long, Long> commitIndex2flushedDataMap;
 
   private int currentBufferIndex;
 
+  private List<DatanodeDetails> failedServers;
+
   /**
    * Creates a new BlockOutputStream.
    *
@@ -157,10 +162,11 @@ public BlockOutputStream(BlockID blockID, String key,
     responseExecutor = Executors.newSingleThreadExecutor();
     commitIndex2flushedDataMap = new ConcurrentHashMap<>();
     totalAckDataLength = 0;
-    futureList = new ArrayList<>();
+    futureMap = new ConcurrentHashMap<>();
     totalDataFlushedLength = 0;
     currentBufferIndex = 0;
     writtenDataLength = 0;
+    failedServers = Collections.emptyList();
   }
 
   public BlockID getBlockID() {
@@ -182,6 +188,9 @@ private long computeBufferData() {
     return dataLength;
   }
 
+  public List<DatanodeDetails> getFailedServers() {
+    return failedServers;
+  }
 
   @Override
   public void write(int b) throws IOException {
@@ -299,7 +308,7 @@ private void updateFlushIndex(long index) {
       Preconditions.checkState(commitIndex2flushedDataMap.containsKey(index));
       totalAckDataLength = commitIndex2flushedDataMap.remove(index);
       LOG.debug("Total data successfully replicated: " + totalAckDataLength);
-      futureList.remove(0);
+      futureMap.remove(totalAckDataLength);
       // Flush has been committed to required servers successful.
       // just swap the bufferList head and tail after clearing.
       ByteBuffer currentBuffer = bufferList.remove(0);
@@ -320,7 +329,7 @@ private void updateFlushIndex(long index) {
   private void handleFullBuffer() throws IOException {
     try {
       checkOpen();
-      if (!futureList.isEmpty()) {
+      if (!futureMap.isEmpty()) {
         waitOnFlushFutures();
       }
     } catch (InterruptedException | ExecutionException e) {
@@ -362,9 +371,22 @@ private void adjustBuffersOnException() {
   private void watchForCommit(long commitIndex) throws IOException {
     checkOpen();
     Preconditions.checkState(!commitIndex2flushedDataMap.isEmpty());
+    long index;
     try {
-      long index =
+      XceiverClientReply reply =
           xceiverClient.watchForCommit(commitIndex, watchTimeout);
+      if (reply == null) {
+        index = 0;
+      } else {
+        List<DatanodeDetails> dnList = reply.getDatanodes();
+        if (!dnList.isEmpty()) {
+          if (failedServers.isEmpty()) {
+            failedServers = new ArrayList<>();
+          }
+          failedServers.addAll(dnList);
+        }
+        index = reply.getLogIndex();
+      }
       adjustBuffers(index);
     } catch (TimeoutException | InterruptedException | ExecutionException e) {
       LOG.warn("watchForCommit failed for index " + commitIndex, e);
@@ -392,8 +414,7 @@ ContainerCommandResponseProto> handlePartialFlush()
         try {
           validateResponse(e);
         } catch (IOException sce) {
-          future.completeExceptionally(sce);
-          return e;
+          throw new CompletionException(sce);
         }
         // if the ioException is not set, putBlock is successful
         if (ioException == null) {
@@ -422,7 +443,7 @@ ContainerCommandResponseProto> handlePartialFlush()
       throw new IOException(
           "Unexpected Storage Container Exception: " + e.toString(), e);
     }
-    futureList.add(flushFuture);
+    futureMap.put(flushPos, flushFuture);
     return flushFuture;
   }
 
@@ -516,8 +537,8 @@ public void close() throws IOException {
 
   private void waitOnFlushFutures()
       throws InterruptedException, ExecutionException {
-    CompletableFuture<Void> combinedFuture = CompletableFuture
-        .allOf(futureList.toArray(new CompletableFuture[futureList.size()]));
+    CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(
+        futureMap.values().toArray(new CompletableFuture[futureMap.size()]));
     // wait for all the transactions to complete
     combinedFuture.get();
   }
@@ -553,10 +574,10 @@ public void cleanup(boolean invalidateClient) {
     }
     xceiverClientManager = null;
     xceiverClient = null;
-    if (futureList != null) {
-      futureList.clear();
+    if (futureMap != null) {
+      futureMap.clear();
     }
-    futureList = null;
+    futureMap = null;
     if (commitIndex2flushedDataMap != null) {
       commitIndex2flushedDataMap.clear();
     }