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]
