errose28 commented on code in PR #10878:
URL: https://github.com/apache/ozone/pull/10878#discussion_r3684467133
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -210,15 +215,64 @@ 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) {
+ try {
+ pipelineProto = withWriteVersion(
+ pipeline.getProtobufMessage(clientVersion, Name.IO_PORTS),
+ computeClusterWriteVersion(pipeline));
+ } catch (NodeNotFoundException e) {
+ throw new IllegalStateException("Datanode not found in NodeManager "
+ + "while computing the write version for pipeline "
+ + pipeline.getId() + " during block allocation. Should not
happen", e);
+ }
+ 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:
+ * the lowest apparent version among the pipeline's datanodes. During a
rolling
+ * upgrade a pipeline may mix finalized and unfinalized datanodes: an
+ * unfinalized datanode runs the newer software but reports the older
apparent
+ * version until it finalizes. Clients must not enable newer write-path
+ * features until every datanode handling their writes has finalized, so the
+ * minimum apparent version across the pipeline is used.
+ */
+ private int computeClusterWriteVersion(Pipeline pipeline) throws
NodeNotFoundException {
Review Comment:
```suggestion
private int computePipelineWriteVersion(Pipeline pipeline) throws
NodeNotFoundException {
```
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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.protocol;
+
+import static
org.apache.hadoop.hdds.protocol.MockDatanodeDetails.randomDatanodeDetails;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.HDDSVersion;
+import org.apache.hadoop.hdds.client.ContainerBlockID;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
+import
org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.AllocateScmBlockRequestProto;
+import
org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.AllocateScmBlockResponseProto;
+import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
+import org.apache.hadoop.ozone.ClientVersion;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that {@link ScmBlockLocationProtocolServerSideTranslatorPB} forwards a
+ * clamped write version (based on the cluster's finalization status) in the
+ * {@code currentVersion} of every pipeline member it returns on block
+ * allocation, without mutating SCM's in-memory state.
+ */
+class TestScmBlockLocationProtocolServerSideTranslatorPB {
+
+ /** The version each source datanode reports as its own software version. */
+ private static final int DN_SOFTWARE_VERSION =
+ HDDSVersion.ZDU.serialize();
+
+ private ScmBlockLocationProtocol impl;
+ private NodeManager nodeManager;
+ private ScmBlockLocationProtocolServerSideTranslatorPB service;
+ private List<DatanodeDetails> nodes;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ impl = mock(ScmBlockLocationProtocol.class);
+ StorageContainerManager scm = mock(StorageContainerManager.class);
+ nodeManager = mock(NodeManager.class);
+
+ nodes = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ DatanodeDetails dn = randomDatanodeDetails();
+ dn.setCurrentVersion(DN_SOFTWARE_VERSION);
+ nodes.add(dn);
+ }
+
+ Pipeline pipeline = Pipeline.newBuilder()
+ .setId(PipelineID.randomId())
+ .setState(PipelineState.OPEN)
+ .setReplicationConfig(
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(nodes)
+ .build();
+ AllocatedBlock block = new AllocatedBlock.Builder()
+ .setContainerBlockID(new ContainerBlockID(1L, 1L))
+ .setPipeline(pipeline)
+ .build();
+
+ List<AllocatedBlock> blocks = new ArrayList<>();
+ blocks.add(block);
+ when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class),
+ anyString(), any(), anyString())).thenReturn(blocks);
+ when(scm.getScmNodeManager()).thenReturn(nodeManager);
+ // Exercise the real min-across-nodes helper, driven by the getNode stubs.
+
lenient().when(nodeManager.getLowestApparentVersion(any(DatanodeDetails[].class)))
+ .thenCallRealMethod();
+
+ // Default: every datanode finalized to ZDU.
+ for (DatanodeDetails dn : nodes) {
+ setDatanodeApparentVersion(dn, HDDSVersion.ZDU);
+ }
+
+ service = new ScmBlockLocationProtocolServerSideTranslatorPB(
+ impl, scm, mock(ProtocolMessageMetrics.class));
+ }
+
+ private void setDatanodeApparentVersion(DatanodeDetails dn,
+ ComponentVersion version) {
+ DatanodeInfo info = mock(DatanodeInfo.class);
+ lenient().when(info.getLastKnownApparentVersion()).thenReturn(version);
+ when(nodeManager.getNode(dn.getID())).thenReturn(info);
+ }
+
+ private List<DatanodeDetailsProto> allocateAndGetMembers() throws Exception {
+ return allocate(1).getBlocks(0).getPipeline().getMembersList();
+ }
+
+ private AllocateScmBlockResponseProto allocate(int numBlocks)
+ throws Exception {
+ AllocateScmBlockRequestProto request =
+ AllocateScmBlockRequestProto.newBuilder()
+ .setSize(1024)
+ .setNumBlocks(numBlocks)
+ .setType(ReplicationType.RATIS)
+ .setFactor(ReplicationFactor.THREE)
+ .setOwner("owner")
+ .build();
+ return service.allocateScmBlock(request,
ClientVersion.CURRENT.serialize());
+ }
+
+ private Pipeline buildPipeline(List<DatanodeDetails> pipelineNodes) {
+ return Pipeline.newBuilder()
+ .setId(PipelineID.randomId())
+ .setState(PipelineState.OPEN)
+ .setReplicationConfig(
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(pipelineNodes)
+ .build();
+ }
+
+ private AllocatedBlock blockOn(long localId, Pipeline pipeline) {
+ return new AllocatedBlock.Builder()
+ .setContainerBlockID(new ContainerBlockID(1L, localId))
+ .setPipeline(pipeline)
+ .build();
+ }
+
+ private void setAllocatedBlocks(List<AllocatedBlock> blocks)
+ throws Exception {
+ when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class),
+ anyString(), any(), anyString())).thenReturn(blocks);
+ }
+
+ private void assertAllMembersHaveVersion(int expected,
+ List<DatanodeDetailsProto> members) {
+ assertEquals(nodes.size(), members.size());
+ for (DatanodeDetailsProto member : members) {
+ assertEquals(expected, member.getCurrentVersion());
+ }
+ }
+
+ @Test
+ void preFinalizedClusterClampsClientVersionDown() throws Exception {
+ // Datanodes still run ZDU software but have not finalized past
+ // STREAM_BLOCK_SUPPORT, so their apparent version is the older one.
+ for (DatanodeDetails dn : nodes) {
+ setDatanodeApparentVersion(dn, HDDSVersion.STREAM_BLOCK_SUPPORT);
+ }
+
+ assertAllMembersHaveVersion(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(),
+ allocateAndGetMembers());
+ }
+
+ @Test
+ void finalizedClusterForwardsRealVersion() throws Exception {
+ // Default setup: SCM and all datanodes at ZDU.
+ assertAllMembersHaveVersion(HDDSVersion.ZDU.serialize(),
Review Comment:
ZDU won't be the finalized version in future releases. `SOFTWARE_VERSION`
would need to be used for that.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -210,15 +215,64 @@ 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) {
+ try {
+ pipelineProto = withWriteVersion(
+ pipeline.getProtobufMessage(clientVersion, Name.IO_PORTS),
+ computeClusterWriteVersion(pipeline));
+ } catch (NodeNotFoundException e) {
+ throw new IllegalStateException("Datanode not found in NodeManager "
+ + "while computing the write version for pipeline "
+ + pipeline.getId() + " during block allocation. Should not
happen", e);
+ }
+ 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:
+ * the lowest apparent version among the pipeline's datanodes. During a
rolling
+ * upgrade a pipeline may mix finalized and unfinalized datanodes: an
+ * unfinalized datanode runs the newer software but reports the older
apparent
+ * version until it finalizes. Clients must not enable newer write-path
+ * features until every datanode handling their writes has finalized, so the
+ * minimum apparent version across the pipeline is used.
+ */
+ private int computeClusterWriteVersion(Pipeline pipeline) throws
NodeNotFoundException {
+ return scm.getScmNodeManager()
+ .getLowestApparentVersion(pipeline.getNodes().toArray(new
DatanodeDetails[0]))
+ .serialize();
Review Comment:
We can return `ComponentVersion` here and save serialization for the last
step before proto conversion.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -210,15 +215,64 @@ 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) {
+ try {
+ pipelineProto = withWriteVersion(
+ pipeline.getProtobufMessage(clientVersion, Name.IO_PORTS),
+ computeClusterWriteVersion(pipeline));
Review Comment:
We shouldn't have a pipeline with zero DNs, but just in case we should
validate this and return a checked exception if it happens. Otherwise will hit
the unchecked `IllegalArgumentException` when finding the minimum version.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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.protocol;
+
+import static
org.apache.hadoop.hdds.protocol.MockDatanodeDetails.randomDatanodeDetails;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.HDDSVersion;
+import org.apache.hadoop.hdds.client.ContainerBlockID;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
+import
org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.AllocateScmBlockRequestProto;
+import
org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.AllocateScmBlockResponseProto;
+import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
+import org.apache.hadoop.ozone.ClientVersion;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that {@link ScmBlockLocationProtocolServerSideTranslatorPB} forwards a
+ * clamped write version (based on the cluster's finalization status) in the
+ * {@code currentVersion} of every pipeline member it returns on block
+ * allocation, without mutating SCM's in-memory state.
+ */
+class TestScmBlockLocationProtocolServerSideTranslatorPB {
+
+ /** The version each source datanode reports as its own software version. */
+ private static final int DN_SOFTWARE_VERSION =
+ HDDSVersion.ZDU.serialize();
+
+ private ScmBlockLocationProtocol impl;
+ private NodeManager nodeManager;
+ private ScmBlockLocationProtocolServerSideTranslatorPB service;
+ private List<DatanodeDetails> nodes;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ impl = mock(ScmBlockLocationProtocol.class);
+ StorageContainerManager scm = mock(StorageContainerManager.class);
+ nodeManager = mock(NodeManager.class);
+
+ nodes = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ DatanodeDetails dn = randomDatanodeDetails();
+ dn.setCurrentVersion(DN_SOFTWARE_VERSION);
+ nodes.add(dn);
+ }
+
+ Pipeline pipeline = Pipeline.newBuilder()
+ .setId(PipelineID.randomId())
+ .setState(PipelineState.OPEN)
+ .setReplicationConfig(
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(nodes)
+ .build();
+ AllocatedBlock block = new AllocatedBlock.Builder()
+ .setContainerBlockID(new ContainerBlockID(1L, 1L))
+ .setPipeline(pipeline)
+ .build();
+
+ List<AllocatedBlock> blocks = new ArrayList<>();
+ blocks.add(block);
+ when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class),
+ anyString(), any(), anyString())).thenReturn(blocks);
+ when(scm.getScmNodeManager()).thenReturn(nodeManager);
+ // Exercise the real min-across-nodes helper, driven by the getNode stubs.
+
lenient().when(nodeManager.getLowestApparentVersion(any(DatanodeDetails[].class)))
+ .thenCallRealMethod();
+
+ // Default: every datanode finalized to ZDU.
+ for (DatanodeDetails dn : nodes) {
+ setDatanodeApparentVersion(dn, HDDSVersion.ZDU);
+ }
+
+ service = new ScmBlockLocationProtocolServerSideTranslatorPB(
+ impl, scm, mock(ProtocolMessageMetrics.class));
+ }
+
+ private void setDatanodeApparentVersion(DatanodeDetails dn,
+ ComponentVersion version) {
+ DatanodeInfo info = mock(DatanodeInfo.class);
+ lenient().when(info.getLastKnownApparentVersion()).thenReturn(version);
+ when(nodeManager.getNode(dn.getID())).thenReturn(info);
+ }
+
+ private List<DatanodeDetailsProto> allocateAndGetMembers() throws Exception {
+ return allocate(1).getBlocks(0).getPipeline().getMembersList();
+ }
+
+ private AllocateScmBlockResponseProto allocate(int numBlocks)
+ throws Exception {
+ AllocateScmBlockRequestProto request =
+ AllocateScmBlockRequestProto.newBuilder()
+ .setSize(1024)
+ .setNumBlocks(numBlocks)
+ .setType(ReplicationType.RATIS)
+ .setFactor(ReplicationFactor.THREE)
+ .setOwner("owner")
+ .build();
+ return service.allocateScmBlock(request,
ClientVersion.CURRENT.serialize());
+ }
+
+ private Pipeline buildPipeline(List<DatanodeDetails> pipelineNodes) {
+ return Pipeline.newBuilder()
+ .setId(PipelineID.randomId())
+ .setState(PipelineState.OPEN)
+ .setReplicationConfig(
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(pipelineNodes)
+ .build();
+ }
+
+ private AllocatedBlock blockOn(long localId, Pipeline pipeline) {
+ return new AllocatedBlock.Builder()
+ .setContainerBlockID(new ContainerBlockID(1L, localId))
+ .setPipeline(pipeline)
+ .build();
+ }
+
+ private void setAllocatedBlocks(List<AllocatedBlock> blocks)
+ throws Exception {
+ when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class),
+ anyString(), any(), anyString())).thenReturn(blocks);
+ }
+
+ private void assertAllMembersHaveVersion(int expected,
+ List<DatanodeDetailsProto> members) {
+ assertEquals(nodes.size(), members.size());
+ for (DatanodeDetailsProto member : members) {
+ assertEquals(expected, member.getCurrentVersion());
+ }
+ }
+
+ @Test
+ void preFinalizedClusterClampsClientVersionDown() throws Exception {
+ // Datanodes still run ZDU software but have not finalized past
+ // STREAM_BLOCK_SUPPORT, so their apparent version is the older one.
+ for (DatanodeDetails dn : nodes) {
+ setDatanodeApparentVersion(dn, HDDSVersion.STREAM_BLOCK_SUPPORT);
Review Comment:
Technically we should use `HDDSLayoutFeature` for the earlier versions,
since that's what Datanodes will be reporting to SCM when pre-finalized for ZDU
from an older version. We should update the tests for clarity, but it won't
matter in practice because nothing within the datanode will be reading this
version field until a write pipeline incompatibility is potentially introduced
after ZDU.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -210,15 +215,64 @@ 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) {
+ try {
+ pipelineProto = withWriteVersion(
+ pipeline.getProtobufMessage(clientVersion, Name.IO_PORTS),
+ computeClusterWriteVersion(pipeline));
+ } catch (NodeNotFoundException e) {
+ throw new IllegalStateException("Datanode not found in NodeManager "
+ + "while computing the write version for pipeline "
+ + pipeline.getId() + " during block allocation. Should not
happen", e);
+ }
+ 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:
+ * the lowest apparent version among the pipeline's datanodes. During a
rolling
+ * upgrade a pipeline may mix finalized and unfinalized datanodes: an
+ * unfinalized datanode runs the newer software but reports the older
apparent
+ * version until it finalizes. Clients must not enable newer write-path
+ * features until every datanode handling their writes has finalized, so the
+ * minimum apparent version across the pipeline is used.
+ */
+ private int computeClusterWriteVersion(Pipeline pipeline) throws
NodeNotFoundException {
+ return scm.getScmNodeManager()
+ .getLowestApparentVersion(pipeline.getNodes().toArray(new
DatanodeDetails[0]))
+ .serialize();
+ }
+
+ /**
+ * Returns a copy of the pipeline proto with every member's currentVersion
+ * overridden with the computed pipeline write version. The override is
applied
+ * only to the outgoing proto sent to the client; the in-memory pipeline and
+ * its {@link DatanodeDetails} objects (shared with SCM internal state) are
+ * left untouched, so persistence and admin paths keep the real datanode
+ * version. Member order is preserved, keeping {@code memberOrders} and
+ * {@code memberReplicaIndexes} indices valid.
+ */
+ private static HddsProtos.Pipeline withWriteVersion(
Review Comment:
This is doing an extra copy of the pipeline object. We can add the apparent
pipeline version as a `ComponentVersion` parameter to
`Pipeline#getProtobufMessage`. It would then be passed through to
`DatanodeDetails#toProtoBuilder` as a new parameter as well to set the apparent
version of the node. We may need to add overloads for these methods when they
are used elsewhere that use the `DatanodeDetail`'s existing version when a new
one is not specified.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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.protocol;
+
+import static
org.apache.hadoop.hdds.protocol.MockDatanodeDetails.randomDatanodeDetails;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.HDDSVersion;
+import org.apache.hadoop.hdds.client.ContainerBlockID;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
+import
org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.AllocateScmBlockRequestProto;
+import
org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.AllocateScmBlockResponseProto;
+import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
+import org.apache.hadoop.ozone.ClientVersion;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that {@link ScmBlockLocationProtocolServerSideTranslatorPB} forwards a
+ * clamped write version (based on the cluster's finalization status) in the
+ * {@code currentVersion} of every pipeline member it returns on block
+ * allocation, without mutating SCM's in-memory state.
+ */
+class TestScmBlockLocationProtocolServerSideTranslatorPB {
+
+ /** The version each source datanode reports as its own software version. */
+ private static final int DN_SOFTWARE_VERSION =
+ HDDSVersion.ZDU.serialize();
Review Comment:
Datanodes report their apparent and software versions, but the functionality
under test is only concerned with apparent version. It looks like this is
aiming to set the default apparent version reported to show nodes as finalized,
which should be `SOFTWARE_VERSION`.
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * 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.protocol;
+
+import static
org.apache.hadoop.hdds.protocol.MockDatanodeDetails.randomDatanodeDetails;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.apache.hadoop.hdds.ComponentVersion;
+import org.apache.hadoop.hdds.HDDSVersion;
+import org.apache.hadoop.hdds.client.ContainerBlockID;
+import org.apache.hadoop.hdds.client.RatisReplicationConfig;
+import org.apache.hadoop.hdds.client.ReplicationConfig;
+import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationType;
+import
org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.AllocateScmBlockRequestProto;
+import
org.apache.hadoop.hdds.protocol.proto.ScmBlockLocationProtocolProtos.AllocateScmBlockResponseProto;
+import org.apache.hadoop.hdds.scm.container.common.helpers.AllocatedBlock;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
+import org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState;
+import org.apache.hadoop.hdds.scm.pipeline.PipelineID;
+import org.apache.hadoop.hdds.scm.server.StorageContainerManager;
+import org.apache.hadoop.hdds.utils.ProtocolMessageMetrics;
+import org.apache.hadoop.ozone.ClientVersion;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests that {@link ScmBlockLocationProtocolServerSideTranslatorPB} forwards a
+ * clamped write version (based on the cluster's finalization status) in the
+ * {@code currentVersion} of every pipeline member it returns on block
+ * allocation, without mutating SCM's in-memory state.
+ */
+class TestScmBlockLocationProtocolServerSideTranslatorPB {
+
+ /** The version each source datanode reports as its own software version. */
+ private static final int DN_SOFTWARE_VERSION =
+ HDDSVersion.ZDU.serialize();
+
+ private ScmBlockLocationProtocol impl;
+ private NodeManager nodeManager;
+ private ScmBlockLocationProtocolServerSideTranslatorPB service;
+ private List<DatanodeDetails> nodes;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ impl = mock(ScmBlockLocationProtocol.class);
+ StorageContainerManager scm = mock(StorageContainerManager.class);
+ nodeManager = mock(NodeManager.class);
+
+ nodes = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ DatanodeDetails dn = randomDatanodeDetails();
+ dn.setCurrentVersion(DN_SOFTWARE_VERSION);
+ nodes.add(dn);
+ }
+
+ Pipeline pipeline = Pipeline.newBuilder()
+ .setId(PipelineID.randomId())
+ .setState(PipelineState.OPEN)
+ .setReplicationConfig(
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(nodes)
+ .build();
+ AllocatedBlock block = new AllocatedBlock.Builder()
+ .setContainerBlockID(new ContainerBlockID(1L, 1L))
+ .setPipeline(pipeline)
+ .build();
+
+ List<AllocatedBlock> blocks = new ArrayList<>();
+ blocks.add(block);
+ when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class),
+ anyString(), any(), anyString())).thenReturn(blocks);
+ when(scm.getScmNodeManager()).thenReturn(nodeManager);
+ // Exercise the real min-across-nodes helper, driven by the getNode stubs.
+
lenient().when(nodeManager.getLowestApparentVersion(any(DatanodeDetails[].class)))
+ .thenCallRealMethod();
+
+ // Default: every datanode finalized to ZDU.
+ for (DatanodeDetails dn : nodes) {
+ setDatanodeApparentVersion(dn, HDDSVersion.ZDU);
+ }
+
+ service = new ScmBlockLocationProtocolServerSideTranslatorPB(
+ impl, scm, mock(ProtocolMessageMetrics.class));
+ }
+
+ private void setDatanodeApparentVersion(DatanodeDetails dn,
+ ComponentVersion version) {
+ DatanodeInfo info = mock(DatanodeInfo.class);
+ lenient().when(info.getLastKnownApparentVersion()).thenReturn(version);
+ when(nodeManager.getNode(dn.getID())).thenReturn(info);
+ }
+
+ private List<DatanodeDetailsProto> allocateAndGetMembers() throws Exception {
+ return allocate(1).getBlocks(0).getPipeline().getMembersList();
+ }
+
+ private AllocateScmBlockResponseProto allocate(int numBlocks)
+ throws Exception {
+ AllocateScmBlockRequestProto request =
+ AllocateScmBlockRequestProto.newBuilder()
+ .setSize(1024)
+ .setNumBlocks(numBlocks)
+ .setType(ReplicationType.RATIS)
+ .setFactor(ReplicationFactor.THREE)
+ .setOwner("owner")
+ .build();
+ return service.allocateScmBlock(request,
ClientVersion.CURRENT.serialize());
+ }
+
+ private Pipeline buildPipeline(List<DatanodeDetails> pipelineNodes) {
+ return Pipeline.newBuilder()
+ .setId(PipelineID.randomId())
+ .setState(PipelineState.OPEN)
+ .setReplicationConfig(
+ RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+ .setNodes(pipelineNodes)
+ .build();
+ }
+
+ private AllocatedBlock blockOn(long localId, Pipeline pipeline) {
+ return new AllocatedBlock.Builder()
+ .setContainerBlockID(new ContainerBlockID(1L, localId))
+ .setPipeline(pipeline)
+ .build();
+ }
+
+ private void setAllocatedBlocks(List<AllocatedBlock> blocks)
+ throws Exception {
+ when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class),
+ anyString(), any(), anyString())).thenReturn(blocks);
+ }
+
+ private void assertAllMembersHaveVersion(int expected,
+ List<DatanodeDetailsProto> members) {
+ assertEquals(nodes.size(), members.size());
+ for (DatanodeDetailsProto member : members) {
+ assertEquals(expected, member.getCurrentVersion());
+ }
+ }
+
+ @Test
+ void preFinalizedClusterClampsClientVersionDown() throws Exception {
+ // Datanodes still run ZDU software but have not finalized past
+ // STREAM_BLOCK_SUPPORT, so their apparent version is the older one.
+ for (DatanodeDetails dn : nodes) {
+ setDatanodeApparentVersion(dn, HDDSVersion.STREAM_BLOCK_SUPPORT);
+ }
+
+ assertAllMembersHaveVersion(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(),
+ allocateAndGetMembers());
Review Comment:
nit. We don't need to wrap the lines this much.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java:
##########
@@ -210,15 +215,64 @@ 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) {
+ try {
+ pipelineProto = withWriteVersion(
+ pipeline.getProtobufMessage(clientVersion, Name.IO_PORTS),
+ computeClusterWriteVersion(pipeline));
+ } catch (NodeNotFoundException e) {
+ throw new IllegalStateException("Datanode not found in NodeManager "
+ + "while computing the write version for pipeline "
+ + pipeline.getId() + " during block allocation. Should not
happen", e);
Review Comment:
In this case I think we should convert to `ScmException` with result code
`NO_SUCH_DATANODE` to cleanly fail the RPC call.
--
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]
