This is an automated email from the ASF dual-hosted git repository.
errose28 pushed a commit to branch HDDS-14496-zdu
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/HDDS-14496-zdu by this push:
new 03b462c681a HDDS-15641. SCM should send write pipeline version with
DatanodeDetails on block allocation (#10878)
03b462c681a is described below
commit 03b462c681a703cedffc377ac6bedfb1405effff
Author: Zita Dombi <[email protected]>
AuthorDate: Thu Aug 6 00:16:10 2026 +0200
HDDS-15641. SCM should send write pipeline version with DatanodeDetails on
block allocation (#10878)
---
.../hadoop/hdds/protocol/DatanodeDetails.java | 23 +-
.../apache/hadoop/hdds/scm/pipeline/Pipeline.java | 8 +-
...lockLocationProtocolServerSideTranslatorPB.java | 48 +++-
...lockLocationProtocolServerSideTranslatorPB.java | 269 +++++++++++++++++++++
.../client/rpc/TestBlockDataStreamOutput.java | 9 +-
5 files changed, 347 insertions(+), 10 deletions(-)
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java
index 4b0108aa9ae..98b51ef705a 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java
+++
b/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;
}
diff --git
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
index 62c3855f1af..5675787c003 100644
---
a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/pipeline/Pipeline.java
+++
b/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));
}
diff --git
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java
b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java
index 007376670ba..9fb688fb8fe 100644
---
a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/protocol/ScmBlockLocationProtocolServerSideTranslatorPB.java
+++
b/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
)
diff --git
a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java
new file mode 100644
index 00000000000..6b7674d7642
--- /dev/null
+++
b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java
@@ -0,0 +1,269 @@
+/*
+ * 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.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.Collections;
+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.exceptions.SCMException;
+import org.apache.hadoop.hdds.scm.node.DatanodeInfo;
+import org.apache.hadoop.hdds.scm.node.NodeManager;
+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.scm.server.upgrade.ScmVersionManager;
+import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature;
+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 SOFTWARE_VERSION =
HDDSVersion.SOFTWARE_VERSION.serialize();
+
+ private ScmBlockLocationProtocol impl;
+ private NodeManager nodeManager;
+ private ScmVersionManager versionManager;
+ 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);
+ versionManager = mock(ScmVersionManager.class);
+
+ nodes = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ DatanodeDetails dn = randomDatanodeDetails();
+ dn.setCurrentVersion(SOFTWARE_VERSION);
+ nodes.add(dn);
+ }
+
+ Pipeline pipeline = buildPipeline(nodes);
+ AllocatedBlock block = blockOn(1L, pipeline);
+
+ when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class),
+ anyString(), any(),
anyString())).thenReturn(Collections.singletonList(block));
+ when(scm.getScmNodeManager()).thenReturn(nodeManager);
+ when(scm.getVersionManager()).thenReturn(versionManager);
+ // Exercise the real min-across-nodes helper, driven by the getNode stubs.
+
lenient().when(nodeManager.getLowestApparentVersion(any(DatanodeDetails[].class))).thenCallRealMethod();
+
+ // Default: ZDU is finalized and every datanode is at the software version.
+ when(versionManager.isAllowed(HDDSVersion.ZDU)).thenReturn(true);
+ for (DatanodeDetails dn : nodes) {
+ setDatanodeApparentVersion(dn, HDDSVersion.SOFTWARE_VERSION);
+ }
+
+ 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 {
+ // Before ZDU is finalized, datanodes report apparent versions from the
+ // HDDSLayoutFeature enum. Regardless of what they report, clients must be
+ // clamped to the last HDDSVersion before ZDU.
+ when(versionManager.isAllowed(HDDSVersion.ZDU)).thenReturn(false);
+ for (DatanodeDetails dn : nodes) {
+ setDatanodeApparentVersion(dn,
HDDSLayoutFeature.STORAGE_SPACE_DISTRIBUTION);
+ }
+
+ assertAllMembersHaveVersion(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(),
allocateAndGetMembers());
+ }
+
+ @Test
+ void finalizedClusterForwardsRealVersion() throws Exception {
+ assertAllMembersHaveVersion(SOFTWARE_VERSION, allocateAndGetMembers());
+ }
+
+ @Test
+ void finalizedPipelineForwardsLowestApparentVersion() throws Exception {
+ // ZDU is finalized, but the pipeline mixes datanodes at different apparent
+ // versions (as during a later rolling upgrade). The lowest is forwarded so
+ // clients do not enable a feature an un-upgraded datanode cannot handle.
+ setDatanodeApparentVersion(nodes.get(1),
HDDSVersion.COMBINED_PUTBLOCK_WRITECHUNK_RPC);
+
+
assertAllMembersHaveVersion(HDDSVersion.COMBINED_PUTBLOCK_WRITECHUNK_RPC.serialize(),
allocateAndGetMembers());
+ }
+
+ @Test
+ void missingDatanodeInfoFailsAllocation() throws Exception {
+ when(nodeManager.getNode(nodes.get(2).getID())).thenReturn(null);
+
+ SCMException e = assertThrows(SCMException.class, () -> allocate(1));
+ assertEquals(SCMException.ResultCodes.NO_SUCH_DATANODE, e.getResult());
+ }
+
+ @Test
+ void emptyPipelineFailsAllocation() throws Exception {
+ setAllocatedBlocks(Collections.singletonList(blockOn(1L,
buildPipeline(Collections.emptyList()))));
+
+ SCMException e = assertThrows(SCMException.class, () -> allocate(1));
+ assertEquals(SCMException.ResultCodes.NO_SUCH_DATANODE, e.getResult());
+ }
+
+ @Test
+ void blocksSharingPipelineAllGetClampedVersion() throws Exception {
+ // One straggler datanode clamps the shared pipeline's write version.
+ setDatanodeApparentVersion(nodes.get(1),
HDDSVersion.COMBINED_PUTBLOCK_WRITECHUNK_RPC);
+
+ // Three blocks allocated on the same pipeline object; the memoized proto
+ // must be returned for each block with the clamped version intact.
+ Pipeline pipeline = buildPipeline(nodes);
+ setAllocatedBlocks(Arrays.asList(blockOn(1L, pipeline), blockOn(2L,
pipeline), blockOn(3L, pipeline)));
+
+ AllocateScmBlockResponseProto response = allocate(3);
+
+ assertEquals(3, response.getBlocksCount());
+ for (int i = 0; i < response.getBlocksCount(); i++) {
+
assertAllMembersHaveVersion(HDDSVersion.COMBINED_PUTBLOCK_WRITECHUNK_RPC.serialize(),
+ response.getBlocks(i).getPipeline().getMembersList());
+ }
+
+ // The write version is memoized per pipeline: each node is looked up once
+ // for the whole batch, not once per block.
+ for (DatanodeDetails dn : nodes) {
+ verify(nodeManager, times(1)).getNode(dn.getID());
+ }
+ }
+
+ @Test
+ void blocksOnDistinctPipelinesGetOwnVersion() throws Exception {
+ // A second, distinct pipeline whose datanodes are at an older
+ // version than the software-version pipeline built in setUp().
+ List<DatanodeDetails> otherNodes = new ArrayList<>();
+ for (int i = 0; i < 3; i++) {
+ DatanodeDetails dn = randomDatanodeDetails();
+ dn.setCurrentVersion(SOFTWARE_VERSION);
+ setDatanodeApparentVersion(dn, HDDSVersion.STREAM_BLOCK_SUPPORT);
+ otherNodes.add(dn);
+ }
+
+ Pipeline finalized = buildPipeline(nodes);
+ Pipeline straggler = buildPipeline(otherNodes);
+ setAllocatedBlocks(Arrays.asList(blockOn(1L, finalized), blockOn(2L,
straggler)));
+
+ AllocateScmBlockResponseProto response = allocate(2);
+
+ assertEquals(2, response.getBlocksCount());
+ assertAllMembersHaveVersion(SOFTWARE_VERSION,
response.getBlocks(0).getPipeline().getMembersList());
+ assertEquals(otherNodes.size(),
response.getBlocks(1).getPipeline().getMembersCount());
+ for (DatanodeDetailsProto member :
response.getBlocks(1).getPipeline().getMembersList()) {
+ assertEquals(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(),
member.getCurrentVersion());
+ }
+ }
+
+ @Test
+ void doesNotMutateSourcePipelineDatanodes() throws Exception {
+ setDatanodeApparentVersion(nodes.get(1), HDDSVersion.STREAM_BLOCK_SUPPORT);
+
+ allocateAndGetMembers();
+
+ // The in-memory DatanodeDetails (shared with SCM internal state) must keep
+ // their real software version; only the outgoing proto is overridden.
+ for (DatanodeDetails dn : nodes) {
+ assertEquals(SOFTWARE_VERSION, dn.getCurrentVersion());
+ }
+ }
+}
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java
index 4a57e96b7cc..6be6a99a658 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java
@@ -135,6 +135,7 @@ static MiniOzoneCluster createCluster() throws IOException,
.setNumDatanodes(5)
.setDatanodeFactory(UniformDatanodesFactory.newBuilder()
.setCurrentVersion(DN_OLD_VERSION)
+ .setApparentVersion(HDDSVersion.SOFTWARE_VERSION.serialize())
.build())
.build();
cluster.waitForPipelineTobeReady(HddsProtos.ReplicationFactor.THREE,
@@ -337,7 +338,7 @@ public void testTotalAckDataLength(boolean flushDelay)
throws Exception {
public void testDatanodeVersion(boolean flushDelay) throws Exception {
OzoneClientConfig config = newClientConfig(cluster.getConf(), flushDelay);
try (OzoneClient client = newClient(cluster.getConf(), config)) {
- // Verify all DNs internally have versions set correctly
+ // Each datanode advertises its own (older) currentVersion internally.
List<HddsDatanodeService> dns = cluster.getHddsDatanodes();
for (HddsDatanodeService dn : dns) {
DatanodeDetails details = dn.getDatanodeDetails();
@@ -350,10 +351,12 @@ public void testDatanodeVersion(boolean flushDelay)
throws Exception {
KeyDataStreamOutput keyDataStreamOutput = (KeyDataStreamOutput)
key.getByteBufStreamOutput();
BlockDataStreamOutputEntry stream =
keyDataStreamOutput.getStreamEntries().get(0);
- // Now check 3 DNs in a random pipeline returns the correct DN versions
+ // The cluster is finalized for ZDU, so the pipeline SCM returns on
block allocation
+ // stamps each member's currentVersion with the lowest apparent version
across the
+ // pipeline, which here is the software version the datanodes have
finalized to.
List<DatanodeDetails> streamDnDetails = stream.getPipeline().getNodes();
for (DatanodeDetails details : streamDnDetails) {
- assertEquals(DN_OLD_VERSION,
+ assertEquals(HDDSVersion.SOFTWARE_VERSION,
HDDSVersion.deserialize(details.getCurrentVersion()));
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
