HDDS-15680. Cache per-block EC checksum pipeline within a file

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/checksum/ECFileChecksumHelper.java

@@ -49,6 +49,11 @@
  */
 public class ECFileChecksumHelper extends BaseFileChecksumHelper {
 
+  // Blocks in the same EC placement group share identical node sets, so the
+  // STANDALONE checksum pipeline is the same for every block. Cache it to
+  // avoid rebuilding it (node filtering, UUID computation, object allocation) per block.
+  private final Map<PipelineID, Pipeline> checksumPipelineCache = new HashMap<>();
+
   public ECFileChecksumHelper(OzoneVolume volume, OzoneBucket bucket,
       String keyName, long length, OzoneClientConfig.ChecksumCombineMode
       checksumCombineMode, ClientProtocol rpcClient, OmKeyInfo keyInfo)
@@ -64,74 +69,69 @@ protected AbstractBlockChecksumComputer getBlockChecksumComputer(List<ContainerP
   }
 
   @Override
-  protected List<ContainerProtos.ChunkInfo> getChunkInfos(OmKeyLocationInfo
-                                                              keyLocationInfo) throws IOException {
-    // To read an EC block, we create a STANDALONE pipeline that contains the
-    // single location for the block index we want to read. The EC blocks are
-    // indexed from 1 to N, however the data locations are stored in the
-    // dataLocations array indexed from zero.
+  protected List<ContainerProtos.ChunkInfo> getChunkInfos(OmKeyLocationInfo keyLocationInfo) throws IOException {
     Token<OzoneBlockTokenIdentifier> token = keyLocationInfo.getToken();
     BlockID blockID = keyLocationInfo.getBlockID();
+    Pipeline ecPipeline = keyLocationInfo.getPipeline();
+    Pipeline checksumPipeline = checksumPipelineCache.computeIfAbsent(
+        ecPipeline.getId(), id -> buildChecksumPipeline(ecPipeline));
 
-    Pipeline pipeline = keyLocationInfo.getPipeline();
+    List<ContainerProtos.ChunkInfo> chunks;
+    XceiverClientSpi xceiverClientSpi = null;
+    try {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Initializing BlockInputStream for get key to access {}",
+            blockID.getContainerID());
+      }
+      xceiverClientSpi = getXceiverClientFactory().acquireClientForReadData(checksumPipeline);
 
-    List<DatanodeDetails> nodes = new ArrayList<>();
-    Map<DatanodeDetails, Integer> selectedReplicaIndexes = new HashMap<>();
-    ECReplicationConfig repConfig = (ECReplicationConfig)
-        pipeline.getReplicationConfig();
+      ContainerProtos.GetBlockResponseProto response = ContainerProtocolCalls
+          .getBlock(xceiverClientSpi, blockID, token, checksumPipeline.getReplicaIndexes());
 
-    for (DatanodeDetails dn : pipeline.getNodes()) {
-      int replicaIndex = pipeline.getReplicaIndex(dn);
+      chunks = response.getBlockData().getChunksList();
+    } finally {
+      if (xceiverClientSpi != null) {
+        getXceiverClientFactory().releaseClientForReadData(xceiverClientSpi, false);
+      }
+    }
+    return chunks;
+  }
+
+  // To read an EC block we need a STANDALONE pipeline containing only replica
+  // index 1 (which holds stripe checksums) and the parity nodes. Blocks in the
+  // same placement group share the same node set, so this result is cached by
+  // the caller and built at most once per placement group per file.
+  //
+  // The pipeline ID is derived deterministically from the sorted node UUIDs so
+  // that XceiverClientManager can reuse the same gRPC connection across files
+  // that land on the same EC placement group (cross-file reuse).
+  // Pipeline.newBuilder() is used instead of toBuilder() because toBuilder()
+  // copies the EC nodeStatus; the size mismatch when setNodes() is called
+  // triggers PipelineID.randomId() and defeats the deterministic ID.
+  private Pipeline buildChecksumPipeline(Pipeline ecPipeline) {
+    ECReplicationConfig repConfig = (ECReplicationConfig) ecPipeline.getReplicationConfig();
+    List<DatanodeDetails> nodes = new ArrayList<>();
+    Map<DatanodeDetails, Integer> replicaIndexes = new HashMap<>();
+    for (DatanodeDetails dn : ecPipeline.getNodes()) {
+      int replicaIndex = ecPipeline.getReplicaIndex(dn);
       if (replicaIndex == 1 || replicaIndex > repConfig.getData()) {
-        // The stripe checksum we need to calculate checksums is only stored on
-        // replica_index = 1 and all the parity nodes.
         nodes.add(dn);
-        selectedReplicaIndexes.put(dn, replicaIndex);
+        replicaIndexes.put(dn, replicaIndex);
       }
     }
-
-    // Build a deterministic pipeline ID from the sorted node UUIDs so that
-    // XceiverClientManager can cache and reuse the gRPC connection across files
-    // that share the same EC placement group (avoids a new connection per file).
     String nodeKey = nodes.stream()
         .map(DatanodeDetails::getUuidString)
         .sorted()
         .collect(Collectors.joining(","));
     PipelineID deterministicId = PipelineID.valueOf(
         UUID.nameUUIDFromBytes(nodeKey.getBytes(StandardCharsets.UTF_8)));
-
-    // Use Pipeline.newBuilder() (not toBuilder()) so that nodeStatus starts null.
-    // toBuilder() would copy the 5-node EC nodeStatus, causing setNodes(3 nodes)
-    // to detect the size mismatch and call PipelineID.randomId() -> SecureRandom
-    // even though setId(deterministicId) immediately overrides it.
-    pipeline = Pipeline.newBuilder()
+    return Pipeline.newBuilder()
         .setId(deterministicId)
         .setReplicationConfig(StandaloneReplicationConfig
             .getInstance(HddsProtos.ReplicationFactor.THREE))
-        .setState(pipeline.getPipelineState())
+        .setState(ecPipeline.getPipelineState())
         .setNodes(nodes)
-        .setReplicaIndexes(selectedReplicaIndexes)
+        .setReplicaIndexes(replicaIndexes)
         .build();
-
-    List<ContainerProtos.ChunkInfo> chunks;
-    XceiverClientSpi xceiverClientSpi = null;
-    try {
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Initializing BlockInputStream for get key to access {}",
-            blockID.getContainerID());
-      }
-      xceiverClientSpi = getXceiverClientFactory().acquireClientForReadData(pipeline);
-
-      ContainerProtos.GetBlockResponseProto response = ContainerProtocolCalls
-          .getBlock(xceiverClientSpi, blockID, token, pipeline.getReplicaIndexes());
-
-      chunks = response.getBlockData().getChunksList();
-    } finally {
-      if (xceiverClientSpi != null) {
-        getXceiverClientFactory().releaseClientForReadData(
-            xceiverClientSpi, false);
-      }
-    }
-    return chunks;
   }
 }