HDDS-10983. EC Key read corruption when the replica index of container in DN mismatches

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

@@ -71,6 +71,7 @@
 import org.slf4j.LoggerFactory;
 
 import static org.apache.hadoop.hdds.HddsUtils.processForDebug;
+import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls.getContainerCommandRequestProtoBuilder;
 
 /**
  * {@link XceiverClientSpi} implementation, the standalone client.
@@ -337,8 +338,7 @@ private XceiverClientReply sendCommandWithTraceIDAndRetry(
 
     return TracingUtil.executeInNewSpan(spanName,
         () -> {
-          ContainerCommandRequestProto finalPayload =
-              ContainerCommandRequestProto.newBuilder(request)
+          ContainerCommandRequestProto finalPayload = getContainerCommandRequestProtoBuilder(request)
                   .setTraceID(TracingUtil.exportCurrentSpan()).build();
           return sendCommandWithRetry(finalPayload, validators);
         });
@@ -490,8 +490,7 @@ public XceiverClientReply sendCommandAsync(
 
     try (Scope ignored = GlobalTracer.get().activateSpan(span)) {
 
-      ContainerCommandRequestProto finalPayload =
-          ContainerCommandRequestProto.newBuilder(request)
+      ContainerCommandRequestProto finalPayload = getContainerCommandRequestProtoBuilder(request)
               .setTraceID(TracingUtil.exportCurrentSpan())
               .build();
       XceiverClientReply asyncReply =

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

@@ -62,6 +62,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls.getContainerCommandRequestProtoBuilder;
 import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls.putBlockAsync;
 
 /**
@@ -204,7 +205,7 @@ private DataStreamOutput setupStream(Pipeline pipeline) throws IOException {
     //  it or remove it completely if possible
     String id = pipeline.getFirstNode().getUuidString();
     ContainerProtos.ContainerCommandRequestProto.Builder builder =
-        ContainerProtos.ContainerCommandRequestProto.newBuilder()
+        getContainerCommandRequestProtoBuilder()
             .setCmdType(ContainerProtos.Type.StreamInit)
             .setContainerID(blockID.get().getContainerID())
             .setDatanodeUuid(id).setWriteChunk(writeChunkRequest);

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

@@ -26,6 +26,8 @@
 import org.apache.hadoop.fs.CanUnbuffer;
 import org.apache.hadoop.fs.Seekable;
 import org.apache.hadoop.hdds.client.BlockID;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
@@ -60,6 +62,7 @@ public class ChunkInputStream extends InputStream
   private final ChunkInfo chunkInfo;
   private final long length;
   private final BlockID blockID;
+  private ContainerProtos.DatanodeBlockID datanodeBlockID;
   private final XceiverClientFactory xceiverClientFactory;
   private XceiverClientSpi xceiverClient;
   private final Supplier<Pipeline> pipelineSupplier;
@@ -290,13 +293,27 @@ protected synchronized void releaseClient() {
     }
   }
 
+  /**
+   * Updates DatanodeBlockId which based on blockId.
+   */
+  private void updateDatanodeBlockId() throws IOException {
+    DatanodeDetails closestNode = pipelineSupplier.get().getClosestNode();
+    int replicaIdx = pipelineSupplier.get().getReplicaIndex(closestNode);
+    ContainerProtos.DatanodeBlockID.Builder builder = blockID.getDatanodeBlockIDProtobufBuilder();
+    if (replicaIdx > 0) {
+      builder.setReplicaIndex(replicaIdx);
+    }
+    datanodeBlockID = builder.build();
+  }
+
   /**
    * Acquire new client if previous one was released.
    */
   protected synchronized void acquireClient() throws IOException {
     if (xceiverClientFactory != null && xceiverClient == null) {
       xceiverClient = xceiverClientFactory.acquireClientForReadData(
           pipelineSupplier.get());
+      updateDatanodeBlockId();
     }
   }
 
