From 8ccbbd2417bf78c2e3e5c4a612b2b4c8db0afd23 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Wed, 24 Jun 2026 14:55:08 +0200 Subject: [PATCH 1/5] HDDS-15641. SCM should send write pipeline version with DatanodeDetails on block allocation --- ...ocationProtocolServerSideTranslatorPB.java | 56 +++- ...ocationProtocolServerSideTranslatorPB.java | 290 ++++++++++++++++++ 2 files changed, 345 insertions(+), 1 deletion(-) create mode 100644 hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java 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 007376670ba4..ee2f643484c9 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,7 +20,9 @@ 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.annotation.InterfaceAudience; import org.apache.hadoop.hdds.client.ReplicationConfig; @@ -47,6 +49,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 +215,64 @@ public AllocateScmBlockResponseProto allocateScmBlock( " blocks. Requested " + request.getNumBlocks() + " blocks", SCMException.ResultCodes.FAILED_TO_ALLOCATE_ENOUGH_BLOCKS); } + Map 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( + HddsProtos.Pipeline proto, int writeVersion) { + HddsProtos.Pipeline.Builder builder = proto.toBuilder(); + for (int i = 0; i < builder.getMembersCount(); i++) { + builder.setMembers(i, + builder.getMembers(i).toBuilder().setCurrentVersion(writeVersion)); + } + return builder.build(); + } + 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 000000000000..7f424e5c3463 --- /dev/null +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/protocol/TestScmBlockLocationProtocolServerSideTranslatorPB.java @@ -0,0 +1,290 @@ +/* + * 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 StorageContainerManager scm; + private NodeManager nodeManager; + private ScmBlockLocationProtocolServerSideTranslatorPB service; + private List nodes; + + @BeforeEach + void setUp() throws Exception { + impl = mock(ScmBlockLocationProtocol.class); + 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 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 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 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 blocks) + throws Exception { + when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class), + anyString(), any(), anyString())).thenReturn(blocks); + } + + private void assertAllMembersHaveVersion(int expected, + List 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(), + allocateAndGetMembers()); + } + + @Test + void unfinalizedDatanodeClampsToMinimum() throws Exception { + // SCM is finalized, but one straggler datanode is still at an older + // apparent version; the client must be clamped down to it. + 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); + + IllegalStateException e = + assertThrows(IllegalStateException.class, () -> allocate(1)); + assertInstanceOf(NodeNotFoundException.class, e.getCause()); + } + + @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 clamped to an older + // version than the default (ZDU) pipeline built in setUp(). + List otherNodes = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + DatanodeDetails dn = randomDatanodeDetails(); + dn.setCurrentVersion(DN_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(HDDSVersion.ZDU.serialize(), + 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 { + for (DatanodeDetails dn : nodes) { + setDatanodeApparentVersion(dn, 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(DN_SOFTWARE_VERSION, dn.getCurrentVersion()); + } + } +} From c94e46b13d98f89481c34205403a780c8ce6a930 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Thu, 23 Jul 2026 16:31:13 +0200 Subject: [PATCH 2/5] Fix pmd --- .../TestScmBlockLocationProtocolServerSideTranslatorPB.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 index 7f424e5c3463..027d392a07c9 100644 --- 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 @@ -71,7 +71,6 @@ class TestScmBlockLocationProtocolServerSideTranslatorPB { HDDSVersion.ZDU.serialize(); private ScmBlockLocationProtocol impl; - private StorageContainerManager scm; private NodeManager nodeManager; private ScmBlockLocationProtocolServerSideTranslatorPB service; private List nodes; @@ -79,7 +78,7 @@ class TestScmBlockLocationProtocolServerSideTranslatorPB { @BeforeEach void setUp() throws Exception { impl = mock(ScmBlockLocationProtocol.class); - scm = mock(StorageContainerManager.class); + StorageContainerManager scm = mock(StorageContainerManager.class); nodeManager = mock(NodeManager.class); nodes = new ArrayList<>(); From de14c1013b4fac682ab41dda163a23812e06b61a Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Fri, 24 Jul 2026 12:43:43 +0200 Subject: [PATCH 3/5] Set apparent version in TestBlockDataStreamOutput --- .../hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java | 1 + 1 file changed, 1 insertion(+) 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 4a57e96b7cc4..bbc26177f35d 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(DN_OLD_VERSION.serialize()) .build()) .build(); cluster.waitForPipelineTobeReady(HddsProtos.ReplicationFactor.THREE, From a66c18bbf9be846d11781ab42b7b72c900ae9c04 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Fri, 24 Jul 2026 15:15:31 +0200 Subject: [PATCH 4/5] Set SCM apparent version for the test --- .../hadoop/ozone/client/rpc/TestBlockDataStreamOutput.java | 3 +++ 1 file changed, 3 insertions(+) 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 bbc26177f35d..62a6d4bdc6ec 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 @@ -47,6 +47,7 @@ import org.apache.hadoop.hdds.scm.OzoneClientConfig; import org.apache.hadoop.hdds.scm.XceiverClientManager; import org.apache.hadoop.hdds.scm.XceiverClientMetrics; +import org.apache.hadoop.hdds.scm.server.SCMStorageConfig; import org.apache.hadoop.hdds.scm.storage.BlockDataStreamOutput; import org.apache.hadoop.hdds.scm.storage.ByteBufferStreamOutput; import org.apache.hadoop.ozone.ClientConfigForTesting; @@ -103,6 +104,8 @@ static MiniOzoneCluster createCluster() throws IOException, conf.setBoolean(OzoneConfigKeys.OZONE_HBASE_ENHANCEMENTS_ALLOWED, true); conf.setBoolean("ozone.client.hbase.enhancements.allowed", true); + conf.setInt(SCMStorageConfig.TESTING_INIT_APPARENT_VERSION_KEY, DN_OLD_VERSION.serialize()); + DatanodeRatisServerConfig ratisServerConfig = conf.getObject(DatanodeRatisServerConfig.class); ratisServerConfig.setRequestTimeOut(Duration.ofSeconds(3)); From 5a0a9f372ec29be7e422dc8f8888bc74782dd9a9 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Fri, 31 Jul 2026 15:57:45 +0200 Subject: [PATCH 5/5] Address review comments --- .../hadoop/hdds/protocol/DatanodeDetails.java | 23 ++- .../hadoop/hdds/scm/pipeline/Pipeline.java | 8 +- ...ocationProtocolServerSideTranslatorPB.java | 68 ++++---- ...ocationProtocolServerSideTranslatorPB.java | 152 ++++++++---------- 4 files changed, 121 insertions(+), 130 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 4b0108aa9aeb..98b51ef705a6 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 return toProtoBuilder(clientVersion, filterPorts).build(); } + public HddsProtos.DatanodeDetailsProto toProto(int clientVersion, Set filterPorts, + ComponentVersion versionOverride) { + return toProtoBuilder(clientVersion, filterPorts, versionOverride).build(); + } + + public HddsProtos.DatanodeDetailsProto.Builder toProtoBuilder( + int clientVersion, Set 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 filterPorts) { + int clientVersion, Set 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 62c3855f1af5..5675787c003b 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 filterPorts) { + return getProtobufMessage(clientVersion, filterPorts, null); + } + + public HddsProtos.Pipeline getProtobufMessage(int clientVersion, Set filterPorts, + ComponentVersion versionOverride) { List members = new ArrayList<>(); List 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 ee2f643484c9..9fb688fb8fee 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 @@ -24,6 +24,8 @@ 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; @@ -220,15 +222,8 @@ public AllocateScmBlockResponseProto allocateScmBlock( 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); - } + pipelineProto = pipeline.getProtobufMessage(clientVersion, Name.IO_PORTS, + computePipelineWriteVersion(pipeline)); pipelineProtoCache.put(pipeline.getId(), pipelineProto); } builder.addBlocks(AllocateBlockResponse.newBuilder() @@ -240,37 +235,34 @@ public AllocateScmBlockResponseProto allocateScmBlock( } /** - * 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. + * 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 static HddsProtos.Pipeline withWriteVersion( - HddsProtos.Pipeline proto, int writeVersion) { - HddsProtos.Pipeline.Builder builder = proto.toBuilder(); - for (int i = 0; i < builder.getMembersCount(); i++) { - builder.setMembers(i, - builder.getMembers(i).toBuilder().setCurrentVersion(writeVersion)); + private ComponentVersion computePipelineWriteVersion(Pipeline pipeline) throws SCMException { + List 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); } - return builder.build(); } public DeleteScmKeyBlocksResponseProto deleteScmKeyBlocks( 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 index 027d392a07c9..6b7674d76429 100644 --- 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 @@ -19,7 +19,6 @@ 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; @@ -33,6 +32,7 @@ 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; @@ -46,13 +46,15 @@ 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.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.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; @@ -67,11 +69,11 @@ class TestScmBlockLocationProtocolServerSideTranslatorPB { /** The version each source datanode reports as its own software version. */ - private static final int DN_SOFTWARE_VERSION = - HDDSVersion.ZDU.serialize(); + private static final int SOFTWARE_VERSION = HDDSVersion.SOFTWARE_VERSION.serialize(); private ScmBlockLocationProtocol impl; private NodeManager nodeManager; + private ScmVersionManager versionManager; private ScmBlockLocationProtocolServerSideTranslatorPB service; private List nodes; @@ -80,46 +82,35 @@ 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(DN_SOFTWARE_VERSION); + dn.setCurrentVersion(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(); + Pipeline pipeline = buildPipeline(nodes); + AllocatedBlock block = blockOn(1L, pipeline); - List blocks = new ArrayList<>(); - blocks.add(block); when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class), - anyString(), any(), anyString())).thenReturn(blocks); + 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(); + lenient().when(nodeManager.getLowestApparentVersion(any(DatanodeDetails[].class))).thenCallRealMethod(); - // Default: every datanode finalized to ZDU. + // 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.ZDU); + setDatanodeApparentVersion(dn, HDDSVersion.SOFTWARE_VERSION); } - service = new ScmBlockLocationProtocolServerSideTranslatorPB( - impl, scm, mock(ProtocolMessageMetrics.class)); + service = new ScmBlockLocationProtocolServerSideTranslatorPB(impl, scm, mock(ProtocolMessageMetrics.class)); } - private void setDatanodeApparentVersion(DatanodeDetails dn, - ComponentVersion version) { + 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); @@ -129,16 +120,14 @@ private List 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(); + 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()); } @@ -146,8 +135,7 @@ private Pipeline buildPipeline(List pipelineNodes) { return Pipeline.newBuilder() .setId(PipelineID.randomId()) .setState(PipelineState.OPEN) - .setReplicationConfig( - RatisReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setReplicationConfig(RatisReplicationConfig.getInstance(ReplicationFactor.THREE)) .setNodes(pipelineNodes) .build(); } @@ -159,14 +147,12 @@ private AllocatedBlock blockOn(long localId, Pipeline pipeline) { .build(); } - private void setAllocatedBlocks(List blocks) - throws Exception { + private void setAllocatedBlocks(List blocks) throws Exception { when(impl.allocateBlock(anyLong(), anyInt(), any(ReplicationConfig.class), anyString(), any(), anyString())).thenReturn(blocks); } - private void assertAllMembersHaveVersion(int expected, - List members) { + private void assertAllMembersHaveVersion(int expected, List members) { assertEquals(nodes.size(), members.size()); for (DatanodeDetailsProto member : members) { assertEquals(expected, member.getCurrentVersion()); @@ -175,62 +161,63 @@ private void assertAllMembersHaveVersion(int expected, @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. + // 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, HDDSVersion.STREAM_BLOCK_SUPPORT); + setDatanodeApparentVersion(dn, HDDSLayoutFeature.STORAGE_SPACE_DISTRIBUTION); } - assertAllMembersHaveVersion(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(), - allocateAndGetMembers()); + assertAllMembersHaveVersion(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(), allocateAndGetMembers()); } @Test void finalizedClusterForwardsRealVersion() throws Exception { - // Default setup: SCM and all datanodes at ZDU. - assertAllMembersHaveVersion(HDDSVersion.ZDU.serialize(), - allocateAndGetMembers()); + assertAllMembersHaveVersion(SOFTWARE_VERSION, allocateAndGetMembers()); } @Test - void unfinalizedDatanodeClampsToMinimum() throws Exception { - // SCM is finalized, but one straggler datanode is still at an older - // apparent version; the client must be clamped down to it. - setDatanodeApparentVersion(nodes.get(1), - HDDSVersion.COMBINED_PUTBLOCK_WRITECHUNK_RPC); - - assertAllMembersHaveVersion( - HDDSVersion.COMBINED_PUTBLOCK_WRITECHUNK_RPC.serialize(), - allocateAndGetMembers()); + 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); - IllegalStateException e = - assertThrows(IllegalStateException.class, () -> allocate(1)); - assertInstanceOf(NodeNotFoundException.class, e.getCause()); + 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); + 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))); + 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(), + assertAllMembersHaveVersion(HDDSVersion.COMBINED_PUTBLOCK_WRITECHUNK_RPC.serialize(), response.getBlocks(i).getPipeline().getMembersList()); } @@ -243,47 +230,40 @@ void blocksSharingPipelineAllGetClampedVersion() throws Exception { @Test void blocksOnDistinctPipelinesGetOwnVersion() throws Exception { - // A second, distinct pipeline whose datanodes are clamped to an older - // version than the default (ZDU) pipeline built in setUp(). + // A second, distinct pipeline whose datanodes are at an older + // version than the software-version pipeline built in setUp(). List otherNodes = new ArrayList<>(); for (int i = 0; i < 3; i++) { DatanodeDetails dn = randomDatanodeDetails(); - dn.setCurrentVersion(DN_SOFTWARE_VERSION); + 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))); + setAllocatedBlocks(Arrays.asList(blockOn(1L, finalized), blockOn(2L, straggler))); AllocateScmBlockResponseProto response = allocate(2); assertEquals(2, response.getBlocksCount()); - assertAllMembersHaveVersion(HDDSVersion.ZDU.serialize(), - 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()); + 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 { - for (DatanodeDetails dn : nodes) { - setDatanodeApparentVersion(dn, HDDSVersion.STREAM_BLOCK_SUPPORT); - } + 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(DN_SOFTWARE_VERSION, dn.getCurrentVersion()); + assertEquals(SOFTWARE_VERSION, dn.getCurrentVersion()); } } }