From 61d8375b6a006381d91b4d17bb5bfd45f00467d7 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Thu, 18 Jun 2026 14:36:01 +0200 Subject: [PATCH 1/9] HDDS-15417. Handle mixed datanode versions on the replication path --- .../ReplicateContainerCommandHandler.java | 6 + .../replication/ReplicationTask.java | 11 ++ .../commands/ReplicateContainerCommand.java | 17 +++ .../TestReplicateContainerCommandHandler.java | 108 ++++++++++++++++++ .../TestReplicateContainerCommand.java | 104 +++++++++++++++++ .../ScmServerDatanodeHeartbeatProtocol.proto | 1 + .../replication/ReplicationManager.java | 27 +++++ .../replication/TestReplicationManager.java | 76 ++++++++++++ 8 files changed, 350 insertions(+) create mode 100644 hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java index 135c6fdb0391..391b42f5c046 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java @@ -84,6 +84,12 @@ public void handle(SCMCommand command, OzoneContainer container, downloadReplicator : pushReplicator; ReplicationTask task = new ReplicationTask(replicateCommand, replicator); + + int serverVersion = replicateCommand.getPeerApparentVersion(); + int localVersion = context.getParent().getVersionManager() + .getApparentVersion().serialize(); + task.setLowestCommonApparentVersion(Math.min(localVersion, serverVersion)); + supervisor.addTask(task); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java index a32e9b41ab1b..0a7b0dbfe46e 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java @@ -19,6 +19,7 @@ import java.util.List; import java.util.Objects; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; @@ -33,6 +34,8 @@ public class ReplicationTask extends AbstractReplicationTask { public static final String METRIC_NAME = "ContainerReplications"; public static final String METRIC_DESCRIPTION_SEGMENT = "container replications"; + private int lowestCommonApparentVersion = HDDSVersion.DEFAULT_VERSION.serialize(); + /** * Counter for the transferred bytes. */ @@ -125,6 +128,14 @@ public void setTransferredBytes(long transferredBytes) { this.transferredBytes = transferredBytes; } + public int getLowestCommonApparentVersion() { + return lowestCommonApparentVersion; + } + + public void setLowestCommonApparentVersion(int lowestCommonApparentVersion) { + this.lowestCommonApparentVersion = lowestCommonApparentVersion; + } + DatanodeDetails getTarget() { return cmd.getTargetDatanode(); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java index bc8040b24bfc..a4612fccef42 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java @@ -22,6 +22,7 @@ import java.util.List; import java.util.Objects; import java.util.stream.Collectors; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicateContainerCommandProto; @@ -42,6 +43,7 @@ public final class ReplicateContainerCommand private int replicaIndex = 0; private ReplicationCommandPriority priority = ReplicationCommandPriority.NORMAL; + private int peerApparentVersion = HDDSVersion.DEFAULT_VERSION.serialize(); public static ReplicateContainerCommand fromSources(long containerID, List sourceDatanodes) { @@ -82,6 +84,14 @@ public void setPriority(ReplicationCommandPriority priority) { this.priority = priority; } + public void setPeerApparentVersion(int version) { + this.peerApparentVersion = version; + } + + public int getPeerApparentVersion() { + return peerApparentVersion; + } + @Override public Type getType() { return SCMCommandProto.Type.replicateContainerCommand; @@ -105,6 +115,7 @@ public ReplicateContainerCommandProto getProto() { builder.setTarget(targetDatanode.getProtoBufMessage()); } builder.setPriority(priority); + builder.setPeerApparentVersion(peerApparentVersion); return builder.build(); } @@ -131,6 +142,10 @@ public static ReplicateContainerCommand getFromProtobuf( if (protoMessage.hasPriority()) { cmd.setPriority(protoMessage.getPriority()); } + if (protoMessage.hasPeerApparentVersion()) { + cmd.setPeerApparentVersion( + protoMessage.getPeerApparentVersion()); + } return cmd; } @@ -170,6 +185,8 @@ public String toString() { sb.append(", sourceNodes=").append(sourceDatanodes); } sb.append(", priority=").append(priority); + sb.append(", peerApparentVersion=") + .append(peerApparentVersion); return sb.toString(); } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java index b88b6da7ea7d..8dd50d4fe949 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java @@ -27,21 +27,25 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl; import org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics; +import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager; import org.apache.hadoop.ozone.container.common.statemachine.StateContext; import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; import org.apache.hadoop.ozone.container.replication.ContainerReplicator; import org.apache.hadoop.ozone.container.replication.ReplicationSupervisor; import org.apache.hadoop.ozone.container.replication.ReplicationTask; +import org.apache.hadoop.ozone.container.upgrade.DatanodeVersionManager; import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; /** * Test cases to verify {@link ReplicateContainerCommandHandler}. @@ -64,6 +68,14 @@ public void setUp() { ozoneContainer = mock(OzoneContainer.class); connectionManager = mock(SCMConnectionManager.class); stateContext = mock(StateContext.class); + + DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); + DatanodeVersionManager versionManager = + mock(DatanodeVersionManager.class); + when(stateContext.getParent()).thenReturn(dsm); + when(dsm.getVersionManager()).thenReturn(versionManager); + when(versionManager.getApparentVersion()) + .thenReturn(HDDSVersion.SOFTWARE_VERSION); } @Test @@ -116,4 +128,100 @@ public void testMetrics() { metrics.unRegister(); } } + + @Test + public void testEffectiveVersionServerIsOlder() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(ReplicationTask.class); + doNothing().when(supervisor).addTask(captor.capture()); + + DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); + DatanodeVersionManager versionManager = + mock(DatanodeVersionManager.class); + when(stateContext.getParent()).thenReturn(dsm); + when(dsm.getVersionManager()).thenReturn(versionManager); + when(versionManager.getApparentVersion()) + .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); + + ReplicateContainerCommandHandler handler = + new ReplicateContainerCommandHandler(conf, supervisor, + downloadReplicator, pushReplicator); + + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + List sources = new ArrayList<>(); + sources.add(source); + + ReplicateContainerCommand cmd = + ReplicateContainerCommand.fromSources(1, sources); + cmd.setPeerApparentVersion( + HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE.serialize()); + + handler.handle(cmd, ozoneContainer, stateContext, connectionManager); + + ReplicationTask task = captor.getValue(); + assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE.serialize(), + task.getLowestCommonApparentVersion()); + } + + @Test + public void testEffectiveVersionLocalIsOlder() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(ReplicationTask.class); + doNothing().when(supervisor).addTask(captor.capture()); + + DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); + DatanodeVersionManager versionManager = + mock(DatanodeVersionManager.class); + when(stateContext.getParent()).thenReturn(dsm); + when(dsm.getVersionManager()).thenReturn(versionManager); + when(versionManager.getApparentVersion()) + .thenReturn(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); + + ReplicateContainerCommandHandler handler = + new ReplicateContainerCommandHandler(conf, supervisor, + downloadReplicator, pushReplicator); + + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + ReplicateContainerCommand cmd = + ReplicateContainerCommand.toTarget(1, target); + cmd.setPeerApparentVersion(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize()); + + handler.handle(cmd, ozoneContainer, stateContext, connectionManager); + + ReplicationTask task = captor.getValue(); + assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE.serialize(), + task.getLowestCommonApparentVersion()); + } + + @Test + public void testEffectiveVersionDefaultWhenNotSet() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(ReplicationTask.class); + doNothing().when(supervisor).addTask(captor.capture()); + + DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); + DatanodeVersionManager versionManager = + mock(DatanodeVersionManager.class); + when(stateContext.getParent()).thenReturn(dsm); + when(dsm.getVersionManager()).thenReturn(versionManager); + when(versionManager.getApparentVersion()) + .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); + + ReplicateContainerCommandHandler handler = + new ReplicateContainerCommandHandler(conf, supervisor, + downloadReplicator, pushReplicator); + + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + List sources = new ArrayList<>(); + sources.add(source); + + ReplicateContainerCommand cmd = + ReplicateContainerCommand.fromSources(1, sources); + + handler.handle(cmd, ozoneContainer, stateContext, connectionManager); + + ReplicationTask task = captor.getValue(); + assertEquals(HDDSVersion.DEFAULT_VERSION.serialize(), + task.getLowestCommonApparentVersion()); + } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java new file mode 100644 index 000000000000..641f6c910ac9 --- /dev/null +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java @@ -0,0 +1,104 @@ +/* + * 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.ozone.protocol.commands; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +import java.util.Collections; +import java.util.List; +import org.apache.hadoop.hdds.HDDSVersion; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; +import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicateContainerCommandProto; +import org.junit.jupiter.api.Test; + +/** + * Test cases to verify {@link ReplicateContainerCommand} serialization. + */ +public class TestReplicateContainerCommand { + + @Test + public void testServerApparentVersionRoundTrip() { + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + List sources = Collections.singletonList(source); + + ReplicateContainerCommand cmd = + ReplicateContainerCommand.fromSources(1L, sources); + cmd.setPeerApparentVersion( + HDDSVersion.STREAM_BLOCK_SUPPORT.serialize()); + + ReplicateContainerCommandProto proto = cmd.getProto(); + ReplicateContainerCommand deserialized = + ReplicateContainerCommand.getFromProtobuf(proto); + + assertEquals(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(), + deserialized.getPeerApparentVersion()); + } + + @Test + public void testServerApparentVersionDefaultWhenAbsent() { + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + + ReplicateContainerCommand cmd = + ReplicateContainerCommand.toTarget(1L, target); + + ReplicateContainerCommandProto proto = + ReplicateContainerCommandProto.newBuilder() + .setContainerID(1L) + .setCmdId(cmd.getId()) + .setTarget(target.getProtoBufMessage()) + .build(); + + ReplicateContainerCommand deserialized = + ReplicateContainerCommand.getFromProtobuf(proto); + + assertEquals(HDDSVersion.DEFAULT_VERSION.serialize(), + deserialized.getPeerApparentVersion()); + } + + @Test + public void testServerApparentVersionInPushMode() { + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + + ReplicateContainerCommand cmd = + ReplicateContainerCommand.toTarget(1L, target); + cmd.setPeerApparentVersion(HDDSVersion.ZDU.serialize()); + + ReplicateContainerCommandProto proto = cmd.getProto(); + ReplicateContainerCommand deserialized = + ReplicateContainerCommand.getFromProtobuf(proto); + + assertEquals(HDDSVersion.ZDU.serialize(), + deserialized.getPeerApparentVersion()); + assertEquals(target.getUuid(), + deserialized.getTargetDatanode().getUuid()); + } + + @Test + public void testToStringIncludesServerApparentVersion() { + ReplicateContainerCommand cmd = + ReplicateContainerCommand.forTest(1L); + cmd.setPeerApparentVersion( + HDDSVersion.STREAM_BLOCK_SUPPORT.serialize()); + + String str = cmd.toString(); + assertEquals(true, + str.contains("peerApparentVersion=" + + HDDSVersion.STREAM_BLOCK_SUPPORT.serialize())); + } +} diff --git a/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto b/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto index 3cc92e6b9961..8326f5a35447 100644 --- a/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto +++ b/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto @@ -429,6 +429,7 @@ message ReplicateContainerCommandProto { optional int32 replicaIndex = 4; optional DatanodeDetailsProto target = 5; optional ReplicationCommandPriority priority = 6 [default = NORMAL]; + optional uint32 peerApparentVersion = 7; } /** diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java index 8cd8444d1d2f..a2eef56ba11a 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java @@ -41,6 +41,7 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.client.ReplicationConfig; import org.apache.hadoop.hdds.conf.Config; @@ -78,6 +79,7 @@ import org.apache.hadoop.hdds.scm.events.SCMEvents; import org.apache.hadoop.hdds.scm.ha.SCMContext; import org.apache.hadoop.hdds.scm.ha.SCMService; +import org.apache.hadoop.hdds.scm.node.DatanodeInfo; import org.apache.hadoop.hdds.scm.node.NodeManager; import org.apache.hadoop.hdds.scm.node.NodeStatus; import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException; @@ -670,11 +672,36 @@ public void sendDatanodeCommand(SCMCommand command, scmDeadlineEpochMs); command.setTerm(getScmTerm()); command.setDeadline(datanodeDeadline); + if (command.getType() == Type.replicateContainerCommand) { + enrichReplicateCommandWithPeerVersion( + (ReplicateContainerCommand) command); + } nodeManager.addDatanodeCommand(target.getID(), command); adjustPendingOpsAndMetrics(containerInfo, command, target, scmDeadlineEpochMs); } + private void enrichReplicateCommandWithPeerVersion( + ReplicateContainerCommand cmd) { + if (cmd.getTargetDatanode() != null) { + cmd.setPeerApparentVersion( + lookupApparentVersion(cmd.getTargetDatanode())); + } else { + int min = cmd.getSourceDatanodes().stream() + .mapToInt(this::lookupApparentVersion) + .min() + .orElse(HDDSVersion.DEFAULT_VERSION.serialize()); + cmd.setPeerApparentVersion(min); + } + } + + private int lookupApparentVersion(DatanodeDetails dn) { + DatanodeInfo info = nodeManager.getDatanodeInfo(dn); + return info != null + ? info.getLastKnownApparentVersion().serialize() + : HDDSVersion.DEFAULT_VERSION.serialize(); + } + private void adjustPendingOpsAndMetrics(ContainerInfo containerInfo, SCMCommand cmd, DatanodeDetails targetDatanode, long scmDeadlineEpochMs) { diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java index 1d7d619efb0f..592a1701cbaf 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java @@ -87,8 +87,10 @@ import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementStatusDefault; import org.apache.hadoop.hdds.scm.events.SCMEvents; import org.apache.hadoop.hdds.scm.exceptions.SCMException; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.scm.ha.SCMContext; import org.apache.hadoop.hdds.scm.ha.SCMServiceManager; +import org.apache.hadoop.hdds.scm.node.DatanodeInfo; import org.apache.hadoop.hdds.scm.node.NodeManager; import org.apache.hadoop.hdds.scm.node.NodeStatus; import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException; @@ -1801,4 +1803,78 @@ private void mockReplicationCommandCounts( }); } + @Test + public void testPeerApparentVersionSetForPushCommand() + throws NotLeaderException { + ContainerInfo containerInfo = + ReplicationTestUtil.createContainerInfo(repConfig, 1, + HddsProtos.LifeCycleState.CLOSED, 10, 20); + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + + DatanodeInfo targetInfo = mock(DatanodeInfo.class); + when(targetInfo.getLastKnownApparentVersion()) + .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); + when(nodeManager.getDatanodeInfo(target)).thenReturn(targetInfo); + + ReplicateContainerCommand command = + ReplicateContainerCommand.toTarget( + containerInfo.getContainerID(), target); + command.setReplicaIndex(1); + replicationManager.sendDatanodeCommand(command, containerInfo, source); + + assertEquals(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(), + command.getPeerApparentVersion()); + } + + @Test + public void testPeerApparentVersionSetForPullCommand() + throws NotLeaderException { + ContainerInfo containerInfo = + ReplicationTestUtil.createContainerInfo(repConfig, 1, + HddsProtos.LifeCycleState.CLOSED, 10, 20); + DatanodeDetails source1 = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source2 = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + + DatanodeInfo source1Info = mock(DatanodeInfo.class); + when(source1Info.getLastKnownApparentVersion()) + .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); + DatanodeInfo source2Info = mock(DatanodeInfo.class); + when(source2Info.getLastKnownApparentVersion()) + .thenReturn(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); + + when(nodeManager.getDatanodeInfo(source1)).thenReturn(source1Info); + when(nodeManager.getDatanodeInfo(source2)).thenReturn(source2Info); + + ReplicateContainerCommand command = + ReplicateContainerCommand.fromSources( + containerInfo.getContainerID(), + Lists.newArrayList(source1, source2)); + replicationManager.sendDatanodeCommand(command, containerInfo, target); + + assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE.serialize(), + command.getPeerApparentVersion()); + } + + @Test + public void testPeerApparentVersionDefaultWhenNodeNotFound() + throws NotLeaderException { + ContainerInfo containerInfo = + ReplicationTestUtil.createContainerInfo(repConfig, 1, + HddsProtos.LifeCycleState.CLOSED, 10, 20); + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + + when(nodeManager.getDatanodeInfo(target)).thenReturn(null); + + ReplicateContainerCommand command = + ReplicateContainerCommand.toTarget( + containerInfo.getContainerID(), target); + replicationManager.sendDatanodeCommand(command, containerInfo, source); + + assertEquals(HDDSVersion.DEFAULT_VERSION.serialize(), + command.getPeerApparentVersion()); + } + } From 91a48ae7b9073e77f84df4b3c44a28a7aaa291a4 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Thu, 18 Jun 2026 14:49:48 +0200 Subject: [PATCH 2/9] Fix checkstyle and pmd --- .../ozone/protocol/commands/ReplicateContainerCommand.java | 5 ++--- .../protocol/commands/TestReplicateContainerCommand.java | 6 +++--- .../scm/container/replication/TestReplicationManager.java | 2 +- 3 files changed, 6 insertions(+), 7 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java index a4612fccef42..2656bd08a30b 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java @@ -184,9 +184,8 @@ public String toString() { } else { sb.append(", sourceNodes=").append(sourceDatanodes); } - sb.append(", priority=").append(priority); - sb.append(", peerApparentVersion=") - .append(peerApparentVersion); + sb.append(", priority=").append(priority) + .append(", peerApparentVersion=").append(peerApparentVersion); return sb.toString(); } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java index 641f6c910ac9..45c89fea8eb3 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java @@ -18,6 +18,7 @@ package org.apache.hadoop.ozone.protocol.commands; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.Collections; import java.util.List; @@ -97,8 +98,7 @@ public void testToStringIncludesServerApparentVersion() { HDDSVersion.STREAM_BLOCK_SUPPORT.serialize()); String str = cmd.toString(); - assertEquals(true, - str.contains("peerApparentVersion=" - + HDDSVersion.STREAM_BLOCK_SUPPORT.serialize())); + assertTrue(str.contains("peerApparentVersion=" + + HDDSVersion.STREAM_BLOCK_SUPPORT.serialize())); } } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java index 592a1701cbaf..592c391e4dd5 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java @@ -65,6 +65,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationConfig; @@ -87,7 +88,6 @@ import org.apache.hadoop.hdds.scm.container.placement.algorithms.ContainerPlacementStatusDefault; import org.apache.hadoop.hdds.scm.events.SCMEvents; import org.apache.hadoop.hdds.scm.exceptions.SCMException; -import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.scm.ha.SCMContext; import org.apache.hadoop.hdds.scm.ha.SCMServiceManager; import org.apache.hadoop.hdds.scm.node.DatanodeInfo; From ecbe86dd1455d11269b05c43c5ce15a77ea8aef4 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Mon, 22 Jun 2026 14:38:49 +0200 Subject: [PATCH 3/9] Handle UNKNOWN_VERSION better, make test names more consitent, add comment to getLowestCommonApparentVersion. --- .../ReplicateContainerCommandHandler.java | 4 +- .../replication/ReplicationTask.java | 7 +++ .../TestReplicateContainerCommandHandler.java | 47 +++++++------------ .../TestReplicateContainerCommand.java | 8 ++-- .../replication/ReplicationManager.java | 24 +++++++--- .../replication/TestReplicationManager.java | 26 ++++++++++ 6 files changed, 73 insertions(+), 43 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java index 391b42f5c046..80e3a01e6b26 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java @@ -85,10 +85,10 @@ public void handle(SCMCommand command, OzoneContainer container, ReplicationTask task = new ReplicationTask(replicateCommand, replicator); - int serverVersion = replicateCommand.getPeerApparentVersion(); + int peerVersion = replicateCommand.getPeerApparentVersion(); int localVersion = context.getParent().getVersionManager() .getApparentVersion().serialize(); - task.setLowestCommonApparentVersion(Math.min(localVersion, serverVersion)); + task.setLowestCommonApparentVersion(Math.min(localVersion, peerVersion)); supervisor.addTask(task); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java index 0a7b0dbfe46e..2cf4a6deea84 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java @@ -128,6 +128,13 @@ public void setTransferredBytes(long transferredBytes) { this.transferredBytes = transferredBytes; } + /** + * @return the lowest apparent version supported by both this datanode and + * the replication peer, computed as {@code min(localApparentVersion, + * peerApparentVersion)}. Replicators can use this to gate + * version-dependent protocol features so that the newer side downgrades + * to a behavior the older side understands. + */ public int getLowestCommonApparentVersion() { return lowestCommonApparentVersion; } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java index 8dd50d4fe949..14d154b8915d 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java @@ -58,6 +58,7 @@ public class TestReplicateContainerCommandHandler { private OzoneContainer ozoneContainer; private StateContext stateContext; private SCMConnectionManager connectionManager; + private DatanodeVersionManager versionManager; @BeforeEach public void setUp() { @@ -70,14 +71,24 @@ public void setUp() { stateContext = mock(StateContext.class); DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); - DatanodeVersionManager versionManager = - mock(DatanodeVersionManager.class); + versionManager = mock(DatanodeVersionManager.class); when(stateContext.getParent()).thenReturn(dsm); when(dsm.getVersionManager()).thenReturn(versionManager); when(versionManager.getApparentVersion()) .thenReturn(HDDSVersion.SOFTWARE_VERSION); } + /** + * Stubs {@link ReplicationSupervisor#addTask} and returns a captor for the + * {@link ReplicationTask} the handler submits. + */ + private ArgumentCaptor captureSubmittedTask() { + ArgumentCaptor captor = + ArgumentCaptor.forClass(ReplicationTask.class); + doNothing().when(supervisor).addTask(captor.capture()); + return captor; + } + @Test public void testMetrics() { ReplicateContainerCommandHandler commandHandler = @@ -130,16 +141,8 @@ public void testMetrics() { } @Test - public void testEffectiveVersionServerIsOlder() { - ArgumentCaptor captor = - ArgumentCaptor.forClass(ReplicationTask.class); - doNothing().when(supervisor).addTask(captor.capture()); - - DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); - DatanodeVersionManager versionManager = - mock(DatanodeVersionManager.class); - when(stateContext.getParent()).thenReturn(dsm); - when(dsm.getVersionManager()).thenReturn(versionManager); + public void testEffectiveVersionPeerIsOlder() { + ArgumentCaptor captor = captureSubmittedTask(); when(versionManager.getApparentVersion()) .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); @@ -165,15 +168,7 @@ public void testEffectiveVersionServerIsOlder() { @Test public void testEffectiveVersionLocalIsOlder() { - ArgumentCaptor captor = - ArgumentCaptor.forClass(ReplicationTask.class); - doNothing().when(supervisor).addTask(captor.capture()); - - DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); - DatanodeVersionManager versionManager = - mock(DatanodeVersionManager.class); - when(stateContext.getParent()).thenReturn(dsm); - when(dsm.getVersionManager()).thenReturn(versionManager); + ArgumentCaptor captor = captureSubmittedTask(); when(versionManager.getApparentVersion()) .thenReturn(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); @@ -195,15 +190,7 @@ public void testEffectiveVersionLocalIsOlder() { @Test public void testEffectiveVersionDefaultWhenNotSet() { - ArgumentCaptor captor = - ArgumentCaptor.forClass(ReplicationTask.class); - doNothing().when(supervisor).addTask(captor.capture()); - - DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); - DatanodeVersionManager versionManager = - mock(DatanodeVersionManager.class); - when(stateContext.getParent()).thenReturn(dsm); - when(dsm.getVersionManager()).thenReturn(versionManager); + ArgumentCaptor captor = captureSubmittedTask(); when(versionManager.getApparentVersion()) .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java index 45c89fea8eb3..ccc430e46595 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java @@ -34,7 +34,7 @@ public class TestReplicateContainerCommand { @Test - public void testServerApparentVersionRoundTrip() { + public void testPeerApparentVersionRoundTrip() { DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); List sources = Collections.singletonList(source); @@ -52,7 +52,7 @@ public void testServerApparentVersionRoundTrip() { } @Test - public void testServerApparentVersionDefaultWhenAbsent() { + public void testPeerApparentVersionDefaultWhenAbsent() { DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); ReplicateContainerCommand cmd = @@ -73,7 +73,7 @@ public void testServerApparentVersionDefaultWhenAbsent() { } @Test - public void testServerApparentVersionInPushMode() { + public void testPeerApparentVersionInPushMode() { DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); ReplicateContainerCommand cmd = @@ -91,7 +91,7 @@ public void testServerApparentVersionInPushMode() { } @Test - public void testToStringIncludesServerApparentVersion() { + public void testToStringIncludesPeerApparentVersion() { ReplicateContainerCommand cmd = ReplicateContainerCommand.forTest(1L); cmd.setPeerApparentVersion( diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java index a2eef56ba11a..42421a15e6bd 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java @@ -41,6 +41,7 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.ComponentVersion; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.client.ReplicationConfig; @@ -666,16 +667,16 @@ public void sendDatanodeCommand(SCMCommand command, throws NotLeaderException { long datanodeDeadline = scmDeadlineEpochMs - rmConf.getDatanodeTimeoutOffset(); + if (command.getType() == Type.replicateContainerCommand) { + enrichReplicateCommandWithPeerVersion( + (ReplicateContainerCommand) command); + } LOG.info("Sending command [{}] for container {} to {} with datanode " + "deadline {} and scm deadline {}", command, containerInfo, target, datanodeDeadline, scmDeadlineEpochMs); command.setTerm(getScmTerm()); command.setDeadline(datanodeDeadline); - if (command.getType() == Type.replicateContainerCommand) { - enrichReplicateCommandWithPeerVersion( - (ReplicateContainerCommand) command); - } nodeManager.addDatanodeCommand(target.getID(), command); adjustPendingOpsAndMetrics(containerInfo, command, target, scmDeadlineEpochMs); @@ -697,9 +698,18 @@ private void enrichReplicateCommandWithPeerVersion( private int lookupApparentVersion(DatanodeDetails dn) { DatanodeInfo info = nodeManager.getDatanodeInfo(dn); - return info != null - ? info.getLastKnownApparentVersion().serialize() - : HDDSVersion.DEFAULT_VERSION.serialize(); + if (info == null) { + return HDDSVersion.DEFAULT_VERSION.serialize(); + } + ComponentVersion peerVersion = info.getLastKnownApparentVersion(); + // A peer reporting a version newer than this SCM can recognize + // deserializes to UNKNOWN_VERSION (-1). Treat it as the oldest known + // version so the client datanode falls back to the most conservative, + // backward-compatible behavior rather than propagating the sentinel. + if (peerVersion == null || peerVersion == HDDSVersion.UNKNOWN_VERSION) { + return HDDSVersion.DEFAULT_VERSION.serialize(); + } + return peerVersion.serialize(); } private void adjustPendingOpsAndMetrics(ContainerInfo containerInfo, diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java index 592c391e4dd5..92e4360287d1 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java @@ -1877,4 +1877,30 @@ public void testPeerApparentVersionDefaultWhenNodeNotFound() command.getPeerApparentVersion()); } + @Test + public void testPeerApparentVersionDefaultWhenPeerVersionUnknown() + throws NotLeaderException { + ContainerInfo containerInfo = + ReplicationTestUtil.createContainerInfo(repConfig, 1, + HddsProtos.LifeCycleState.CLOSED, 10, 20); + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + + // A peer reporting a version newer than this SCM can recognize + // deserializes to UNKNOWN_VERSION; it must fall back to the oldest + // version rather than propagating the sentinel. + DatanodeInfo targetInfo = mock(DatanodeInfo.class); + when(targetInfo.getLastKnownApparentVersion()) + .thenReturn(HDDSVersion.UNKNOWN_VERSION); + when(nodeManager.getDatanodeInfo(target)).thenReturn(targetInfo); + + ReplicateContainerCommand command = + ReplicateContainerCommand.toTarget( + containerInfo.getContainerID(), target); + replicationManager.sendDatanodeCommand(command, containerInfo, source); + + assertEquals(HDDSVersion.DEFAULT_VERSION.serialize(), + command.getPeerApparentVersion()); + } + } From 2990c971c21f4fdf4c456ed14a42d71906171afa Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Tue, 30 Jun 2026 11:20:00 +0200 Subject: [PATCH 4/9] Address review comments --- .../ReplicateContainerCommandHandler.java | 6 - .../replication/ReplicationTask.java | 20 +-- .../commands/ReplicateContainerCommand.java | 37 ++--- .../TestReplicateContainerCommandHandler.java | 86 +++--------- .../TestGrpcReplicationService.java | 3 +- .../replication/TestPushReplicator.java | 7 +- .../TestReplicationSupervisor.java | 4 +- .../TestReplicateContainerCommand.java | 51 ++----- .../ScmServerDatanodeHeartbeatProtocol.proto | 2 +- .../replication/ReplicationManager.java | 96 ++++++++----- .../replication/ReplicationTestUtil.java | 3 +- .../TestContainerReplicaPendingOps.java | 4 +- .../replication/TestReplicationManager.java | 127 ++++++++++-------- .../TestReplicationManagerScenarios.java | 10 ++ .../hdds/scm/node/TestCommandQueue.java | 7 +- .../hdds/scm/node/TestSCMNodeManager.java | 3 +- .../replication/TestContainerReplication.java | 9 +- 17 files changed, 225 insertions(+), 250 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java index 80e3a01e6b26..135c6fdb0391 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java @@ -84,12 +84,6 @@ public void handle(SCMCommand command, OzoneContainer container, downloadReplicator : pushReplicator; ReplicationTask task = new ReplicationTask(replicateCommand, replicator); - - int peerVersion = replicateCommand.getPeerApparentVersion(); - int localVersion = context.getParent().getVersionManager() - .getApparentVersion().serialize(); - task.setLowestCommonApparentVersion(Math.min(localVersion, peerVersion)); - supervisor.addTask(task); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java index 2cf4a6deea84..1b0fc4f2b495 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java @@ -31,11 +31,10 @@ public class ReplicationTask extends AbstractReplicationTask { private final ReplicateContainerCommand cmd; private final ContainerReplicator replicator; private final String debugString; + private final HDDSVersion apparentVersion; public static final String METRIC_NAME = "ContainerReplications"; public static final String METRIC_DESCRIPTION_SEGMENT = "container replications"; - private int lowestCommonApparentVersion = HDDSVersion.DEFAULT_VERSION.serialize(); - /** * Counter for the transferred bytes. */ @@ -47,6 +46,7 @@ public ReplicationTask(ReplicateContainerCommand cmd, setPriority(cmd.getPriority()); this.cmd = cmd; this.replicator = replicator; + this.apparentVersion = cmd.getApparentVersion(); if (cmd.getTargetDatanode() != null) { // Only push replication will have a target datanode set, and it must be // sent to the source datanode to be executed. It is possible the source @@ -129,18 +129,12 @@ public void setTransferredBytes(long transferredBytes) { } /** - * @return the lowest apparent version supported by both this datanode and - * the replication peer, computed as {@code min(localApparentVersion, - * peerApparentVersion)}. Replicators can use this to gate - * version-dependent protocol features so that the newer side downgrades - * to a behavior the older side understands. + * @return the apparent version to use for this replication task. Replicators + * can use this to gate version-dependent protocol features so that the + * newer side downgrades to a behavior the older side understands. */ - public int getLowestCommonApparentVersion() { - return lowestCommonApparentVersion; - } - - public void setLowestCommonApparentVersion(int lowestCommonApparentVersion) { - this.lowestCommonApparentVersion = lowestCommonApparentVersion; + public HDDSVersion getApparentVersion() { + return apparentVersion; } DatanodeDetails getTarget() { diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java index 2656bd08a30b..4cab11b14a78 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java @@ -43,7 +43,7 @@ public final class ReplicateContainerCommand private int replicaIndex = 0; private ReplicationCommandPriority priority = ReplicationCommandPriority.NORMAL; - private int peerApparentVersion = HDDSVersion.DEFAULT_VERSION.serialize(); + private HDDSVersion apparentVersion = HDDSVersion.DEFAULT_VERSION; public static ReplicateContainerCommand fromSources(long containerID, List sourceDatanodes) { @@ -51,12 +51,18 @@ public static ReplicateContainerCommand fromSources(long containerID, } public static ReplicateContainerCommand toTarget(long containerID, - DatanodeDetails target) { - return new ReplicateContainerCommand(containerID, emptyList(), target); + DatanodeDetails target, HDDSVersion apparentVersion) { + ReplicateContainerCommand cmd = + new ReplicateContainerCommand(containerID, emptyList(), target); + cmd.apparentVersion = apparentVersion; + return cmd; } public static ReplicateContainerCommand forTest(long containerID) { - return new ReplicateContainerCommand(containerID, emptyList(), null); + ReplicateContainerCommand cmd = + new ReplicateContainerCommand(containerID, emptyList(), null); + cmd.apparentVersion = HDDSVersion.SOFTWARE_VERSION; + return cmd; } private ReplicateContainerCommand(long containerID, @@ -84,12 +90,13 @@ public void setPriority(ReplicationCommandPriority priority) { this.priority = priority; } - public void setPeerApparentVersion(int version) { - this.peerApparentVersion = version; - } - - public int getPeerApparentVersion() { - return peerApparentVersion; + /** + * @return the apparent version that should be used to carry out this + * replication. SCM computes this as the lowest apparent version among the + * nodes involved. + */ + public HDDSVersion getApparentVersion() { + return apparentVersion; } @Override @@ -115,7 +122,7 @@ public ReplicateContainerCommandProto getProto() { builder.setTarget(targetDatanode.getProtoBufMessage()); } builder.setPriority(priority); - builder.setPeerApparentVersion(peerApparentVersion); + builder.setApparentVersion(apparentVersion.serialize()); return builder.build(); } @@ -142,9 +149,9 @@ public static ReplicateContainerCommand getFromProtobuf( if (protoMessage.hasPriority()) { cmd.setPriority(protoMessage.getPriority()); } - if (protoMessage.hasPeerApparentVersion()) { - cmd.setPeerApparentVersion( - protoMessage.getPeerApparentVersion()); + if (protoMessage.hasApparentVersion()) { + cmd.apparentVersion = + HDDSVersion.deserialize(protoMessage.getApparentVersion()); } return cmd; } @@ -185,7 +192,7 @@ public String toString() { sb.append(", sourceNodes=").append(sourceDatanodes); } sb.append(", priority=").append(priority) - .append(", peerApparentVersion=").append(peerApparentVersion); + .append(", apparentVersion=").append(apparentVersion); return sb.toString(); } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java index 14d154b8915d..41a17779b429 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java @@ -34,14 +34,12 @@ import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; import org.apache.hadoop.metrics2.impl.MetricsCollectorImpl; import org.apache.hadoop.ozone.container.common.helpers.CommandHandlerMetrics; -import org.apache.hadoop.ozone.container.common.statemachine.DatanodeStateMachine; import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager; import org.apache.hadoop.ozone.container.common.statemachine.StateContext; import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; import org.apache.hadoop.ozone.container.replication.ContainerReplicator; import org.apache.hadoop.ozone.container.replication.ReplicationSupervisor; import org.apache.hadoop.ozone.container.replication.ReplicationTask; -import org.apache.hadoop.ozone.container.upgrade.DatanodeVersionManager; import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -58,7 +56,6 @@ public class TestReplicateContainerCommandHandler { private OzoneContainer ozoneContainer; private StateContext stateContext; private SCMConnectionManager connectionManager; - private DatanodeVersionManager versionManager; @BeforeEach public void setUp() { @@ -69,13 +66,6 @@ public void setUp() { ozoneContainer = mock(OzoneContainer.class); connectionManager = mock(SCMConnectionManager.class); stateContext = mock(StateContext.class); - - DatanodeStateMachine dsm = mock(DatanodeStateMachine.class); - versionManager = mock(DatanodeVersionManager.class); - when(stateContext.getParent()).thenReturn(dsm); - when(dsm.getVersionManager()).thenReturn(versionManager); - when(versionManager.getApparentVersion()) - .thenReturn(HDDSVersion.SOFTWARE_VERSION); } /** @@ -116,9 +106,13 @@ public void testMetrics() { ozoneContainer, stateContext, connectionManager); commandHandler.handle(ReplicateContainerCommand.fromSources(3, sourceList), ozoneContainer, stateContext, connectionManager); - commandHandler.handle(ReplicateContainerCommand.toTarget(4, target), + commandHandler.handle( + ReplicateContainerCommand.toTarget(4, target, + HDDSVersion.SOFTWARE_VERSION), ozoneContainer, stateContext, connectionManager); - commandHandler.handle(ReplicateContainerCommand.toTarget(5, target), + commandHandler.handle( + ReplicateContainerCommand.toTarget(5, target, + HDDSVersion.SOFTWARE_VERSION), ozoneContainer, stateContext, connectionManager); commandHandler.handle(ReplicateContainerCommand.fromSources(6, sourceList), ozoneContainer, stateContext, connectionManager); @@ -140,75 +134,27 @@ public void testMetrics() { } } + /** + * The datanode follows the apparent version decided by SCM and carried in the + * command, without recomputing it. The submitted task should expose exactly + * the command's apparent version. + */ @Test - public void testEffectiveVersionPeerIsOlder() { - ArgumentCaptor captor = captureSubmittedTask(); - when(versionManager.getApparentVersion()) - .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); - - ReplicateContainerCommandHandler handler = - new ReplicateContainerCommandHandler(conf, supervisor, - downloadReplicator, pushReplicator); - - DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - List sources = new ArrayList<>(); - sources.add(source); - - ReplicateContainerCommand cmd = - ReplicateContainerCommand.fromSources(1, sources); - cmd.setPeerApparentVersion( - HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE.serialize()); - - handler.handle(cmd, ozoneContainer, stateContext, connectionManager); - - ReplicationTask task = captor.getValue(); - assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE.serialize(), - task.getLowestCommonApparentVersion()); - } - - @Test - public void testEffectiveVersionLocalIsOlder() { + public void testApparentVersionTakenFromCommand() { ArgumentCaptor captor = captureSubmittedTask(); - when(versionManager.getApparentVersion()) - .thenReturn(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); ReplicateContainerCommandHandler handler = new ReplicateContainerCommandHandler(conf, supervisor, downloadReplicator, pushReplicator); DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); - ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(1, target); - cmd.setPeerApparentVersion(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize()); - - handler.handle(cmd, ozoneContainer, stateContext, connectionManager); - - ReplicationTask task = captor.getValue(); - assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE.serialize(), - task.getLowestCommonApparentVersion()); - } - - @Test - public void testEffectiveVersionDefaultWhenNotSet() { - ArgumentCaptor captor = captureSubmittedTask(); - when(versionManager.getApparentVersion()) - .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); - - ReplicateContainerCommandHandler handler = - new ReplicateContainerCommandHandler(conf, supervisor, - downloadReplicator, pushReplicator); - - DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - List sources = new ArrayList<>(); - sources.add(source); - - ReplicateContainerCommand cmd = - ReplicateContainerCommand.fromSources(1, sources); + ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget( + 1, target, HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); handler.handle(cmd, ozoneContainer, stateContext, connectionManager); ReplicationTask task = captor.getValue(); - assertEquals(HDDSVersion.DEFAULT_VERSION.serialize(), - task.getLowestCommonApparentVersion()); + assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE, + task.getApparentVersion()); } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java index 8b831fa06466..f148db791595 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java @@ -40,6 +40,7 @@ import java.util.Collections; import java.util.UUID; import java.util.concurrent.atomic.AtomicLong; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; @@ -198,7 +199,7 @@ public void testUpload() { PushReplicator pushReplicator = new PushReplicator(conf, source, uploader); ReplicationTask task = - new ReplicationTask(toTarget(CONTAINER_ID, datanode), pushReplicator); + new ReplicationTask(toTarget(CONTAINER_ID, datanode, HDDSVersion.SOFTWARE_VERSION), pushReplicator); pushReplicator.replicate(task); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestPushReplicator.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestPushReplicator.java index a4463410cea5..94f3a23ccdbb 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestPushReplicator.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestPushReplicator.java @@ -32,6 +32,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ThreadLocalRandom; import java.util.function.Consumer; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; @@ -68,7 +69,7 @@ void uploadCompletesNormally(CopyContainerCompression compression) SpyOutputStream output = new SpyOutputStream(NULL_OUTPUT_STREAM); ContainerReplicator subject = createSubject(containerID, target, output, completion, compression); - ReplicationTask task = new ReplicationTask(toTarget(containerID, target), + ReplicationTask task = new ReplicationTask(toTarget(containerID, target, HDDSVersion.SOFTWARE_VERSION), subject); // WHEN @@ -89,7 +90,7 @@ void uploadFailsWithException() throws IOException { fut -> fut.completeExceptionally(new Exception("testing")); ContainerReplicator subject = createSubject(containerID, target, output, completion, NO_COMPRESSION); - ReplicationTask task = new ReplicationTask(toTarget(containerID, target), + ReplicationTask task = new ReplicationTask(toTarget(containerID, target, HDDSVersion.SOFTWARE_VERSION), subject); // WHEN @@ -111,7 +112,7 @@ void packFailsWithException() throws IOException { }; ContainerReplicator subject = createSubject(containerID, target, output, completion, NO_COMPRESSION); - ReplicationTask task = new ReplicationTask(toTarget(containerID, target), + ReplicationTask task = new ReplicationTask(toTarget(containerID, target, HDDSVersion.SOFTWARE_VERSION), subject); // WHEN diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java index 9ceb0a99e9f0..0fb29efdae5c 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java @@ -73,6 +73,7 @@ import org.apache.commons.compress.archivers.tar.TarArchiveEntry; import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream; import org.apache.commons.io.IOUtils; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.conf.ConfigurationSource; import org.apache.hadoop.hdds.conf.OzoneConfiguration; @@ -528,7 +529,8 @@ public void testDatanodeOutOfService(ContainerLayoutVersion layout) { HddsProtos.NodeOperationalState.DECOMMISSIONING); ReplicateContainerCommand pushCmd = ReplicateContainerCommand.toTarget( - 1, MockDatanodeDetails.randomDatanodeDetails()); + 1, MockDatanodeDetails.randomDatanodeDetails(), + HDDSVersion.SOFTWARE_VERSION); pushCmd.setTerm(CURRENT_TERM); ReplicateContainerCommand pullCmd = createCommand(2); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java index ccc430e46595..b36e6b075bf0 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java @@ -20,8 +20,6 @@ import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; -import java.util.Collections; -import java.util.List; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; @@ -34,29 +32,29 @@ public class TestReplicateContainerCommand { @Test - public void testPeerApparentVersionRoundTrip() { - DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - List sources = Collections.singletonList(source); + public void testApparentVersionRoundTrip() { + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); ReplicateContainerCommand cmd = - ReplicateContainerCommand.fromSources(1L, sources); - cmd.setPeerApparentVersion( - HDDSVersion.STREAM_BLOCK_SUPPORT.serialize()); + ReplicateContainerCommand.toTarget(1L, target, + HDDSVersion.STREAM_BLOCK_SUPPORT); ReplicateContainerCommandProto proto = cmd.getProto(); ReplicateContainerCommand deserialized = ReplicateContainerCommand.getFromProtobuf(proto); - assertEquals(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(), - deserialized.getPeerApparentVersion()); + assertEquals(HDDSVersion.STREAM_BLOCK_SUPPORT, + deserialized.getApparentVersion()); + assertEquals(target.getUuid(), + deserialized.getTargetDatanode().getUuid()); } @Test - public void testPeerApparentVersionDefaultWhenAbsent() { + public void testApparentVersionDefaultWhenAbsent() { DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(1L, target); + ReplicateContainerCommand.toTarget(1L, target, HDDSVersion.ZDU); ReplicateContainerCommandProto proto = ReplicateContainerCommandProto.newBuilder() @@ -68,37 +66,16 @@ public void testPeerApparentVersionDefaultWhenAbsent() { ReplicateContainerCommand deserialized = ReplicateContainerCommand.getFromProtobuf(proto); - assertEquals(HDDSVersion.DEFAULT_VERSION.serialize(), - deserialized.getPeerApparentVersion()); - } - - @Test - public void testPeerApparentVersionInPushMode() { - DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); - - ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(1L, target); - cmd.setPeerApparentVersion(HDDSVersion.ZDU.serialize()); - - ReplicateContainerCommandProto proto = cmd.getProto(); - ReplicateContainerCommand deserialized = - ReplicateContainerCommand.getFromProtobuf(proto); - - assertEquals(HDDSVersion.ZDU.serialize(), - deserialized.getPeerApparentVersion()); - assertEquals(target.getUuid(), - deserialized.getTargetDatanode().getUuid()); + assertEquals(HDDSVersion.DEFAULT_VERSION, + deserialized.getApparentVersion()); } @Test - public void testToStringIncludesPeerApparentVersion() { + public void testToStringIncludesApparentVersion() { ReplicateContainerCommand cmd = ReplicateContainerCommand.forTest(1L); - cmd.setPeerApparentVersion( - HDDSVersion.STREAM_BLOCK_SUPPORT.serialize()); String str = cmd.toString(); - assertTrue(str.contains("peerApparentVersion=" - + HDDSVersion.STREAM_BLOCK_SUPPORT.serialize())); + assertTrue(str.contains("apparentVersion=" + HDDSVersion.SOFTWARE_VERSION)); } } diff --git a/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto b/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto index 8326f5a35447..c6ed8fdfe20b 100644 --- a/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto +++ b/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto @@ -429,7 +429,7 @@ message ReplicateContainerCommandProto { optional int32 replicaIndex = 4; optional DatanodeDetailsProto target = 5; optional ReplicationCommandPriority priority = 6 [default = NORMAL]; - optional uint32 peerApparentVersion = 7; + optional uint32 apparentVersion = 7; } /** diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java index 42421a15e6bd..73a8c3dd49dc 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java @@ -531,10 +531,14 @@ public void sendThrottledReplicationCommand(ContainerInfo containerInfo, DatanodeDetails source = selectAndOptionallyExcludeDatanode( 1, sourceWithCmds); - ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(containerID, target); - cmd.setReplicaIndex(replicaIndex); - sendDatanodeCommand(cmd, containerInfo, source); + try { + ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget( + containerID, target, lowestApparentVersion(source, target)); + cmd.setReplicaIndex(replicaIndex); + sendDatanodeCommand(cmd, containerInfo, source); + } catch (NodeNotFoundException e) { + throw new IllegalArgumentException("Datanode not found in NodeManager. Should not happen"); + } } public void sendThrottledReconstructionCommand(ContainerInfo containerInfo, @@ -631,11 +635,16 @@ public void sendLowPriorityReplicateContainerCommand( final ContainerInfo container, int replicaIndex, DatanodeDetails source, DatanodeDetails target, long scmDeadlineEpochMs) throws NotLeaderException { - final ReplicateContainerCommand command = ReplicateContainerCommand - .toTarget(container.getContainerID(), target); - command.setReplicaIndex(replicaIndex); - command.setPriority(ReplicationCommandPriority.LOW); - sendDatanodeCommand(command, container, source, scmDeadlineEpochMs); + try { + final ReplicateContainerCommand command = ReplicateContainerCommand.toTarget( + container.getContainerID(), target, + lowestApparentVersion(source, target)); + command.setReplicaIndex(replicaIndex); + command.setPriority(ReplicationCommandPriority.LOW); + sendDatanodeCommand(command, container, source, scmDeadlineEpochMs); + } catch (NodeNotFoundException e) { + throw new IllegalArgumentException("Datanode not found in NodeManager. Should not happen"); + } } /** @@ -667,10 +676,6 @@ public void sendDatanodeCommand(SCMCommand command, throws NotLeaderException { long datanodeDeadline = scmDeadlineEpochMs - rmConf.getDatanodeTimeoutOffset(); - if (command.getType() == Type.replicateContainerCommand) { - enrichReplicateCommandWithPeerVersion( - (ReplicateContainerCommand) command); - } LOG.info("Sending command [{}] for container {} to {} with datanode " + "deadline {} and scm deadline {}", command, containerInfo, target, datanodeDeadline, @@ -682,34 +687,53 @@ public void sendDatanodeCommand(SCMCommand command, scmDeadlineEpochMs); } - private void enrichReplicateCommandWithPeerVersion( - ReplicateContainerCommand cmd) { - if (cmd.getTargetDatanode() != null) { - cmd.setPeerApparentVersion( - lookupApparentVersion(cmd.getTargetDatanode())); - } else { - int min = cmd.getSourceDatanodes().stream() - .mapToInt(this::lookupApparentVersion) - .min() - .orElse(HDDSVersion.DEFAULT_VERSION.serialize()); - cmd.setPeerApparentVersion(min); + /** + * Computes the apparent version a replication should use as the lowest + * apparent version among the given nodes. SCM acts as the coordinator: it + * decides the version so the datanodes carrying out the replication can simply + * follow it. SCM's view may be slightly stale, but this is safe because an + * apparent version never goes backwards and older versions remain supported. + * + *

If SCM has no information for one of the nodes we must not send a + * replication command involving it, so this fails fast with an unchecked + * exception (the node lookup should always succeed on the replication path). + */ + private HDDSVersion lowestApparentVersion(DatanodeDetails... nodes) + throws NodeNotFoundException { + HDDSVersion lowest = HDDSVersion.SOFTWARE_VERSION; + for (DatanodeDetails dn : nodes) { + HDDSVersion version = lookupApparentVersion(dn); + if (version.serialize() < lowest.serialize()) { + lowest = version; + } } + return lowest; } - private int lookupApparentVersion(DatanodeDetails dn) { + private HDDSVersion lookupApparentVersion(DatanodeDetails dn) + throws NodeNotFoundException { DatanodeInfo info = nodeManager.getDatanodeInfo(dn); if (info == null) { - return HDDSVersion.DEFAULT_VERSION.serialize(); - } - ComponentVersion peerVersion = info.getLastKnownApparentVersion(); - // A peer reporting a version newer than this SCM can recognize - // deserializes to UNKNOWN_VERSION (-1). Treat it as the oldest known - // version so the client datanode falls back to the most conservative, - // backward-compatible behavior rather than propagating the sentinel. - if (peerVersion == null || peerVersion == HDDSVersion.UNKNOWN_VERSION) { - return HDDSVersion.DEFAULT_VERSION.serialize(); - } - return peerVersion.serialize(); + throw new NodeNotFoundException(dn.getID()); + } + ComponentVersion apparentVersion = info.getLastKnownApparentVersion(); + if (apparentVersion == null) { + // Datanodes are expected to report their version on every heartbeat. + // Warn rather than silently defaulting forever if reporting is broken. + LOG.warn("Datanode {} has no reported apparent version; falling back " + + "to {}.", dn, HDDSVersion.DEFAULT_VERSION); + return HDDSVersion.DEFAULT_VERSION; + } + if (apparentVersion == HDDSVersion.UNKNOWN_VERSION) { + // The datanode reported a version newer than this SCM recognizes. SCM + // must be upgraded before datanodes, so a newer datanode is illegal and + // should have been fenced out of the cluster. Fail fast to stop any + // further operations on it. + throw new IllegalStateException("Datanode " + dn + " reported an " + + "apparent version newer than SCM recognizes. SCM must be upgraded " + + "before datanodes."); + } + return HDDSVersion.deserialize(apparentVersion.serialize()); } private void adjustPendingOpsAndMetrics(ContainerInfo containerInfo, diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java index 63f4c6c9f338..40553006c02e 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java @@ -34,6 +34,7 @@ import java.util.Set; import java.util.concurrent.atomic.AtomicBoolean; import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ReplicationConfig; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.conf.StorageUnit; @@ -459,7 +460,7 @@ public static void mockRMSendThrottleReplicateCommand(ReplicationManager mock, ContainerInfo containerInfo = invocationOnMock.getArgument(0); ReplicateContainerCommand command = ReplicateContainerCommand .toTarget(containerInfo.getContainerID(), - invocationOnMock.getArgument(2)); + invocationOnMock.getArgument(2), HDDSVersion.SOFTWARE_VERSION); command.setReplicaIndex(invocationOnMock.getArgument(3)); commandsSent.add(Pair.of(sources.get(0), command)); return null; diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestContainerReplicaPendingOps.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestContainerReplicaPendingOps.java index ee813f0942cc..50bb098127df 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestContainerReplicaPendingOps.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestContainerReplicaPendingOps.java @@ -36,6 +36,7 @@ import java.util.List; import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ReplicationType; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.DatanodeDetails; @@ -84,7 +85,8 @@ public void setup() { dn2 = MockDatanodeDetails.randomDatanodeDetails(); dn3 = MockDatanodeDetails.randomDatanodeDetails(); - addCmd = ReplicateContainerCommand.toTarget(1, dn3); + addCmd = ReplicateContainerCommand.toTarget(1, dn3, + HDDSVersion.SOFTWARE_VERSION); deleteCmd = new DeleteContainerCommand(1, false); } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java index 92e4360287d1..ebf670364413 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java @@ -65,6 +65,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.ComponentVersion; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.RatisReplicationConfig; @@ -162,6 +163,15 @@ public void setup() throws IOException { return null; }).when(nodeManager).addDatanodeCommand(any(), any()); + // By default every datanode is known and reports the current version, so + // the replication path can resolve an apparent version. Individual tests + // override this for specific nodes. + DatanodeInfo defaultNodeInfo = mock(DatanodeInfo.class); + when(defaultNodeInfo.getLastKnownApparentVersion()) + .thenReturn(HDDSVersion.SOFTWARE_VERSION); + when(nodeManager.getDatanodeInfo(any(DatanodeDetails.class))) + .thenReturn(defaultNodeInfo); + clock = new TestClock(Instant.now(), ZoneId.systemDefault()); containerReplicaPendingOps = new ContainerReplicaPendingOps(clock, null); @@ -1272,7 +1282,7 @@ public void testReplicateContainerCommandToTarget() DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); ReplicateContainerCommand command = ReplicateContainerCommand.toTarget( - containerInfo.getContainerID(), target); + containerInfo.getContainerID(), target, HDDSVersion.SOFTWARE_VERSION); command.setReplicaIndex(1); replicationManager.sendDatanodeCommand(command, containerInfo, source); @@ -1803,8 +1813,27 @@ private void mockReplicationCommandCounts( }); } + /** + * Captures the ReplicateContainerCommand SCM sends to a datanode. + */ + private ReplicateContainerCommand captureSentReplicateCommand() { + ArgumentCaptor> command = + ArgumentCaptor.forClass(SCMCommand.class); + verify(nodeManager).addDatanodeCommand(any(DatanodeID.class), + command.capture()); + return (ReplicateContainerCommand) command.getValue(); + } + + private DatanodeInfo mockDatanodeWithApparentVersion( + DatanodeDetails dn, ComponentVersion version) { + DatanodeInfo info = mock(DatanodeInfo.class); + when(info.getLastKnownApparentVersion()).thenReturn(version); + when(nodeManager.getDatanodeInfo(dn)).thenReturn(info); + return info; + } + @Test - public void testPeerApparentVersionSetForPushCommand() + public void testApparentVersionIsLowestOfSourceAndTarget() throws NotLeaderException { ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(repConfig, 1, @@ -1812,73 +1841,59 @@ public void testPeerApparentVersionSetForPushCommand() DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - DatanodeInfo targetInfo = mock(DatanodeInfo.class); - when(targetInfo.getLastKnownApparentVersion()) - .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); - when(nodeManager.getDatanodeInfo(target)).thenReturn(targetInfo); + // Source is newer than target, so the lowest common version is the + // target's version, not the peer the command is sent to. + mockDatanodeWithApparentVersion(source, HDDSVersion.STREAM_BLOCK_SUPPORT); + mockDatanodeWithApparentVersion(target, + HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); - ReplicateContainerCommand command = - ReplicateContainerCommand.toTarget( - containerInfo.getContainerID(), target); - command.setReplicaIndex(1); - replicationManager.sendDatanodeCommand(command, containerInfo, source); + replicationManager.sendLowPriorityReplicateContainerCommand(containerInfo, + 1, source, target, clock.millis() + rmConf.getEventTimeout()); - assertEquals(HDDSVersion.STREAM_BLOCK_SUPPORT.serialize(), - command.getPeerApparentVersion()); + assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE, + captureSentReplicateCommand().getApparentVersion()); } @Test - public void testPeerApparentVersionSetForPullCommand() - throws NotLeaderException { + public void testApparentVersionLookupThrowsWhenNodeNotFound() { ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); - DatanodeDetails source1 = MockDatanodeDetails.randomDatanodeDetails(); - DatanodeDetails source2 = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - DatanodeInfo source1Info = mock(DatanodeInfo.class); - when(source1Info.getLastKnownApparentVersion()) - .thenReturn(HDDSVersion.STREAM_BLOCK_SUPPORT); - DatanodeInfo source2Info = mock(DatanodeInfo.class); - when(source2Info.getLastKnownApparentVersion()) - .thenReturn(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); - - when(nodeManager.getDatanodeInfo(source1)).thenReturn(source1Info); - when(nodeManager.getDatanodeInfo(source2)).thenReturn(source2Info); - - ReplicateContainerCommand command = - ReplicateContainerCommand.fromSources( - containerInfo.getContainerID(), - Lists.newArrayList(source1, source2)); - replicationManager.sendDatanodeCommand(command, containerInfo, target); + mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); + // SCM has no information for the target. + when(nodeManager.getDatanodeInfo(target)).thenReturn(null); - assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE.serialize(), - command.getPeerApparentVersion()); + // We must not proceed with a replication command for a node we don't know. + assertThrows(IllegalArgumentException.class, () -> + replicationManager.sendLowPriorityReplicateContainerCommand( + containerInfo, 0, source, target, + clock.millis() + rmConf.getEventTimeout())); } @Test - public void testPeerApparentVersionDefaultWhenNodeNotFound() - throws NotLeaderException { + public void testApparentVersionLookupThrowsWhenPeerVersionUnknown() { ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - when(nodeManager.getDatanodeInfo(target)).thenReturn(null); - - ReplicateContainerCommand command = - ReplicateContainerCommand.toTarget( - containerInfo.getContainerID(), target); - replicationManager.sendDatanodeCommand(command, containerInfo, source); + mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); + // The target reports a version newer than SCM recognizes. SCM must be + // upgraded before datanodes, so this is illegal and should fail fast. + mockDatanodeWithApparentVersion(target, HDDSVersion.UNKNOWN_VERSION); - assertEquals(HDDSVersion.DEFAULT_VERSION.serialize(), - command.getPeerApparentVersion()); + assertThrows(IllegalStateException.class, () -> + replicationManager.sendLowPriorityReplicateContainerCommand( + containerInfo, 0, source, target, + clock.millis() + rmConf.getEventTimeout())); } @Test - public void testPeerApparentVersionDefaultWhenPeerVersionUnknown() + public void testApparentVersionDefaultWhenPeerVersionNotReported() throws NotLeaderException { ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(repConfig, 1, @@ -1886,21 +1901,15 @@ public void testPeerApparentVersionDefaultWhenPeerVersionUnknown() DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - // A peer reporting a version newer than this SCM can recognize - // deserializes to UNKNOWN_VERSION; it must fall back to the oldest - // version rather than propagating the sentinel. - DatanodeInfo targetInfo = mock(DatanodeInfo.class); - when(targetInfo.getLastKnownApparentVersion()) - .thenReturn(HDDSVersion.UNKNOWN_VERSION); - when(nodeManager.getDatanodeInfo(target)).thenReturn(targetInfo); - - ReplicateContainerCommand command = - ReplicateContainerCommand.toTarget( - containerInfo.getContainerID(), target); - replicationManager.sendDatanodeCommand(command, containerInfo, source); + mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); + // The target has not reported any apparent version yet. + mockDatanodeWithApparentVersion(target, null); + + replicationManager.sendLowPriorityReplicateContainerCommand(containerInfo, + 0, source, target, clock.millis() + rmConf.getEventTimeout()); - assertEquals(HDDSVersion.DEFAULT_VERSION.serialize(), - command.getPeerApparentVersion()); + assertEquals(HDDSVersion.DEFAULT_VERSION, + captureSentReplicateCommand().getApparentVersion()); } } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerScenarios.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerScenarios.java index f5b06f901092..dc5759b560a1 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerScenarios.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerScenarios.java @@ -45,6 +45,7 @@ import java.util.Set; import java.util.stream.Stream; import org.apache.commons.lang3.tuple.Pair; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.RatisReplicationConfig; import org.apache.hadoop.hdds.client.ReplicationConfig; @@ -63,6 +64,7 @@ import org.apache.hadoop.hdds.scm.container.ContainerReplica; import org.apache.hadoop.hdds.scm.container.ReplicationManagerReport; import org.apache.hadoop.hdds.scm.ha.SCMContext; +import org.apache.hadoop.hdds.scm.node.DatanodeInfo; import org.apache.hadoop.hdds.scm.node.NodeManager; import org.apache.hadoop.hdds.scm.node.NodeStatus; import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException; @@ -201,6 +203,14 @@ public void setup() throws IOException, NodeNotFoundException { return NODE_STATUS_MAP.getOrDefault(dn, NodeStatus.inServiceHealthy()); }); + // Replication commands look up each involved datanode's apparent version, + // so return an info reporting the current version for every node. + DatanodeInfo defaultNodeInfo = mock(DatanodeInfo.class); + when(defaultNodeInfo.getLastKnownApparentVersion()) + .thenReturn(HDDSVersion.SOFTWARE_VERSION); + when(nodeManager.getDatanodeInfo(any(DatanodeDetails.class))) + .thenReturn(defaultNodeInfo); + final HashMap countMap = new HashMap<>(); for (SCMCommandProto.Type type : SCMCommandProto.Type.values()) { countMap.put(type, 0); diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestCommandQueue.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestCommandQueue.java index d0cfa0be699a..3d076a4f9db4 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestCommandQueue.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestCommandQueue.java @@ -22,6 +22,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeID; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; @@ -50,10 +51,12 @@ public void testSummaryUpdated() { HddsProtos.ReplicationType.RATIS, HddsProtos.ReplicationFactor.THREE, Collections.emptyList()); SCMCommand replicationCommand = ReplicateContainerCommand.toTarget( - containerID, MockDatanodeDetails.randomDatanodeDetails()); + containerID, MockDatanodeDetails.randomDatanodeDetails(), + HDDSVersion.SOFTWARE_VERSION); ReplicateContainerCommand lowReplicationCommand = ReplicateContainerCommand - .toTarget(containerID, MockDatanodeDetails.randomDatanodeDetails()); + .toTarget(containerID, MockDatanodeDetails.randomDatanodeDetails(), + HDDSVersion.SOFTWARE_VERSION); lowReplicationCommand.setPriority(StorageContainerDatanodeProtocolProtos .ReplicationCommandPriority.LOW); diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java index eb5ffda37ac2..8a4ed9657d1e 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java @@ -977,7 +977,8 @@ scmStorageConfig, eventPublisher, new NetworkTopologyImpl(conf), times(1)).fireEvent(NEW_NODE, node1); for (int i = 0; i < 3; i++) { nodeManager.addDatanodeCommand(node1.getID(), ReplicateContainerCommand - .toTarget(1, MockDatanodeDetails.randomDatanodeDetails())); + .toTarget(1, MockDatanodeDetails.randomDatanodeDetails(), + HDDSVersion.SOFTWARE_VERSION)); } for (int i = 0; i < 5; i++) { nodeManager.addDatanodeCommand(node1.getID(), diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java index 968e331103e3..339f31192fe4 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java @@ -36,6 +36,7 @@ import java.util.function.ToLongFunction; import java.util.stream.IntStream; import java.util.stream.Stream; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.BlockID; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.conf.StorageUnit; @@ -111,7 +112,8 @@ void testPush(CopyContainerCompression compression) throws Exception { long containerID = createNewClosedContainer(source); DatanodeDetails target = selectOtherNode(source); ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(containerID, target); + ReplicateContainerCommand.toTarget(containerID, target, + HDDSVersion.SOFTWARE_VERSION); queueAndWaitForCompletion(cmd, source, ReplicationSupervisor::getReplicationSuccessCount); @@ -165,7 +167,7 @@ void pushUnknownContainer() throws Exception { DatanodeDetails target = selectOtherNode(source); ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget(CONTAINER_ID.incrementAndGet(), - target); + target, HDDSVersion.SOFTWARE_VERSION); queueAndWaitForCompletion(cmd, source, ReplicationSupervisor::getReplicationFailureCount); @@ -211,7 +213,8 @@ void testPushWithOverAllocatedContainer(String testName, Long containerSize) // Create replication command to push container to target ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(containerID, target); + ReplicateContainerCommand.toTarget(containerID, target, + HDDSVersion.SOFTWARE_VERSION); // Execute push replication queueAndWaitForCompletion(cmd, source, From 8ab318e0d05ac5ae610570ca906c0b47c0b16f5b Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Tue, 30 Jun 2026 15:20:02 +0200 Subject: [PATCH 5/9] Chain expection --- .../hdds/scm/container/replication/ReplicationManager.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java index 73a8c3dd49dc..47a78ce54c5d 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java @@ -537,7 +537,7 @@ public void sendThrottledReplicationCommand(ContainerInfo containerInfo, cmd.setReplicaIndex(replicaIndex); sendDatanodeCommand(cmd, containerInfo, source); } catch (NodeNotFoundException e) { - throw new IllegalArgumentException("Datanode not found in NodeManager. Should not happen"); + throw new IllegalArgumentException("Datanode not found in NodeManager. Should not happen", e); } } From 0464569ed50a4dda2115ec84005197f8c862f8f8 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Mon, 6 Jul 2026 15:11:32 +0200 Subject: [PATCH 6/9] Add missing expection --- .../hdds/scm/container/replication/ReplicationManager.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java index 47a78ce54c5d..7b03435498b5 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java @@ -643,7 +643,7 @@ public void sendLowPriorityReplicateContainerCommand( command.setPriority(ReplicationCommandPriority.LOW); sendDatanodeCommand(command, container, source, scmDeadlineEpochMs); } catch (NodeNotFoundException e) { - throw new IllegalArgumentException("Datanode not found in NodeManager. Should not happen"); + throw new IllegalArgumentException("Datanode not found in NodeManager. Should not happen", e); } } From c5d9ccb96b7fcf4a3e54072f1d268f69b4ca429f Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Tue, 7 Jul 2026 11:04:15 +0200 Subject: [PATCH 7/9] Move methods from ReplicationManager to a more common place --- .../replication/ReplicationManager.java | 57 +------------------ .../hadoop/hdds/scm/node/DatanodeInfo.java | 28 +++++++++ .../hadoop/hdds/scm/node/NodeManager.java | 26 +++++++++ .../replication/TestReplicationManager.java | 54 +++--------------- 4 files changed, 66 insertions(+), 99 deletions(-) diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java index 7b03435498b5..7cfb7397d60d 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java @@ -41,8 +41,6 @@ import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; import org.apache.commons.lang3.tuple.Pair; -import org.apache.hadoop.hdds.ComponentVersion; -import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.HddsConfigKeys; import org.apache.hadoop.hdds.client.ReplicationConfig; import org.apache.hadoop.hdds.conf.Config; @@ -80,7 +78,6 @@ import org.apache.hadoop.hdds.scm.events.SCMEvents; import org.apache.hadoop.hdds.scm.ha.SCMContext; import org.apache.hadoop.hdds.scm.ha.SCMService; -import org.apache.hadoop.hdds.scm.node.DatanodeInfo; import org.apache.hadoop.hdds.scm.node.NodeManager; import org.apache.hadoop.hdds.scm.node.NodeStatus; import org.apache.hadoop.hdds.scm.node.states.NodeNotFoundException; @@ -533,7 +530,8 @@ public void sendThrottledReplicationCommand(ContainerInfo containerInfo, try { ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget( - containerID, target, lowestApparentVersion(source, target)); + containerID, target, + nodeManager.getLowestApparentVersion(source, target)); cmd.setReplicaIndex(replicaIndex); sendDatanodeCommand(cmd, containerInfo, source); } catch (NodeNotFoundException e) { @@ -638,7 +636,7 @@ public void sendLowPriorityReplicateContainerCommand( try { final ReplicateContainerCommand command = ReplicateContainerCommand.toTarget( container.getContainerID(), target, - lowestApparentVersion(source, target)); + nodeManager.getLowestApparentVersion(source, target)); command.setReplicaIndex(replicaIndex); command.setPriority(ReplicationCommandPriority.LOW); sendDatanodeCommand(command, container, source, scmDeadlineEpochMs); @@ -687,55 +685,6 @@ public void sendDatanodeCommand(SCMCommand command, scmDeadlineEpochMs); } - /** - * Computes the apparent version a replication should use as the lowest - * apparent version among the given nodes. SCM acts as the coordinator: it - * decides the version so the datanodes carrying out the replication can simply - * follow it. SCM's view may be slightly stale, but this is safe because an - * apparent version never goes backwards and older versions remain supported. - * - *

If SCM has no information for one of the nodes we must not send a - * replication command involving it, so this fails fast with an unchecked - * exception (the node lookup should always succeed on the replication path). - */ - private HDDSVersion lowestApparentVersion(DatanodeDetails... nodes) - throws NodeNotFoundException { - HDDSVersion lowest = HDDSVersion.SOFTWARE_VERSION; - for (DatanodeDetails dn : nodes) { - HDDSVersion version = lookupApparentVersion(dn); - if (version.serialize() < lowest.serialize()) { - lowest = version; - } - } - return lowest; - } - - private HDDSVersion lookupApparentVersion(DatanodeDetails dn) - throws NodeNotFoundException { - DatanodeInfo info = nodeManager.getDatanodeInfo(dn); - if (info == null) { - throw new NodeNotFoundException(dn.getID()); - } - ComponentVersion apparentVersion = info.getLastKnownApparentVersion(); - if (apparentVersion == null) { - // Datanodes are expected to report their version on every heartbeat. - // Warn rather than silently defaulting forever if reporting is broken. - LOG.warn("Datanode {} has no reported apparent version; falling back " + - "to {}.", dn, HDDSVersion.DEFAULT_VERSION); - return HDDSVersion.DEFAULT_VERSION; - } - if (apparentVersion == HDDSVersion.UNKNOWN_VERSION) { - // The datanode reported a version newer than this SCM recognizes. SCM - // must be upgraded before datanodes, so a newer datanode is illegal and - // should have been fenced out of the cluster. Fail fast to stop any - // further operations on it. - throw new IllegalStateException("Datanode " + dn + " reported an " + - "apparent version newer than SCM recognizes. SCM must be upgraded " + - "before datanodes."); - } - return HDDSVersion.deserialize(apparentVersion.serialize()); - } - private void adjustPendingOpsAndMetrics(ContainerInfo containerInfo, SCMCommand cmd, DatanodeDetails targetDatanode, long scmDeadlineEpochMs) { diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java index 54e996ec3bf2..6912dba2dc05 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java @@ -25,6 +25,7 @@ import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import org.apache.hadoop.hdds.ComponentVersion; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto; @@ -159,6 +160,33 @@ public ComponentVersion getLastKnownApparentVersion() { } } + /** + * Returns the apparent version last reported by this datanode as + * an {@link HDDSVersion}. SCM uses this as the version datanodes should follow + * so that every node involved in an operation agrees on a mutually-supported + * version. + * + * @throws IllegalStateException if the datanode reported a version newer than + * this SCM recognizes. SCM must be upgraded before datanodes, so a newer + * datanode is illegal and should have been fenced out of the cluster. + */ + public HDDSVersion getApparentHddsVersion() { + ComponentVersion apparentVersion = getLastKnownApparentVersion(); + if (apparentVersion == null) { + // Datanodes are expected to report their version on every heartbeat. + // Warn rather than silently defaulting forever if reporting is broken. + LOG.warn("Datanode {} has no reported apparent version; falling back " + + "to {}.", this, HDDSVersion.DEFAULT_VERSION); + return HDDSVersion.DEFAULT_VERSION; + } + if (apparentVersion == HDDSVersion.UNKNOWN_VERSION) { + throw new IllegalStateException("Datanode " + this + " reported an " + + "apparent version newer than SCM recognizes. SCM must be upgraded " + + "before datanodes."); + } + return HDDSVersion.deserialize(apparentVersion.serialize()); + } + /** * Updates the datanode storage reports. * diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java index cd31eba5c4fd..2330a5317c7a 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java @@ -28,6 +28,7 @@ import java.util.Set; import java.util.function.BiConsumer; import org.apache.hadoop.hdds.ComponentVersion; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.DatanodeID; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState; @@ -197,6 +198,31 @@ default DatanodeFinalizationCounts getDatanodeFinalizationCounts() { return new DatanodeFinalizationCounts(finalizedNodes, totalHealthyNodes); } + /** + * Returns the lowest apparent (finalized) version among the given datanodes, + * so every node involved in an operation uses the same, mutually-supported + * version. + * + * @throws NodeNotFoundException if SCM has no record of one of the nodes; + * callers must not proceed with an operation involving a node SCM does + * not know about. + */ + default HDDSVersion getLowestApparentVersion(DatanodeDetails... nodes) + throws NodeNotFoundException { + HDDSVersion lowest = HDDSVersion.SOFTWARE_VERSION; + for (DatanodeDetails dn : nodes) { + DatanodeInfo info = getDatanodeInfo(dn); + if (info == null) { + throw new NodeNotFoundException(dn.getID()); + } + HDDSVersion version = info.getApparentHddsVersion(); + if (version.serialize() < lowest.serialize()) { + lowest = version; + } + } + return lowest; + } + /** * Returns the aggregated node stats. * @return the aggregated node stats. diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java index ebf670364413..a601b2a80aed 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java @@ -65,7 +65,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Function; import org.apache.commons.lang3.tuple.Pair; -import org.apache.hadoop.hdds.ComponentVersion; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.client.RatisReplicationConfig; @@ -1825,16 +1824,16 @@ private ReplicateContainerCommand captureSentReplicateCommand() { } private DatanodeInfo mockDatanodeWithApparentVersion( - DatanodeDetails dn, ComponentVersion version) { + DatanodeDetails dn, HDDSVersion version) { DatanodeInfo info = mock(DatanodeInfo.class); - when(info.getLastKnownApparentVersion()).thenReturn(version); + when(info.getApparentHddsVersion()).thenReturn(version); when(nodeManager.getDatanodeInfo(dn)).thenReturn(info); return info; } @Test public void testApparentVersionIsLowestOfSourceAndTarget() - throws NotLeaderException { + throws NotLeaderException, NodeNotFoundException { ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); @@ -1846,6 +1845,8 @@ public void testApparentVersionIsLowestOfSourceAndTarget() mockDatanodeWithApparentVersion(source, HDDSVersion.STREAM_BLOCK_SUPPORT); mockDatanodeWithApparentVersion(target, HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); + when(nodeManager.getLowestApparentVersion(source, target)) + .thenCallRealMethod(); replicationManager.sendLowPriorityReplicateContainerCommand(containerInfo, 1, source, target, clock.millis() + rmConf.getEventTimeout()); @@ -1855,7 +1856,8 @@ public void testApparentVersionIsLowestOfSourceAndTarget() } @Test - public void testApparentVersionLookupThrowsWhenNodeNotFound() { + public void testApparentVersionLookupThrowsWhenNodeNotFound() + throws NodeNotFoundException { ContainerInfo containerInfo = ReplicationTestUtil.createContainerInfo(repConfig, 1, HddsProtos.LifeCycleState.CLOSED, 10, 20); @@ -1865,6 +1867,8 @@ public void testApparentVersionLookupThrowsWhenNodeNotFound() { mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); // SCM has no information for the target. when(nodeManager.getDatanodeInfo(target)).thenReturn(null); + when(nodeManager.getLowestApparentVersion(source, target)) + .thenCallRealMethod(); // We must not proceed with a replication command for a node we don't know. assertThrows(IllegalArgumentException.class, () -> @@ -1872,44 +1876,4 @@ public void testApparentVersionLookupThrowsWhenNodeNotFound() { containerInfo, 0, source, target, clock.millis() + rmConf.getEventTimeout())); } - - @Test - public void testApparentVersionLookupThrowsWhenPeerVersionUnknown() { - ContainerInfo containerInfo = - ReplicationTestUtil.createContainerInfo(repConfig, 1, - HddsProtos.LifeCycleState.CLOSED, 10, 20); - DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); - DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - - mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); - // The target reports a version newer than SCM recognizes. SCM must be - // upgraded before datanodes, so this is illegal and should fail fast. - mockDatanodeWithApparentVersion(target, HDDSVersion.UNKNOWN_VERSION); - - assertThrows(IllegalStateException.class, () -> - replicationManager.sendLowPriorityReplicateContainerCommand( - containerInfo, 0, source, target, - clock.millis() + rmConf.getEventTimeout())); - } - - @Test - public void testApparentVersionDefaultWhenPeerVersionNotReported() - throws NotLeaderException { - ContainerInfo containerInfo = - ReplicationTestUtil.createContainerInfo(repConfig, 1, - HddsProtos.LifeCycleState.CLOSED, 10, 20); - DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); - DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); - - mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); - // The target has not reported any apparent version yet. - mockDatanodeWithApparentVersion(target, null); - - replicationManager.sendLowPriorityReplicateContainerCommand(containerInfo, - 0, source, target, clock.millis() + rmConf.getEventTimeout()); - - assertEquals(HDDSVersion.DEFAULT_VERSION, - captureSentReplicateCommand().getApparentVersion()); - } - } From a8820621694867fbfb9345ddb70c8385aee9f5f3 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Wed, 15 Jul 2026 16:23:56 +0200 Subject: [PATCH 8/9] Address review comments --- .../apache/hadoop/hdds/ComponentVersion.java | 22 +++++++++++++++ .../replication/ReplicationTask.java | 6 ++-- .../commands/ReplicateContainerCommand.java | 12 ++++---- .../TestReplicateContainerCommand.java | 26 ++++++++++++++--- .../hadoop/hdds/scm/node/DatanodeInfo.java | 28 ------------------- .../hadoop/hdds/scm/node/NodeManager.java | 20 ++++++------- .../replication/TestReplicationManager.java | 2 +- 7 files changed, 63 insertions(+), 53 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ComponentVersion.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ComponentVersion.java index 9c1223f35c30..77c048e31b4c 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ComponentVersion.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ComponentVersion.java @@ -80,4 +80,26 @@ default boolean isSupportedBy(ComponentVersion other) { default Optional action() { return Optional.empty(); } + + /** + * Returns the version with the lowest feature set among the given versions, + * so callers can pick a version that is mutually supported by all of them. + * Comparison is done through {@link #isSupportedBy}, which respects the + * negative/unknown-future-version convention, rather than comparing the + * opaque {@link #serialize()} values directly. + * + * @throws IllegalArgumentException if no versions are provided. + */ + static ComponentVersion min(ComponentVersion... versions) { + if (versions.length == 0) { + throw new IllegalArgumentException("At least one version is required."); + } + ComponentVersion lowest = versions[0]; + for (int i = 1; i < versions.length; i++) { + if (versions[i].isSupportedBy(lowest)) { + lowest = versions[i]; + } + } + return lowest; + } } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java index 1b0fc4f2b495..0a069cafea0a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java @@ -19,7 +19,7 @@ import java.util.List; import java.util.Objects; -import org.apache.hadoop.hdds.HDDSVersion; +import org.apache.hadoop.hdds.ComponentVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; @@ -31,7 +31,7 @@ public class ReplicationTask extends AbstractReplicationTask { private final ReplicateContainerCommand cmd; private final ContainerReplicator replicator; private final String debugString; - private final HDDSVersion apparentVersion; + private final ComponentVersion apparentVersion; public static final String METRIC_NAME = "ContainerReplications"; public static final String METRIC_DESCRIPTION_SEGMENT = "container replications"; @@ -133,7 +133,7 @@ public void setTransferredBytes(long transferredBytes) { * can use this to gate version-dependent protocol features so that the * newer side downgrades to a behavior the older side understands. */ - public HDDSVersion getApparentVersion() { + public ComponentVersion getApparentVersion() { return apparentVersion; } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java index 4cab11b14a78..493d6717e432 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/commands/ReplicateContainerCommand.java @@ -22,6 +22,7 @@ import java.util.List; import java.util.Objects; import java.util.stream.Collectors; +import org.apache.hadoop.hdds.ComponentVersion; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.DatanodeDetailsProto; @@ -30,6 +31,7 @@ import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicationCommandPriority; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type; +import org.apache.hadoop.hdds.upgrade.HDDSVersionUtils; /** * SCM command to request replication of a container. @@ -43,7 +45,7 @@ public final class ReplicateContainerCommand private int replicaIndex = 0; private ReplicationCommandPriority priority = ReplicationCommandPriority.NORMAL; - private HDDSVersion apparentVersion = HDDSVersion.DEFAULT_VERSION; + private ComponentVersion apparentVersion = HDDSVersion.DEFAULT_VERSION; public static ReplicateContainerCommand fromSources(long containerID, List sourceDatanodes) { @@ -51,7 +53,7 @@ public static ReplicateContainerCommand fromSources(long containerID, } public static ReplicateContainerCommand toTarget(long containerID, - DatanodeDetails target, HDDSVersion apparentVersion) { + DatanodeDetails target, ComponentVersion apparentVersion) { ReplicateContainerCommand cmd = new ReplicateContainerCommand(containerID, emptyList(), target); cmd.apparentVersion = apparentVersion; @@ -95,7 +97,7 @@ public void setPriority(ReplicationCommandPriority priority) { * replication. SCM computes this as the lowest apparent version among the * nodes involved. */ - public HDDSVersion getApparentVersion() { + public ComponentVersion getApparentVersion() { return apparentVersion; } @@ -150,8 +152,8 @@ public static ReplicateContainerCommand getFromProtobuf( cmd.setPriority(protoMessage.getPriority()); } if (protoMessage.hasApparentVersion()) { - cmd.apparentVersion = - HDDSVersion.deserialize(protoMessage.getApparentVersion()); + cmd.apparentVersion = HDDSVersionUtils.deserializeHDDSVersionOrLayoutVersion( + protoMessage.getApparentVersion()); } return cmd; } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java index b36e6b075bf0..6daa298193ad 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java @@ -23,6 +23,7 @@ import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; +import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicateContainerCommandProto; import org.junit.jupiter.api.Test; @@ -36,19 +37,36 @@ public void testApparentVersionRoundTrip() { DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(1L, target, - HDDSVersion.STREAM_BLOCK_SUPPORT); + ReplicateContainerCommand.toTarget(1L, target, HDDSVersion.ZDU); ReplicateContainerCommandProto proto = cmd.getProto(); ReplicateContainerCommand deserialized = ReplicateContainerCommand.getFromProtobuf(proto); - assertEquals(HDDSVersion.STREAM_BLOCK_SUPPORT, - deserialized.getApparentVersion()); + assertEquals(HDDSVersion.ZDU, deserialized.getApparentVersion()); assertEquals(target.getUuid(), deserialized.getTargetDatanode().getUuid()); } + @Test + public void testLayoutFeatureApparentVersionRoundTrip() { + // Pre-ZDU datanodes report a layout feature as their apparent version. + // The type must survive serialization rather than being erased to an + // HDDSVersion. + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + + ReplicateContainerCommand cmd = + ReplicateContainerCommand.toTarget(1L, target, + HDDSLayoutFeature.HBASE_SUPPORT); + + ReplicateContainerCommandProto proto = cmd.getProto(); + ReplicateContainerCommand deserialized = + ReplicateContainerCommand.getFromProtobuf(proto); + + assertEquals(HDDSLayoutFeature.HBASE_SUPPORT, + deserialized.getApparentVersion()); + } + @Test public void testApparentVersionDefaultWhenAbsent() { DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java index 6912dba2dc05..54e996ec3bf2 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/DatanodeInfo.java @@ -25,7 +25,6 @@ import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; import org.apache.hadoop.hdds.ComponentVersion; -import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandQueueReportProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto; @@ -160,33 +159,6 @@ public ComponentVersion getLastKnownApparentVersion() { } } - /** - * Returns the apparent version last reported by this datanode as - * an {@link HDDSVersion}. SCM uses this as the version datanodes should follow - * so that every node involved in an operation agrees on a mutually-supported - * version. - * - * @throws IllegalStateException if the datanode reported a version newer than - * this SCM recognizes. SCM must be upgraded before datanodes, so a newer - * datanode is illegal and should have been fenced out of the cluster. - */ - public HDDSVersion getApparentHddsVersion() { - ComponentVersion apparentVersion = getLastKnownApparentVersion(); - if (apparentVersion == null) { - // Datanodes are expected to report their version on every heartbeat. - // Warn rather than silently defaulting forever if reporting is broken. - LOG.warn("Datanode {} has no reported apparent version; falling back " + - "to {}.", this, HDDSVersion.DEFAULT_VERSION); - return HDDSVersion.DEFAULT_VERSION; - } - if (apparentVersion == HDDSVersion.UNKNOWN_VERSION) { - throw new IllegalStateException("Datanode " + this + " reported an " + - "apparent version newer than SCM recognizes. SCM must be upgraded " + - "before datanodes."); - } - return HDDSVersion.deserialize(apparentVersion.serialize()); - } - /** * Updates the datanode storage reports. * diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java index 2330a5317c7a..b0d3e322473f 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java @@ -28,7 +28,6 @@ import java.util.Set; import java.util.function.BiConsumer; import org.apache.hadoop.hdds.ComponentVersion; -import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.DatanodeID; import org.apache.hadoop.hdds.protocol.proto.HddsProtos.NodeOperationalState; @@ -199,7 +198,7 @@ default DatanodeFinalizationCounts getDatanodeFinalizationCounts() { } /** - * Returns the lowest apparent (finalized) version among the given datanodes, + * Returns the lowest apparent version among the given datanodes, * so every node involved in an operation uses the same, mutually-supported * version. * @@ -207,20 +206,17 @@ default DatanodeFinalizationCounts getDatanodeFinalizationCounts() { * callers must not proceed with an operation involving a node SCM does * not know about. */ - default HDDSVersion getLowestApparentVersion(DatanodeDetails... nodes) + default ComponentVersion getLowestApparentVersion(DatanodeDetails... nodes) throws NodeNotFoundException { - HDDSVersion lowest = HDDSVersion.SOFTWARE_VERSION; - for (DatanodeDetails dn : nodes) { - DatanodeInfo info = getDatanodeInfo(dn); + ComponentVersion[] versions = new ComponentVersion[nodes.length]; + for (int i = 0; i < nodes.length; i++) { + DatanodeInfo info = getDatanodeInfo(nodes[i]); if (info == null) { - throw new NodeNotFoundException(dn.getID()); - } - HDDSVersion version = info.getApparentHddsVersion(); - if (version.serialize() < lowest.serialize()) { - lowest = version; + throw new NodeNotFoundException(nodes[i].getID()); } + versions[i] = info.getLastKnownApparentVersion(); } - return lowest; + return ComponentVersion.min(versions); } /** diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java index a601b2a80aed..90eb83ced87a 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java @@ -1826,7 +1826,7 @@ private ReplicateContainerCommand captureSentReplicateCommand() { private DatanodeInfo mockDatanodeWithApparentVersion( DatanodeDetails dn, HDDSVersion version) { DatanodeInfo info = mock(DatanodeInfo.class); - when(info.getApparentHddsVersion()).thenReturn(version); + when(info.getLastKnownApparentVersion()).thenReturn(version); when(nodeManager.getDatanodeInfo(dn)).thenReturn(info); return info; } From ea4b08b5aa0cd1a3234ac1f73e94423238efc7e8 Mon Sep 17 00:00:00 2001 From: Zita Dombi Date: Wed, 15 Jul 2026 18:11:20 +0200 Subject: [PATCH 9/9] Address issues after HDDS-14496-zdu merge --- .../replication/ReplicationTask.java | 7 ------- .../common/statemachine/TestStateContext.java | 13 +++++++++---- .../TestReplicateContainerCommandHandler.java | 9 ++++----- ...licationSupervisorSchedulingBenchmark.java | 3 ++- .../TestGrpcReplicationService.java | 2 +- .../replication/TestMeasuredReplicator.java | 19 ++++++++++--------- .../TestReplicationSupervisor.java | 3 +-- .../TestReplicateContainerCommand.java | 9 +++++---- .../hadoop/hdds/scm/node/NodeManager.java | 2 +- .../replication/TestReplicationManager.java | 6 +++--- .../TestReplicationManagerScenarios.java | 2 +- .../scm/TestSCMDatanodeProtocolServer.java | 4 +++- .../replication/TestContainerReplication.java | 2 +- 13 files changed, 41 insertions(+), 40 deletions(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java index f4cf00730e62..e21b47827aab 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ReplicationTask.java @@ -46,13 +46,6 @@ public ReplicationTask(ReplicateContainerCommand cmd, this.cmd = cmd; this.replicator = replicator; this.apparentVersion = cmd.getApparentVersion(); - if (cmd.getTargetDatanode() != null) { - // Only push replication will have a target datanode set, and it must be - // sent to the source datanode to be executed. It is possible the source - // is out of service, so we need to set the flag to allow the command to - // run. - setShouldOnlyRunOnInServiceDatanodes(false); - } debugString = cmd.toString(); } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java index 42220c6b99fe..701d36858bd3 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/TestStateContext.java @@ -48,6 +48,7 @@ import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; @@ -703,10 +704,13 @@ public void testGetReports() { @Test public void testCommandQueueSummary() throws IOException { StateContext ctx = createSubject(); - ctx.addCommand(ReplicateContainerCommand.toTarget(1, MockDatanodeDetails.randomDatanodeDetails())); + ctx.addCommand(ReplicateContainerCommand + .toTarget(1, MockDatanodeDetails.randomDatanodeDetails(), HDDSVersion.SOFTWARE_VERSION)); ctx.addCommand(new ClosePipelineCommand(PipelineID.randomId())); - ctx.addCommand(ReplicateContainerCommand.toTarget(2, MockDatanodeDetails.randomDatanodeDetails())); - ctx.addCommand(ReplicateContainerCommand.toTarget(3, MockDatanodeDetails.randomDatanodeDetails())); + ctx.addCommand(ReplicateContainerCommand + .toTarget(2, MockDatanodeDetails.randomDatanodeDetails(), HDDSVersion.SOFTWARE_VERSION)); + ctx.addCommand(ReplicateContainerCommand + .toTarget(3, MockDatanodeDetails.randomDatanodeDetails(), HDDSVersion.SOFTWARE_VERSION)); ctx.addCommand(new ClosePipelineCommand(PipelineID.randomId())); ctx.addCommand(new CloseContainerCommand(1, PipelineID.randomId())); ctx.addCommand(new ReconcileContainerCommand(4, Collections.emptySet())); @@ -773,7 +777,8 @@ private static StateContext createSubject() throws IOException { } private static SCMCommand someCommand() { - return ReplicateContainerCommand.toTarget(1, MockDatanodeDetails.randomDatanodeDetails()); + return ReplicateContainerCommand + .toTarget(1, MockDatanodeDetails.randomDatanodeDetails(), HDDSVersion.SOFTWARE_VERSION); } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java index 548236c73a17..fe3999f935aa 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/TestReplicateContainerCommandHandler.java @@ -84,16 +84,16 @@ public void testMetrics() { DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); ReplicateContainerCommand command = - ReplicateContainerCommand.toTarget(1, target); + ReplicateContainerCommand.toTarget(1, target, HDDSVersion.SOFTWARE_VERSION); commandHandler.handle(command, ozoneContainer, stateContext, connectionManager); String metricsName = ReplicationTask.METRIC_NAME; assertEquals(commandHandler.getMetricsName(), metricsName); when(supervisor.getReplicationRequestCount(metricsName)).thenReturn(1L); assertEquals(commandHandler.getInvocationCount(), 1); - commandHandler.handle(ReplicateContainerCommand.toTarget(2, target), + commandHandler.handle(ReplicateContainerCommand.toTarget(2, target, HDDSVersion.SOFTWARE_VERSION), ozoneContainer, stateContext, connectionManager); - commandHandler.handle(ReplicateContainerCommand.toTarget(3, target), + commandHandler.handle(ReplicateContainerCommand.toTarget(3, target, HDDSVersion.SOFTWARE_VERSION), ozoneContainer, stateContext, connectionManager); commandHandler.handle( ReplicateContainerCommand.toTarget(4, target, @@ -131,8 +131,7 @@ public void testApparentVersionTakenFromCommand() { ArgumentCaptor captor = captureSubmittedTask(); ReplicateContainerCommandHandler handler = - new ReplicateContainerCommandHandler(conf, supervisor, - downloadReplicator, pushReplicator); + new ReplicateContainerCommandHandler(supervisor, pushReplicator); DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget( diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorSchedulingBenchmark.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorSchedulingBenchmark.java index e3d0c4126da1..c574b8661753 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorSchedulingBenchmark.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/ReplicationSupervisorSchedulingBenchmark.java @@ -23,6 +23,7 @@ import java.util.HashMap; import java.util.Map; import java.util.Random; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.DatanodeID; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; @@ -77,7 +78,7 @@ public void test() throws InterruptedException { //schedule 100 container replication for (int i = 0; i < 100; i++) { - rs.addTask(new ReplicationTask(toTarget(i, target), replicator)); + rs.addTask(new ReplicationTask(toTarget(i, target, HDDSVersion.SOFTWARE_VERSION), replicator)); } rs.shutdownAfterFinish(); final long executionTime = Time.monotonicNow() - start; diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java index 6b5a129d5cab..d107696b4374 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestGrpcReplicationService.java @@ -201,7 +201,7 @@ public void copyData(long containerId, OutputStream destination, PushReplicator pushReplicator = new PushReplicator(conf, source, uploader); ReplicationTask task = new ReplicationTask( - toTarget(CONTAINER_ID, datanode), pushReplicator); + toTarget(CONTAINER_ID, datanode, HDDSVersion.SOFTWARE_VERSION), pushReplicator); // WHEN pushReplicator.replicate(task); diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestMeasuredReplicator.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestMeasuredReplicator.java index f0423bcc0fc4..1743714b7691 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestMeasuredReplicator.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestMeasuredReplicator.java @@ -23,6 +23,7 @@ import java.time.Instant; import java.time.temporal.ChronoUnit; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; import org.apache.hadoop.ozone.container.replication.AbstractReplicationTask.Status; @@ -69,9 +70,9 @@ public void closeReplicator() throws Exception { @Test public void measureFailureSuccessAndBytes() { //WHEN - measuredReplicator.replicate(new ReplicationTask(toTarget(1, TARGET), replicator)); - measuredReplicator.replicate(new ReplicationTask(toTarget(2, TARGET), replicator)); - measuredReplicator.replicate(new ReplicationTask(toTarget(3, TARGET), replicator)); + measuredReplicator.replicate(new ReplicationTask(toTarget(1, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator)); + measuredReplicator.replicate(new ReplicationTask(toTarget(2, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator)); + measuredReplicator.replicate(new ReplicationTask(toTarget(3, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator)); //THEN //even containers should be failed @@ -89,9 +90,9 @@ public void measureFailureSuccessAndBytes() { public void testReplicationTime() throws Exception { //WHEN //will wait at least the 300ms - measuredReplicator.replicate(new ReplicationTask(toTarget(101, TARGET), replicator)); - measuredReplicator.replicate(new ReplicationTask(toTarget(201, TARGET), replicator)); - measuredReplicator.replicate(new ReplicationTask(toTarget(300, TARGET), replicator)); + measuredReplicator.replicate(new ReplicationTask(toTarget(101, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator)); + measuredReplicator.replicate(new ReplicationTask(toTarget(201, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator)); + measuredReplicator.replicate(new ReplicationTask(toTarget(300, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator)); //THEN //even containers should be failed @@ -109,7 +110,7 @@ public void testReplicationTime() throws Exception { public void testFailureTimeSuccessExcluded() { //WHEN //will wait at least the 15ms - measuredReplicator.replicate(new ReplicationTask(toTarget(15, TARGET), replicator)); + measuredReplicator.replicate(new ReplicationTask(toTarget(15, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator)); //THEN @@ -121,7 +122,7 @@ public void testFailureTimeSuccessExcluded() { public void testSuccessTimeFailureExcluded() { //WHEN //will wait at least the 10ms - measuredReplicator.replicate(new ReplicationTask(toTarget(10, TARGET), replicator)); + measuredReplicator.replicate(new ReplicationTask(toTarget(10, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator)); //THEN @@ -132,7 +133,7 @@ public void testSuccessTimeFailureExcluded() { @Test public void testReplicationQueueTimeMetrics() { final Instant queued = Instant.now().minus(1, ChronoUnit.SECONDS); - ReplicationTask task = new ReplicationTask(toTarget(100, TARGET), replicator) { + ReplicationTask task = new ReplicationTask(toTarget(100, TARGET, HDDSVersion.SOFTWARE_VERSION), replicator) { @Override public Instant getQueued() { return queued; diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java index f62d076305b0..7a631b457355 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/replication/TestReplicationSupervisor.java @@ -62,7 +62,6 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; -import org.apache.hadoop.hdds.ComponentVersion; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.client.ECReplicationConfig; import org.apache.hadoop.hdds.conf.ConfigurationSource; @@ -1051,7 +1050,7 @@ private void scheduleTasks( List datanodes, ReplicationSupervisor rs) { for (int i = 0; i < 10; i++) { DatanodeDetails target = datanodes.get(i % datanodes.size()); - rs.addTask(new ReplicationTask(toTarget(i, target), noopReplicator)); + rs.addTask(new ReplicationTask(toTarget(i, target, HDDSVersion.SOFTWARE_VERSION), noopReplicator)); } } } diff --git a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java index 6daa298193ad..ff47ccdb04c3 100644 --- a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java @@ -17,14 +17,15 @@ package org.apache.hadoop.ozone.protocol.commands; +import static org.apache.hadoop.hdds.protocol.MockDatanodeDetails.randomDatanodeDetails; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.hdds.protocol.MockDatanodeDetails; -import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicateContainerCommandProto; +import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature; import org.junit.jupiter.api.Test; /** @@ -44,8 +45,8 @@ public void testApparentVersionRoundTrip() { ReplicateContainerCommand.getFromProtobuf(proto); assertEquals(HDDSVersion.ZDU, deserialized.getApparentVersion()); - assertEquals(target.getUuid(), - deserialized.getTargetDatanode().getUuid()); + assertEquals(target.getID(), + deserialized.getTargetDatanode().getID()); } @Test @@ -91,7 +92,7 @@ public void testApparentVersionDefaultWhenAbsent() { @Test public void testToStringIncludesApparentVersion() { ReplicateContainerCommand cmd = - ReplicateContainerCommand.forTest(1L); + ReplicateContainerCommand.toTarget(1L, randomDatanodeDetails(), HDDSVersion.SOFTWARE_VERSION); String str = cmd.toString(); assertTrue(str.contains("apparentVersion=" + HDDSVersion.SOFTWARE_VERSION)); diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java index 26837a02f5c4..15217fa18e31 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/NodeManager.java @@ -203,7 +203,7 @@ default ComponentVersion getLowestApparentVersion(DatanodeDetails... nodes) throws NodeNotFoundException { ComponentVersion[] versions = new ComponentVersion[nodes.length]; for (int i = 0; i < nodes.length; i++) { - DatanodeInfo info = getDatanodeInfo(nodes[i]); + DatanodeInfo info = getNode(nodes[i].getID()); if (info == null) { throw new NodeNotFoundException(nodes[i].getID()); } diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java index 535930c646a1..218fd5090356 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManager.java @@ -168,7 +168,7 @@ public void setup() throws IOException { DatanodeInfo defaultNodeInfo = mock(DatanodeInfo.class); when(defaultNodeInfo.getLastKnownApparentVersion()) .thenReturn(HDDSVersion.SOFTWARE_VERSION); - when(nodeManager.getDatanodeInfo(any(DatanodeDetails.class))) + when(nodeManager.getNode(any(DatanodeID.class))) .thenReturn(defaultNodeInfo); clock = new MockClock(Instant.now(), ZoneId.systemDefault()); @@ -1765,7 +1765,7 @@ private DatanodeInfo mockDatanodeWithApparentVersion( DatanodeDetails dn, HDDSVersion version) { DatanodeInfo info = mock(DatanodeInfo.class); when(info.getLastKnownApparentVersion()).thenReturn(version); - when(nodeManager.getDatanodeInfo(dn)).thenReturn(info); + when(nodeManager.getNode(dn.getID())).thenReturn(info); return info; } @@ -1804,7 +1804,7 @@ public void testApparentVersionLookupThrowsWhenNodeNotFound() mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); // SCM has no information for the target. - when(nodeManager.getDatanodeInfo(target)).thenReturn(null); + when(nodeManager.getNode(target.getID())).thenReturn(null); when(nodeManager.getLowestApparentVersion(source, target)) .thenCallRealMethod(); diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerScenarios.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerScenarios.java index 492c3d642d5b..15cce574ee95 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerScenarios.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/TestReplicationManagerScenarios.java @@ -208,7 +208,7 @@ public void setup() throws IOException, NodeNotFoundException { DatanodeInfo defaultNodeInfo = mock(DatanodeInfo.class); when(defaultNodeInfo.getLastKnownApparentVersion()) .thenReturn(HDDSVersion.SOFTWARE_VERSION); - when(nodeManager.getDatanodeInfo(any(DatanodeDetails.class))) + when(nodeManager.getNode(any(DatanodeID.class))) .thenReturn(defaultNodeInfo); final HashMap countMap = new HashMap<>(); diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMDatanodeProtocolServer.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMDatanodeProtocolServer.java index 1c9a138e33fa..beb7b79ed331 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMDatanodeProtocolServer.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/hdds/scm/TestSCMDatanodeProtocolServer.java @@ -23,6 +23,7 @@ import java.io.IOException; import java.util.concurrent.TimeoutException; +import org.apache.hadoop.hdds.HDDSVersion; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos; import org.apache.hadoop.hdds.scm.server.OzoneStorageContainerManager; import org.apache.hadoop.hdds.scm.server.SCMDatanodeProtocolServer; @@ -40,7 +41,8 @@ public void ensureTermAndDeadlineOnCommands() OzoneStorageContainerManager scm = mock(OzoneStorageContainerManager.class); - ReplicateContainerCommand command = ReplicateContainerCommand.toTarget(1, randomDatanodeDetails()); + ReplicateContainerCommand command = ReplicateContainerCommand + .toTarget(1, randomDatanodeDetails(), HDDSVersion.SOFTWARE_VERSION); command.setTerm(5L); command.setDeadline(1234L); StorageContainerDatanodeProtocolProtos.SCMCommandProto proto = diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java index d583dbb0ff93..7742f477d826 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/container/replication/TestContainerReplication.java @@ -138,7 +138,7 @@ void pushSucceedsWhenSourceBytesUsedIsNegative(long containerSize) throws Except poisonBytesUsed(source, containerID, containerSize); ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(containerID, target); + ReplicateContainerCommand.toTarget(containerID, target, HDDSVersion.SOFTWARE_VERSION); queueAndWaitForCompletion(cmd, source, ReplicationSupervisor::getReplicationSuccessCount);