@@ -423,7 +440,7 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
 
     ReadChunkResponseProto readChunkResponse =
         ContainerProtocolCalls.readChunk(xceiverClient,
-            readChunkInfo, blockID, validators, tokenSupplier.get());
+            readChunkInfo, datanodeBlockID, validators, tokenSupplier.get());
 
     if (readChunkResponse.hasData()) {
       return readChunkResponse.getData().asReadOnlyByteBufferList()

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

@@ -176,8 +176,8 @@ protected BlockExtendedInputStream getOrOpenStream(int locationIndex) {
           .setReplicationConfig(StandaloneReplicationConfig.getInstance(
               HddsProtos.ReplicationFactor.ONE))
           .setNodes(Arrays.asList(dataLocation))
-          .setId(PipelineID.valueOf(dataLocation.getUuid())).setReplicaIndexes(
-              ImmutableMap.of(dataLocation, locationIndex + 1))
+          .setId(PipelineID.valueOf(dataLocation.getUuid()))
+          .setReplicaIndexes(ImmutableMap.of(dataLocation, locationIndex + 1))
           .setState(Pipeline.PipelineState.CLOSED)
           .build();
 
@@ -228,6 +228,7 @@ protected Function<BlockID, BlockLocationInfo> ecPipelineRefreshFunction(
                   HddsProtos.ReplicationFactor.ONE))
           .setNodes(Collections.singletonList(curIndexNode))
           .setId(PipelineID.randomId())
+          .setReplicaIndexes(Collections.singletonMap(curIndexNode, replicaIndex))
           .setState(Pipeline.PipelineState.CLOSED)
           .build();
       blockLocationInfo.setPipeline(pipeline);

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java

@@ -23,30 +23,41 @@
 import java.util.Objects;
 
 /**
- * BlockID of Ozone (containerID + localID + blockCommitSequenceId).
+ * BlockID of Ozone (containerID + localID + blockCommitSequenceId + replicaIndex).
  */
 
 public class BlockID {
 
   private final ContainerBlockID containerBlockID;
   private long blockCommitSequenceId;
+  // null value when not set with private constructor.(This is to avoid confusion of replica index 0 & null value).
+  // This value would be only set when deserializing from ContainerProtos.DatanodeBlockID or copying from another
+  // BlockID object.
+  private Integer replicaIndex;
 
   public BlockID(long containerID, long localID) {
-    this(containerID, localID, 0);
+    this(containerID, localID, 0, null);
   }
 
-  private BlockID(long containerID, long localID, long bcsID) {
+  private BlockID(long containerID, long localID, long bcsID, Integer repIndex) {
     containerBlockID = new ContainerBlockID(containerID, localID);
     blockCommitSequenceId = bcsID;
+    this.replicaIndex = repIndex;
+  }
+
+  public BlockID(BlockID blockID) {
+    this(blockID.getContainerID(), blockID.getLocalID(), blockID.getBlockCommitSequenceId(),
+        blockID.getReplicaIndex());
   }
 
   public BlockID(ContainerBlockID containerBlockID) {
-    this(containerBlockID, 0);
+    this(containerBlockID, 0, null);
   }
 
-  private BlockID(ContainerBlockID containerBlockID, long bcsId) {
+  private BlockID(ContainerBlockID containerBlockID, long bcsId, Integer repIndex) {
     this.containerBlockID = containerBlockID;
     blockCommitSequenceId = bcsId;
+    this.replicaIndex = repIndex;
   }
 
   public long getContainerID() {
@@ -65,6 +76,15 @@ public void setBlockCommitSequenceId(long blockCommitSequenceId) {
     this.blockCommitSequenceId = blockCommitSequenceId;
   }
 
+  // Can return a null value in case it is not set.
+  public Integer getReplicaIndex() {
+    return replicaIndex;
+  }
+
+  public void setReplicaIndex(Integer replicaIndex) {
+    this.replicaIndex = replicaIndex;
+  }
+
   public ContainerBlockID getContainerBlockID() {
     return containerBlockID;
   }
@@ -79,21 +99,32 @@ public String toString() {
   public void appendTo(StringBuilder sb) {
     containerBlockID.appendTo(sb);
     sb.append(" bcsId: ").append(blockCommitSequenceId);
+    sb.append(" replicaIndex: ").append(replicaIndex);
   }
 
   @JsonIgnore
   public ContainerProtos.DatanodeBlockID getDatanodeBlockIDProtobuf() {
+    ContainerProtos.DatanodeBlockID.Builder blockID = getDatanodeBlockIDProtobufBuilder();
+    if (replicaIndex != null) {
+      blockID.setReplicaIndex(replicaIndex);
+    }
+    return blockID.build();
+  }
+
+  @JsonIgnore
+  public ContainerProtos.DatanodeBlockID.Builder getDatanodeBlockIDProtobufBuilder() {
     return ContainerProtos.DatanodeBlockID.newBuilder().
         setContainerID(containerBlockID.getContainerID())
         .setLocalID(containerBlockID.getLocalID())
-        .setBlockCommitSequenceId(blockCommitSequenceId).build();
+        .setBlockCommitSequenceId(blockCommitSequenceId);
   }
 
   @JsonIgnore
-  public static BlockID getFromProtobuf(
-      ContainerProtos.DatanodeBlockID blockID) {
+  public static BlockID getFromProtobuf(ContainerProtos.DatanodeBlockID blockID) {
     return new BlockID(blockID.getContainerID(),
-        blockID.getLocalID(), blockID.getBlockCommitSequenceId());
+        blockID.getLocalID(),
+        blockID.getBlockCommitSequenceId(),
+        blockID.hasReplicaIndex() ? blockID.getReplicaIndex() : null);
   }
 
   @JsonIgnore
@@ -107,7 +138,7 @@ public HddsProtos.BlockID getProtobuf() {
   public static BlockID getFromProtobuf(HddsProtos.BlockID blockID) {
     return new BlockID(
         ContainerBlockID.getFromProtobuf(blockID.getContainerBlockID()),
-        blockID.getBlockCommitSequenceId());
+        blockID.getBlockCommitSequenceId(), null);
   }
 
   @Override
@@ -120,13 +151,13 @@ public boolean equals(Object o) {
     }
     BlockID blockID = (BlockID) o;
     return containerBlockID.equals(blockID.getContainerBlockID())
-        && blockCommitSequenceId == blockID.getBlockCommitSequenceId();
+        && blockCommitSequenceId == blockID.getBlockCommitSequenceId()
+        && Objects.equals(replicaIndex, blockID.getReplicaIndex());
   }
 
   @Override
   public int hashCode() {
-    return Objects
-        .hash(containerBlockID.getContainerID(), containerBlockID.getLocalID(),
-            blockCommitSequenceId);
+    return Objects.hash(containerBlockID.getContainerID(), containerBlockID.getLocalID(),
+        blockCommitSequenceId, replicaIndex);
   }
 }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ratis/ContainerCommandRequestMessage.java

@@ -32,15 +32,16 @@
 import org.apache.ratis.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
 import org.apache.ratis.util.JavaUtils;
 
+import static org.apache.hadoop.hdds.scm.storage.ContainerProtocolCalls.getContainerCommandRequestProtoBuilder;
+
 /**
  * Implementing the {@link Message} interface
  * for {@link ContainerCommandRequestProto}.
  */
 public final class ContainerCommandRequestMessage implements Message {
   public static ContainerCommandRequestMessage toMessage(
       ContainerCommandRequestProto request, String traceId) {
-    final ContainerCommandRequestProto.Builder b
-        = ContainerCommandRequestProto.newBuilder(request);
+    final ContainerCommandRequestProto.Builder b = getContainerCommandRequestProtoBuilder(request);
     if (traceId != null) {
       b.setTraceID(traceId);
     }