apache · szetszwo · Sep 1, 2026
diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockInputStream.java
@@ -35,6 +35,7 @@
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ChunkInfo;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandResponseProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.GetBlockRequestProto;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.GetBlockResponseProto;
 import org.apache.hadoop.hdds.scm.OzoneClientConfig;
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
@@ -268,24 +269,15 @@ protected BlockData getBlockDataUsingSCClient() throws IOException {
           blockID.getContainerID());
     }
 
-    DatanodeBlockID.Builder blkIDBuilder =
-        DatanodeBlockID.newBuilder().setContainerID(blockID.getContainerID())
-            .setLocalID(blockID.getLocalID())
-            .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId());
-
     int replicaIndex = pipeline.getReplicaIndex(xceiverClientShortCircuit.getDn());
-    if (replicaIndex > 0) {
-      blkIDBuilder.setReplicaIndex(replicaIndex);
-    }
-    DatanodeBlockID datanodeBlockID = blkIDBuilder.build();
-    ContainerProtos.GetBlockRequestProto.Builder readBlockRequest =
-        ContainerProtos.GetBlockRequestProto.newBuilder().setBlockID(datanodeBlockID)
-            .setRequestShortCircuitAccess(true);
+    final GetBlockRequestProto.Builder getBlockRequest = GetBlockRequestProto.newBuilder()
+        .setBlockID(blockID.getDatanodeBlockIDProtobufBuilder(replicaIndex))
+        .setRequestShortCircuitAccess(true);
     ContainerProtos.ContainerCommandRequestProto.Builder builder =
         ContainerProtos.ContainerCommandRequestProto.newBuilder()
             .setCmdType(ContainerProtos.Type.GetBlock)
-            .setContainerID(datanodeBlockID.getContainerID())
-            .setGetBlock(readBlockRequest)
+            .setContainerID(blockID.getContainerID())
+            .setGetBlock(getBlockRequest)
             .setClientId(xceiverClientShortCircuit.getClientId())
             .setCallId(xceiverClientShortCircuit.getCallId());
     if (tokenRef.get() != null) {
@@ -294,13 +286,12 @@ protected BlockData getBlockDataUsingSCClient() throws IOException {
     GetBlockResponseProto response = ContainerProtocolCalls.getBlock(xceiverClientShortCircuit,
         VALIDATORS, builder, xceiverClientShortCircuit.getDn());
 
-    blockFileInputStream = xceiverClientShortCircuit.getFileInputStream(
-        builder.getCallId(), datanodeBlockID.getLocalID());
+    blockFileInputStream = xceiverClientShortCircuit.getFileInputStream(builder.getCallId(), blockID.getLocalID());
     if (blockFileInputStream == null) {
-      throw new IOException("Failed to get file InputStream for block " + datanodeBlockID);
+      throw new IOException("Failed to get file InputStream for block " + blockID);
     } else {
       if (LOG.isDebugEnabled()) {
-        LOG.debug("Get the FileInputStream of block {}", datanodeBlockID);
+        LOG.debug("Get the FileInputStream of block {}", blockID);
       }
     }
     return response.getBlockData();
@@ -316,7 +307,7 @@ protected BlockData getBlockDataUsingGRPCClient() throws IOException {
     }
 
     GetBlockResponseProto response = ContainerProtocolCalls.getBlock(
-        xceiverClientGrpc, VALIDATORS, blockID, tokenRef.get(), pipeline.getReplicaIndexes());
+        xceiverClientGrpc, VALIDATORS, blockID, tokenRef.get(), pipeline);
     return response.getBlockData();
   }
 

diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockOutputStream.java
@@ -187,16 +187,9 @@ public BlockOutputStream(
     KeyValue keyValue =
         KeyValue.newBuilder().setKey("TYPE").setValue("KEY").build();
 
-    ContainerProtos.DatanodeBlockID.Builder blkIDBuilder =
-        ContainerProtos.DatanodeBlockID.newBuilder()
-            .setContainerID(blockID.getContainerID())
-            .setLocalID(blockID.getLocalID())
-            .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId());
-    if (replicationIndex > 0) {
-      blkIDBuilder.setReplicaIndex(replicationIndex);
-    }
-    this.containerBlockData = BlockData.newBuilder().setBlockID(
-        blkIDBuilder.build()).addMetadata(keyValue);
+    this.containerBlockData = BlockData.newBuilder()
+        .setBlockID(blockID.getDatanodeBlockIDProtobufBuilder(replicationIndex))
+        .addMetadata(keyValue);
     this.pipeline = pipeline;
     // tell DataNode I will send incremental chunk list
     this.supportIncrementalChunkList = canEnableIncrementalChunkList();

diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/ChunkInputStream.java
@@ -297,11 +297,7 @@ protected synchronized void releaseClient() {
   private void updateDatanodeBlockId(Pipeline pipeline) throws IOException {
     DatanodeDetails closestNode = pipeline.getClosestNode();
     int replicaIdx = pipeline.getReplicaIndex(closestNode);
-    ContainerProtos.DatanodeBlockID.Builder builder = blockID.getDatanodeBlockIDProtobufBuilder();
-    if (replicaIdx > 0) {
-      builder.setReplicaIndex(replicaIdx);
-    }
-    datanodeBlockID = builder.build();
+    datanodeBlockID = blockID.getDatanodeBlockIDProtobufBuilder(replicaIdx).build();
   }
 
   /**

diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/client/BlockID.java
@@ -19,7 +19,7 @@
 
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import java.util.Objects;
-import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 
 /**
@@ -98,24 +98,24 @@ public void appendTo(StringBuilder sb) {
   }
 
   @JsonIgnore
-  public ContainerProtos.DatanodeBlockID getDatanodeBlockIDProtobuf() {
-    ContainerProtos.DatanodeBlockID.Builder blockID = getDatanodeBlockIDProtobufBuilder();
-    if (replicaIndex != null) {
-      blockID.setReplicaIndex(replicaIndex);
-    }
-    return blockID.build();
+  public DatanodeBlockID getDatanodeBlockIDProtobuf() {
+    return getDatanodeBlockIDProtobufBuilder(replicaIndex).build();
   }
 
   @JsonIgnore
-  public ContainerProtos.DatanodeBlockID.Builder getDatanodeBlockIDProtobufBuilder() {
-    return ContainerProtos.DatanodeBlockID.newBuilder().
-        setContainerID(containerBlockID.getContainerID())
+  public DatanodeBlockID.Builder getDatanodeBlockIDProtobufBuilder(Integer replicaIdx) {
+    final DatanodeBlockID.Builder b = DatanodeBlockID.newBuilder()
+        .setContainerID(containerBlockID.getContainerID())
         .setLocalID(containerBlockID.getLocalID())
         .setBlockCommitSequenceId(blockCommitSequenceId);
+    if (replicaIdx != null && replicaIdx > 0) {
+      b.setReplicaIndex(replicaIdx);
+    }
+    return b;
   }
 
   @JsonIgnore
-  public static BlockID getFromProtobuf(ContainerProtos.DatanodeBlockID blockID) {
+  public static BlockID getFromProtobuf(DatanodeBlockID blockID) {
     return new BlockID(blockID.getContainerID(),
         blockID.getLocalID(),
         blockID.getBlockCommitSequenceId(),

diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
@@ -33,8 +33,6 @@
 import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
-import java.util.function.Function;
-import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.client.ReplicatedReplicationConfig;
@@ -240,10 +238,27 @@ public int getReplicaIndex(DatanodeDetails dn) {
   }
 
   /**
-   * Get the replicaIndex Map.
+   * @param fromIndex the replica index starting from (inclusive)
+   * @param toIndex the replica index starting to (exclusive)
+   * @return true iff this pipeline contain all replica indexes within the given range.
    */
-  public Map<DatanodeDetails, Integer> getReplicaIndexes() {
-    return this.getNodes().stream().collect(Collectors.toMap(Function.identity(), this::getReplicaIndex));
+  public boolean containsAllReplicaIndexes(int fromIndex, int toIndex) {
+    final boolean[] contains = new boolean[toIndex - fromIndex];
+    for (int r : replicaIndexes.values()) {
+      if (r >= fromIndex && r < toIndex) {
+        contains[r - fromIndex] = true;
+      }
+    }
+    for (boolean contain : contains) {
+      if (!contain) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  public Map<DatanodeDetails, Integer> getReplicaIndexesForTesting() {
+    return Collections.unmodifiableMap(replicaIndexes);
   }
 
   /**

diff --git a/...-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java b/...-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java
@@ -188,7 +188,7 @@ static <T> T tryEachDatanode(Pipeline pipeline,
    */
   public static GetBlockResponseProto getBlock(XceiverClientSpi xceiverClient,
       List<Validator> validators, BlockID blockID, Token<? extends TokenIdentifier> token,
-      Map<DatanodeDetails, Integer> replicaIndexes) throws IOException {
+      Pipeline pipeline) throws IOException {
     ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto
         .newBuilder()
         .setCmdType(Type.GetBlock)
@@ -198,7 +198,7 @@ public static GetBlockResponseProto getBlock(XceiverClientSpi xceiverClient,
     }
 
     return tryEachDatanode(xceiverClient.getPipeline(),
-        d -> getBlock(xceiverClient, validators, builder, blockID, d, replicaIndexes),
+        d -> getBlock(xceiverClient, validators, builder, blockID, d, pipeline),
         d -> toErrorMessage(blockID, d));
   }
 
@@ -209,8 +209,8 @@ static String toErrorMessage(BlockID blockId, DatanodeDetails d) {
 
   public static GetBlockResponseProto getBlock(XceiverClientSpi xceiverClient,
       BlockID datanodeBlockID,
-      Token<? extends TokenIdentifier> token, Map<DatanodeDetails, Integer> replicaIndexes) throws IOException {
-    return getBlock(xceiverClient, getValidatorList(), datanodeBlockID, token, replicaIndexes);
+      Token<? extends TokenIdentifier> token, Pipeline pipeline) throws IOException {
+    return getBlock(xceiverClient, getValidatorList(), datanodeBlockID, token, pipeline);
   }
 
   /**
@@ -221,7 +221,6 @@ public static GetBlockResponseProto getBlock(XceiverClientSpi xceiverClient,
    * @param blockID blockID to identify container
    * @param token a token for this block (may be null)
    * @param datanode datanode to query
-   * @param replicaIndexes replica indexes for EC pipelines
    * @return container protocol get block response
    * @throws IOException if there is an I/O error while performing the call
    */
@@ -230,31 +229,27 @@ public static GetBlockResponseProto getBlockFromDatanode(
       BlockID blockID,
       Token<? extends TokenIdentifier> token,
       DatanodeDetails datanode,
-      Map<DatanodeDetails, Integer> replicaIndexes) throws IOException {
+      Pipeline pipeline) throws IOException {
     ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto
         .newBuilder()
         .setCmdType(Type.GetBlock)
         .setContainerID(blockID.getContainerID());
     if (token != null) {
       builder.setEncodedToken(token.encodeToUrlString());
     }
-    return getBlock(xceiverClient, getValidatorList(), builder, blockID, datanode,
-        replicaIndexes);
+    return getBlock(xceiverClient, getValidatorList(), builder, blockID, datanode, pipeline);
   }
 
   private static GetBlockResponseProto getBlock(XceiverClientSpi xceiverClient,
       List<Validator> validators,
       ContainerCommandRequestProto.Builder builder, BlockID blockID,
-      DatanodeDetails datanode, Map<DatanodeDetails, Integer> replicaIndexes) throws IOException {
+      DatanodeDetails datanode, Pipeline pipeline) throws IOException {
     String traceId = TracingUtil.exportCurrentSpan();
     if (traceId != null) {
       builder.setTraceID(traceId);
     }
-    final DatanodeBlockID.Builder datanodeBlockID = blockID.getDatanodeBlockIDProtobufBuilder();
-    int replicaIndex = replicaIndexes.getOrDefault(datanode, 0);
-    if (replicaIndex > 0) {
-      datanodeBlockID.setReplicaIndex(replicaIndex);
-    }
+    final int replicaIndex = pipeline.getReplicaIndex(datanode);
+    final DatanodeBlockID.Builder datanodeBlockID = blockID.getDatanodeBlockIDProtobufBuilder(replicaIndex);
     final GetBlockRequestProto.Builder readBlockRequest = GetBlockRequestProto.newBuilder()
         .setBlockID(datanodeBlockID.build());
     final ContainerCommandRequestProto request = builder
@@ -510,16 +505,10 @@ public static XceiverClientReply writeChunkAsync(
       boolean containerAutoCreate)
       throws IOException, ExecutionException, InterruptedException {
 
-    WriteChunkRequestProto.Builder writeChunkRequest =
-        WriteChunkRequestProto.newBuilder()
-            .setBlockID(DatanodeBlockID.newBuilder()
-                .setContainerID(blockID.getContainerID())
-                .setLocalID(blockID.getLocalID())
-                .setBlockCommitSequenceId(blockID.getBlockCommitSequenceId())
-                .setReplicaIndex(replicationIndex)
-                .build())
-            .setChunkData(chunk)
-            .setData(data);
+    final WriteChunkRequestProto.Builder writeChunkRequest = WriteChunkRequestProto.newBuilder()
+        .setBlockID(blockID.getDatanodeBlockIDProtobufBuilder(replicationIndex))
+        .setChunkData(chunk)
+        .setData(data);
     if (blockData != null) {
       PutBlockRequestProto.Builder createBlockRequest =
           PutBlockRequestProto.newBuilder()
@@ -931,41 +920,6 @@ public static List<Validator> toValidatorList(Validator validator) {
     return Collections.unmodifiableList(validators);
   }
 
-  public static HashMap<DatanodeDetails, GetBlockResponseProto>
-      getBlockFromAllNodes(
-      XceiverClientSpi xceiverClient,
-      DatanodeBlockID datanodeBlockID,
-      Token<OzoneBlockTokenIdentifier> token)
-      throws IOException, InterruptedException {
-    GetBlockRequestProto.Builder readBlockRequest = GetBlockRequestProto
-            .newBuilder()
-            .setBlockID(datanodeBlockID);
-    HashMap<DatanodeDetails, GetBlockResponseProto> datanodeToResponseMap
-            = new HashMap<>();
-    String id = xceiverClient.getPipeline().getFirstNode().getUuidString();
-    ContainerCommandRequestProto.Builder builder = ContainerCommandRequestProto
-        .newBuilder()
-        .setCmdType(Type.GetBlock)
-        .setContainerID(datanodeBlockID.getContainerID())
-        .setDatanodeUuid(id)
-        .setGetBlock(readBlockRequest);
-    if (token != null) {
-      builder.setEncodedToken(token.encodeToUrlString());
-    }
-    String traceId = TracingUtil.exportCurrentSpan();
-    if (traceId != null) {
-      builder.setTraceID(traceId);
-    }
-    ContainerCommandRequestProto request = builder.build();
-    Map<DatanodeDetails, ContainerCommandResponseProto> responses =
-            xceiverClient.sendCommandOnAllNodes(request);
-    for (Map.Entry<DatanodeDetails, ContainerCommandResponseProto> entry:
-           responses.entrySet()) {
-      datanodeToResponseMap.put(entry.getKey(), entry.getValue().getGetBlock());
-    }
-    return datanodeToResponseMap;
-  }
-
   public static HashMap<DatanodeDetails, ReadContainerResponseProto>
       readContainerFromAllNodes(XceiverClientSpi client, long containerID,
       String encodedToken) throws IOException, InterruptedException {
@@ -1000,12 +954,12 @@ public static ContainerCommandRequestProto buildReadBlockCommandProto(
       Token<? extends TokenIdentifier> token, Pipeline pipeline)
       throws IOException {
     final DatanodeDetails datanode = pipeline.getClosestNode();
-    final DatanodeBlockID datanodeBlockID = getDatanodeBlockID(blockID, datanode, pipeline.getReplicaIndexes());
+    final int replicaIndex = pipeline.getReplicaIndex(datanode);
     final ReadBlockRequestProto.Builder readBlockRequest = ReadBlockRequestProto.newBuilder()
         .setOffset(offset)
         .setLength(length)
         .setResponseDataSize(responseDataSize)
-        .setBlockID(datanodeBlockID);
+        .setBlockID(blockID.getDatanodeBlockIDProtobufBuilder(replicaIndex));
     final ContainerCommandRequestProto.Builder builder =
         ContainerCommandRequestProto.newBuilder().setCmdType(Type.ReadBlock)
             .setContainerID(blockID.getContainerID());
@@ -1017,14 +971,4 @@ public static ContainerCommandRequestProto buildReadBlockCommandProto(
         .setReadBlock(readBlockRequest)
         .build();
   }
-
-  static DatanodeBlockID getDatanodeBlockID(BlockID blockID, DatanodeDetails datanode,
-      Map<DatanodeDetails, Integer> replicaIndexes) {
-    final DatanodeBlockID.Builder b = blockID.getDatanodeBlockIDProtobufBuilder();
-    final int replicaIndex = replicaIndexes.getOrDefault(datanode, 0);
-    if (replicaIndex > 0) {
-      b.setReplicaIndex(replicaIndex);
-    }
-    return b.build();
-  }
 }
diff --git a/.../apache/hadoop/ozone/container/keyvalue/TestContainerReconciliationWithMockDatanodes.java b/.../apache/hadoop/ozone/container/keyvalue/TestContainerReconciliationWithMockDatanodes.java
@@ -34,7 +34,6 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
-import static org.mockito.ArgumentMatchers.anyMap;
 import static org.mockito.Mockito.doThrow;
 import  static org.mockito.Mockito.spy;
 
@@ -390,7 +389,7 @@ private static void mockContainerProtocolCalls(FailureLocation failureLocation,
         });
 
     // Mock getBlock
-    containerProtocolMock.when(() -> ContainerProtocolCalls.getBlock(any(), any(), any(), any(), anyMap()))
+    containerProtocolMock.when(() -> ContainerProtocolCalls.getBlock(any(), any(), any(), any(), any()))
         .thenAnswer(inv -> {
           XceiverClientSpi xceiverClientSpi = inv.getArgument(0);
           BlockID blockID = inv.getArgument(2);

diff --git a/...li-debug/src/main/java/org/apache/hadoop/ozone/debug/replicas/BlockExistenceVerifier.java b/...li-debug/src/main/java/org/apache/hadoop/ozone/debug/replicas/BlockExistenceVerifier.java
@@ -58,7 +58,7 @@ public BlockVerificationResult verifyBlock(DatanodeDetails datanode, OmKeyLocati
           client,
           keyLocation.getBlockID(),
           keyLocation.getToken(),
-          pipeline.getReplicaIndexes()
+          pipeline
       );
 
       boolean hasBlock = response != null && response.hasBlockData();

diff --git a/...cli-debug/src/main/java/org/apache/hadoop/ozone/debug/replicas/chunk/ChunkKeyHandler.java b/...cli-debug/src/main/java/org/apache/hadoop/ozone/debug/replicas/chunk/ChunkKeyHandler.java
@@ -131,7 +131,7 @@ protected void execute(OzoneClient client, OzoneAddress address)
                         keyLocation.getBlockID(),
                         keyLocation.getToken(),
                         datanodeDetails,
-                        pipeline.getReplicaIndexes());
+                        pipeline);
 
                 if (blockResponse == null || !blockResponse.hasBlockData()) {
                   System.err.printf("GetBlock call failed on %s datanode and %s block.%n",

diff --git a/...ne/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ECFileChecksumHelper.java b/...ne/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ECFileChecksumHelper.java
@@ -123,7 +123,7 @@ protected List<ContainerProtos.ChunkInfo> getChunkInfos(OmKeyLocationInfo
       xceiverClientSpi = getXceiverClientFactory().acquireClientForReadData(pipeline);
 
       ContainerProtos.GetBlockResponseProto response = ContainerProtocolCalls
-          .getBlock(xceiverClientSpi, blockID, token, pipeline.getReplicaIndexes());
+          .getBlock(xceiverClientSpi, blockID, token, pipeline);
 
       chunks = response.getBlockData().getChunksList();
     } finally {