errose28 commented on code in PR #10991:
URL: https://github.com/apache/ozone/pull/10991#discussion_r3874428377


##########
hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto:
##########
@@ -199,6 +199,8 @@ message ContainerCommandRequestProto {
   optional   EchoRequestProto echo = 26;
   optional   GetContainerChecksumInfoRequestProto getContainerChecksumInfo = 
27;
   optional   ReadBlockRequestProto readBlock = 28;
+  // Write version of the pipeline as set by SCM in 
DatanodeDetails.currentVersion

Review Comment:
   ```suggestion
     // Version that datanodes should use for this write operation.
     // Set by SCM and forwarded by the client to the datanodes.
   ```



##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java:
##########


Review Comment:
   EC requests don't go through this class. Although the version should still 
be arriving with the write chunk/put block request, we probably need a similar 
change on the DN side for EC.



##########
hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/utils/TestClientCommandsUtils.java:
##########
@@ -0,0 +1,56 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.utils;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.apache.hadoop.hdds.HDDSVersion;
+import 
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Type;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test for {@link ClientCommandsUtils}.
+ */
+public class TestClientCommandsUtils {

Review Comment:
   We should also add a test that it falls back to stream block version when 
the provided version is unknown.



##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java:
##########
@@ -53,6 +54,11 @@ public final class DispatcherContext {
 
   private final Map<Long, Long> container2BCSIDMap;
 
+  // the component version the client asked the datanode to execute this write 
at
+  // defaults to HDDSVersion.STREAM_BLOCK_SUPPORT (pre-ZDU behavior) when the 
client
+  // did not supply one
+  private final int writeVersion;

Review Comment:
   This should be `HDDSVersion` instead of `int` as well.



##########
hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java:
##########


Review Comment:
   Looks like we also need to attach a version for `finalizeBlock`.



##########
hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/utils/ClientCommandsUtils.java:
##########
@@ -46,4 +47,22 @@ public static ContainerProtos.ReadChunkVersion 
getReadChunkVersion(
       return ContainerProtos.ReadChunkVersion.V0;
     }
   }
+
+  /**
+   * Returns the write pipeline (component) version the client asked the 
datanode
+   * to execute the write at, as set by the client from the SCM-provided 
pipeline
+   * version (HDDS-15641 / HDDS-15718). When the request does not carry the 
field
+   * (a client predating zero downtime upgrade support), the datanode defaults 
to
+   * {@link HDDSVersion#STREAM_BLOCK_SUPPORT}, the last component version 
before ZDU,
+   * so the write keeps pre-ZDU behavior. This matches the value SCM 
advertises for
+   * pipelines while the cluster has not finalized ZDU.
+   */
+  public static int getWritePipelineVersion(
+      ContainerProtos.ContainerCommandRequestProto request) {
+    if (request.hasWritePipelineVersion()) {
+      return request.getWritePipelineVersion();
+    } else {
+      return HDDSVersion.STREAM_BLOCK_SUPPORT.serialize();
+    }
+  }

Review Comment:
   We should use the typed version once we take it out of the proto. We can add 
an extra safety check on the deserialize path as well.
   
   ```suggestion
     /**
      * Returns the write pipeline (component) version the client asked the 
datanode
      * to execute the write at, as set by the client from the SCM-provided 
pipeline
      * version. When the request does not carry the field
      * (a client predating zero downtime upgrade support), the datanode 
defaults to
      * {@link HDDSVersion#STREAM_BLOCK_SUPPORT}, the last component version 
before ZDU,
      * so the write keeps pre-ZDU behavior. This matches the value SCM 
advertises for
      * pipelines while the cluster has not finalized ZDU.
      */
     public static HDDSVersion 
getWritePipelineVersion(ContainerProtos.ContainerCommandRequestProto request) {
       HDDSVersion defaultVersion = HDDSVersion.STREAM_BLOCK_SUPPORT;
       if (request.hasWritePipelineVersion()) {
         int serializedVersion = request.getWritePipelineVersion();
         HDDSVersion writeVersion = HDDSVersion.deserialize(serializedVersion);
         if (writeVersion.equals(HDDSVersion.UNKNOWN_VERSION)) {
           // This case should never happen since SCM is aware of Datanodes' 
software and apparent versions and will
           // choose a version for the pipeline accordingly. If a client 
somehow passes an invalid version, maybe one
           // that did not come from SCM, fall back to the safe default.
           LOG.error("Datanode was provided an invalid write version {} for 
software version {}. Falling back to {}",
               serializedVersion, HDDSVersion.SOFTWARE_VERSION, defaultVersion);
         } else {
           return writeVersion;
         }
       } else {
         LOG.trace("No write pipeline version was supplied by the client. 
Falling back to {}", defaultVersion);
       }
       return defaultVersion;
     }
   ```



##########
hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/storage/TestContainerProtocolCalls.java:
##########
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm.storage;
+
+import static 
org.apache.hadoop.hdds.protocol.MockDatanodeDetails.randomDatanodeDetails;
+import static 
org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hadoop.hdds.HDDSVersion;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import 
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.BlockData;
+import 
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerCommandRequestProto;
+import 
org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.DatanodeBlockID;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test that {@link ContainerProtocolCalls} stamps the pipeline write version 
on
+ * outgoing write requests (HDDS-15718).
+ */
+public class TestContainerProtocolCalls {
+
+  private static Pipeline pipelineWithVersion(int version) {
+    List<DatanodeDetails> nodes = Arrays.asList(
+        randomDatanodeDetails(), randomDatanodeDetails(), 
randomDatanodeDetails());
+    nodes.forEach(dn -> dn.setCurrentVersion(version));
+    return Pipeline.newBuilder()
+        .setId(PipelineID.randomId())
+        .setState(Pipeline.PipelineState.OPEN)
+        .setReplicationConfig(RatisReplicationConfig.getInstance(THREE))
+        .setNodes(nodes)
+        .build();
+  }
+
+  @Test
+  void putBlockRequestCarriesPipelineWriteVersion() throws IOException {

Review Comment:
   We should probably have a test for all write operations here. I'm not sure 
there's a good way to make a dynamic matrix so we'll probably have to add one 
test method for each.



##########
hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java:
##########
@@ -549,7 +551,8 @@ public static PutSmallFileResponseProto writeSmallFile(
             .setCmdType(Type.PutSmallFile)
             .setContainerID(blockID.getContainerID())
             .setDatanodeUuid(id)
-            .setPutSmallFile(putSmallFileRequest);
+            .setPutSmallFile(putSmallFileRequest)
+            
.setWritePipelineVersion(client.getPipeline().getFirstNode().getCurrentVersion());

Review Comment:
   Can we collapse these calls into a `Pipeline#getWriteVersion()` which we use 
to assign this value instead? This can still return the current version of the 
first node, but without the overhead of the existing `getFirstNode` 
implementation. We can also put javadoc on this method to note that all nodes 
in the pipeline are expected to have the same version so we just use the first 
one.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to