HDDS-15641. SCM should send write pipeline version with DatanodeDetails on block allocation

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java

@@ -33,6 +33,7 @@
 import java.util.Set;
 import java.util.UUID;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.hdds.ComponentVersion;
 import org.apache.hadoop.hdds.HDDSVersion;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
@@ -524,16 +525,28 @@ public HddsProtos.DatanodeDetailsProto toProto(int clientVersion, Set<Port.Name>
     return toProtoBuilder(clientVersion, filterPorts).build();
   }
 
+  public HddsProtos.DatanodeDetailsProto toProto(int clientVersion, Set<Port.Name> filterPorts,
+      ComponentVersion versionOverride) {
+    return toProtoBuilder(clientVersion, filterPorts, versionOverride).build();
+  }
+
+  public HddsProtos.DatanodeDetailsProto.Builder toProtoBuilder(
+      int clientVersion, Set<Port.Name> filterPorts) {
+    return toProtoBuilder(clientVersion, filterPorts, null);
+  }
+
   /**
    * Converts the current DatanodeDetails instance into a proto {@link HddsProtos.DatanodeDetailsProto.Builder} object.
    *
-   * @param clientVersion - The client version.
-   * @param filterPorts   - A set of {@link Port.Name} specifying ports to include.
-   *                        If empty, all available ports will be included.
+   * @param clientVersion          - The client version.
+   * @param filterPorts            - A set of {@link Port.Name} specifying ports to include.
+   *                                 If empty, all available ports will be included.
+   * @param versionOverride        - When non-null, its serialized value is set as the proto's currentVersion instead
+   *                                 of this node's own version. Used to advertise a pipeline-wide write version.
    * @return A {@link HddsProtos.DatanodeDetailsProto.Builder} Object.
    */
   public HddsProtos.DatanodeDetailsProto.Builder toProtoBuilder(
-      int clientVersion, Set<Port.Name> filterPorts) {
+      int clientVersion, Set<Port.Name> filterPorts, ComponentVersion versionOverride) {
 
     final HddsProtos.DatanodeIDProto idProto = id.toProto();
     final HddsProtos.DatanodeDetailsProto.Builder builder =
@@ -590,7 +603,7 @@ public HddsProtos.DatanodeDetailsProto.Builder toProtoBuilder(
       }
     }
 
-    builder.setCurrentVersion(currentVersion);
+    builder.setCurrentVersion(versionOverride != null ? versionOverride.serialize() : currentVersion);
 
     return builder;
   }

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java

@@ -38,6 +38,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.builder.EqualsBuilder;
 import org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.hadoop.hdds.ComponentVersion;
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.client.ReplicatedReplicationConfig;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
@@ -368,11 +369,16 @@ public HddsProtos.Pipeline getProtobufMessage(int clientVersion) {
   }
 
   public HddsProtos.Pipeline getProtobufMessage(int clientVersion, Set<DatanodeDetails.Port.Name> filterPorts) {
+    return getProtobufMessage(clientVersion, filterPorts, null);
+  }
+
+  public HddsProtos.Pipeline getProtobufMessage(int clientVersion, Set<DatanodeDetails.Port.Name> filterPorts,
+      ComponentVersion versionOverride) {
     List<HddsProtos.DatanodeDetailsProto> members = new ArrayList<>();
     List<Integer> memberReplicaIndexes = new ArrayList<>();
 
     for (DatanodeDetails dn : nodeStatus.keySet()) {
-      members.add(dn.toProto(clientVersion, filterPorts));
+      members.add(dn.toProto(clientVersion, filterPorts, versionOverride));
       memberReplicaIndexes.add(replicaIndexes.getOrDefault(dn, 0));
     }
 

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java

@@ -20,8 +20,12 @@
 import com.google.protobuf.RpcController;
 import com.google.protobuf.ServiceException;
 import java.io.IOException;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
+import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.HDDSVersion;
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
 import org.apache.hadoop.hdds.client.ReplicationConfig;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -47,6 +51,9 @@
 import org.apache.hadoop.hdds.scm.exceptions.SCMException;
 import org.apache.hadoop.hdds.scm.ha.RatisUtil;
 import org.apache.hadoop.hdds.scm.net.InnerNode;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
 import org.apache.hadoop.hdds.scm.protocolPB.ScmBlockLocationProtocolPB;
 import org.apache.hadoop.hdds.scm.protocolPB.StorageContainerLocationProtocolPB;
 import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
@@ -210,15 +217,54 @@ public AllocateScmBlockResponseProto allocateScmBlock(
           " blocks. Requested " + request.getNumBlocks() + " blocks",
           SCMException.ResultCodes.FAILED_TO_ALLOCATE_ENOUGH_BLOCKS);
     }
+    Map<PipelineID, HddsProtos.Pipeline> pipelineProtoCache = new HashMap<>();
     for (AllocatedBlock block : allocatedBlocks) {
+      Pipeline pipeline = block.getPipeline();
+      HddsProtos.Pipeline pipelineProto = pipelineProtoCache.get(pipeline.getId());
+      if (pipelineProto == null) {
+        pipelineProto = pipeline.getProtobufMessage(clientVersion, Name.IO_PORTS,
+            computePipelineWriteVersion(pipeline));
+        pipelineProtoCache.put(pipeline.getId(), pipelineProto);
+      }
       builder.addBlocks(AllocateBlockResponse.newBuilder()
           .setContainerBlockID(block.getBlockID().getProtobuf())
-          .setPipeline(block.getPipeline().getProtobufMessage(clientVersion, Name.IO_PORTS)));
+          .setPipeline(pipelineProto));
     }
 
     return builder.build();
   }
 
+  /**
+   * Computes the version clients should use for writes to the given pipeline.
+   * Before ZDU is finalized, datanodes report apparent versions from the
+   * {@link HDDSLayoutFeature} enum, which clients cannot compare against the
+   * {@link HDDSVersion} enum they use. In that state we advertise the last
+   * {@link HDDSVersion} before {@code ZDU} so clients keep pre-ZDU write behavior
+   * until the cluster finalizes. Once ZDU is finalized every apparent version is
+   * an {@link HDDSVersion} and safe to share: we return the lowest one across the
+   * pipeline, so during a later rolling upgrade clients do not enable a newer
+   * write-path feature until every datanode in the pipeline has finalized.
+   */
+  private ComponentVersion computePipelineWriteVersion(Pipeline pipeline) throws SCMException {
+    List<DatanodeDetails> nodes = pipeline.getNodes();
+    if (nodes.isEmpty()) {
+      throw new SCMException("Cannot determine the write version for pipeline "
+          + pipeline.getId() + " because it has no datanodes",
+          SCMException.ResultCodes.NO_SUCH_DATANODE);
+    }
+    if (!scm.getVersionManager().isAllowed(HDDSVersion.ZDU)) {
+      return HDDSVersion.STREAM_BLOCK_SUPPORT; // last HDDSVersion before ZDU
+    }
+    try {
+      return scm.getScmNodeManager()
+          .getLowestApparentVersion(nodes.toArray(new DatanodeDetails[0]));
+    } catch (NodeNotFoundException e) {
+      throw new SCMException("Datanode not found while computing the write version "
+          + "for pipeline " + pipeline.getId() + " during block allocation", e,
+          SCMException.ResultCodes.NO_SUCH_DATANODE);
+    }
+  }
+
   public DeleteScmKeyBlocksResponseProto deleteScmKeyBlocks(
       DeleteScmKeyBlocksRequestProto req
   )