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 ce8f535e0c4a..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 @@ -18,6 +18,7 @@ package org.apache.hadoop.ozone.container.replication; import java.util.Objects; +import org.apache.hadoop.hdds.ComponentVersion; import org.apache.hadoop.hdds.protocol.DatanodeDetails; import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand; @@ -29,6 +30,7 @@ public class ReplicationTask extends AbstractReplicationTask { private final ReplicateContainerCommand cmd; private final ContainerReplicator replicator; private final String debugString; + private final ComponentVersion apparentVersion; public static final String METRIC_NAME = "ContainerReplications"; public static final String METRIC_DESCRIPTION_SEGMENT = "container replications"; @@ -43,6 +45,7 @@ public ReplicationTask(ReplicateContainerCommand cmd, setPriority(cmd.getPriority()); this.cmd = cmd; this.replicator = replicator; + this.apparentVersion = cmd.getApparentVersion(); debugString = cmd.toString(); } @@ -101,6 +104,15 @@ public void setTransferredBytes(long transferredBytes) { this.transferredBytes = transferredBytes; } + /** + * @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 ComponentVersion getApparentVersion() { + return apparentVersion; + } + 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 c35439096532..a91ffc8fd786 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 @@ -18,12 +18,15 @@ package org.apache.hadoop.ozone.protocol.commands; import java.util.Objects; +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.ReplicateContainerCommandProto; import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicateContainerCommandProto.Builder; 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 push-replication of a container to a target datanode. @@ -36,10 +39,14 @@ public final class ReplicateContainerCommand private int replicaIndex = 0; private ReplicationCommandPriority priority = ReplicationCommandPriority.NORMAL; + private ComponentVersion apparentVersion = HDDSVersion.DEFAULT_VERSION; public static ReplicateContainerCommand toTarget(long containerID, - DatanodeDetails target) { - return new ReplicateContainerCommand(containerID, target); + DatanodeDetails target, ComponentVersion apparentVersion) { + ReplicateContainerCommand cmd = + new ReplicateContainerCommand(containerID, target); + cmd.apparentVersion = apparentVersion; + return cmd; } private ReplicateContainerCommand(long containerID, DatanodeDetails target) { @@ -63,6 +70,15 @@ public void setPriority(ReplicationCommandPriority priority) { this.priority = priority; } + /** + * @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 ComponentVersion getApparentVersion() { + return apparentVersion; + } + @Override public Type getType() { return SCMCommandProto.Type.replicateContainerCommand; @@ -80,7 +96,8 @@ public ReplicateContainerCommandProto getProto() { .setContainerID(containerID) .setReplicaIndex(replicaIndex) .setTarget(targetDatanode.getProtoBufMessage()) - .setPriority(priority); + .setPriority(priority) + .setApparentVersion(apparentVersion.serialize()); return builder.build(); } @@ -100,6 +117,10 @@ public static ReplicateContainerCommand getFromProtobuf( if (protoMessage.hasPriority()) { cmd.setPriority(protoMessage.getPriority()); } + if (protoMessage.hasApparentVersion()) { + cmd.apparentVersion = HDDSVersionUtils.deserializeHDDSVersionOrLayoutVersion( + protoMessage.getApparentVersion()); + } return cmd; } @@ -129,6 +150,7 @@ public String toString() { + ", containerId=" + getContainerID() + ", replicaIndex=" + getReplicaIndex() + ", targetNode=" + targetDatanode - + ", priority=" + priority; + + ", priority=" + priority + + ", apparentVersion=" + apparentVersion; } } 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 74edeae7aff9..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 @@ -25,6 +25,7 @@ import java.util.HashMap; import java.util.Map; +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.SCMCommandProto; @@ -39,6 +40,7 @@ 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}. @@ -59,6 +61,17 @@ public void setUp() { stateContext = mock(StateContext.class); } + /** + * 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 = @@ -71,20 +84,24 @@ 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), + 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); when(supervisor.getReplicationRequestCount(metricsName)).thenReturn(5L); @@ -103,4 +120,27 @@ public void testMetrics() { metrics.unRegister(); } } + + /** + * 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 testApparentVersionTakenFromCommand() { + ArgumentCaptor captor = captureSubmittedTask(); + + ReplicateContainerCommandHandler handler = + new ReplicateContainerCommandHandler(supervisor, pushReplicator); + + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + ReplicateContainerCommand cmd = ReplicateContainerCommand.toTarget( + 1, target, HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE); + + handler.handle(cmd, ozoneContainer, stateContext, connectionManager); + + ReplicationTask task = captor.getValue(); + assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE, + task.getApparentVersion()); + } } 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 079c41fcbecc..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 @@ -37,6 +37,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; @@ -170,7 +171,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); @@ -200,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/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 026ac6ac76ad..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,6 +62,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; +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; @@ -801,7 +802,7 @@ private ECReconstructionCoordinatorTask createECTaskWithCoordinator(long contain private ReplicateContainerCommand createCommand(long containerId) { ReplicateContainerCommand cmd = - ReplicateContainerCommand.toTarget(containerId, datanode); + ReplicateContainerCommand.toTarget(containerId, datanode, HDDSVersion.SOFTWARE_VERSION); cmd.setTerm(CURRENT_TERM); return cmd; } @@ -1049,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 new file mode 100644 index 000000000000..ff47ccdb04c3 --- /dev/null +++ b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/protocol/commands/TestReplicateContainerCommand.java @@ -0,0 +1,100 @@ +/* + * 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.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.protocol.proto.StorageContainerDatanodeProtocolProtos.ReplicateContainerCommandProto; +import org.apache.hadoop.hdds.upgrade.HDDSLayoutFeature; +import org.junit.jupiter.api.Test; + +/** + * Test cases to verify {@link ReplicateContainerCommand} serialization. + */ +public class TestReplicateContainerCommand { + + @Test + public void testApparentVersionRoundTrip() { + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + + ReplicateContainerCommand cmd = + ReplicateContainerCommand.toTarget(1L, target, HDDSVersion.ZDU); + + ReplicateContainerCommandProto proto = cmd.getProto(); + ReplicateContainerCommand deserialized = + ReplicateContainerCommand.getFromProtobuf(proto); + + assertEquals(HDDSVersion.ZDU, deserialized.getApparentVersion()); + assertEquals(target.getID(), + deserialized.getTargetDatanode().getID()); + } + + @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(); + + ReplicateContainerCommand cmd = + ReplicateContainerCommand.toTarget(1L, target, HDDSVersion.ZDU); + + ReplicateContainerCommandProto proto = + ReplicateContainerCommandProto.newBuilder() + .setContainerID(1L) + .setCmdId(cmd.getId()) + .setTarget(target.getProtoBufMessage()) + .build(); + + ReplicateContainerCommand deserialized = + ReplicateContainerCommand.getFromProtobuf(proto); + + assertEquals(HDDSVersion.DEFAULT_VERSION, + deserialized.getApparentVersion()); + } + + @Test + public void testToStringIncludesApparentVersion() { + ReplicateContainerCommand cmd = + ReplicateContainerCommand.toTarget(1L, randomDatanodeDetails(), HDDSVersion.SOFTWARE_VERSION); + + String str = cmd.toString(); + 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 3d08f39992c6..4f337361ae61 100644 --- a/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto +++ b/hadoop-hdds/interface-server/src/main/proto/ScmServerDatanodeHeartbeatProtocol.proto @@ -431,6 +431,7 @@ message ReplicateContainerCommandProto { optional int32 replicaIndex = 4; optional DatanodeDetailsProto target = 5; optional ReplicationCommandPriority priority = 6 [default = NORMAL]; + 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 f890fb6a082a..0d251f6d2815 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 @@ -527,10 +527,15 @@ 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, + nodeManager.getLowestApparentVersion(source, target)); + cmd.setReplicaIndex(replicaIndex); + sendDatanodeCommand(cmd, containerInfo, source); + } catch (NodeNotFoundException e) { + throw new IllegalArgumentException("Datanode not found in NodeManager. Should not happen", e); + } } public void sendThrottledReconstructionCommand(ContainerInfo containerInfo, @@ -627,11 +632,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, + nodeManager.getLowestApparentVersion(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", e); + } } /** 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 96eafa0b2aa8..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 @@ -190,6 +190,28 @@ default DatanodeFinalizationCounts getDatanodeFinalizationCounts() { return new DatanodeFinalizationCounts(finalizedNodes, totalHealthyNodes); } + /** + * Returns the lowest apparent 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 ComponentVersion getLowestApparentVersion(DatanodeDetails... nodes) + throws NodeNotFoundException { + ComponentVersion[] versions = new ComponentVersion[nodes.length]; + for (int i = 0; i < nodes.length; i++) { + DatanodeInfo info = getNode(nodes[i].getID()); + if (info == null) { + throw new NodeNotFoundException(nodes[i].getID()); + } + versions[i] = info.getLastKnownApparentVersion(); + } + return ComponentVersion.min(versions); + } + /** * 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/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 217e752278fb..9bee53680e0e 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 bcb3dea97680..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 @@ -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; @@ -89,6 +90,7 @@ import org.apache.hadoop.hdds.scm.exceptions.SCMException; 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; @@ -160,6 +162,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.getNode(any(DatanodeID.class))) + .thenReturn(defaultNodeInfo); + clock = new MockClock(Instant.now(), ZoneId.systemDefault()); containerReplicaPendingOps = new ContainerReplicaPendingOps(clock, null); @@ -1208,7 +1219,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); @@ -1739,4 +1750,68 @@ 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, HDDSVersion version) { + DatanodeInfo info = mock(DatanodeInfo.class); + when(info.getLastKnownApparentVersion()).thenReturn(version); + when(nodeManager.getNode(dn.getID())).thenReturn(info); + return info; + } + + @Test + public void testApparentVersionIsLowestOfSourceAndTarget() + throws NotLeaderException, NodeNotFoundException { + ContainerInfo containerInfo = + ReplicationTestUtil.createContainerInfo(repConfig, 1, + HddsProtos.LifeCycleState.CLOSED, 10, 20); + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + + // 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); + when(nodeManager.getLowestApparentVersion(source, target)) + .thenCallRealMethod(); + + replicationManager.sendLowPriorityReplicateContainerCommand(containerInfo, + 1, source, target, clock.millis() + rmConf.getEventTimeout()); + + assertEquals(HDDSVersion.SEPARATE_RATIS_PORTS_AVAILABLE, + captureSentReplicateCommand().getApparentVersion()); + } + + @Test + public void testApparentVersionLookupThrowsWhenNodeNotFound() + throws NodeNotFoundException { + ContainerInfo containerInfo = + ReplicationTestUtil.createContainerInfo(repConfig, 1, + HddsProtos.LifeCycleState.CLOSED, 10, 20); + DatanodeDetails target = MockDatanodeDetails.randomDatanodeDetails(); + DatanodeDetails source = MockDatanodeDetails.randomDatanodeDetails(); + + mockDatanodeWithApparentVersion(source, HDDSVersion.SOFTWARE_VERSION); + // SCM has no information for the target. + when(nodeManager.getNode(target.getID())).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, () -> + replicationManager.sendLowPriorityReplicateContainerCommand( + containerInfo, 0, source, target, + clock.millis() + rmConf.getEventTimeout())); + } } 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 7662ed5ba78e..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 @@ -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.getNode(any(DatanodeID.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 289b5db922a7..06101b1aae28 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 @@ -904,7 +904,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/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 2fbd737625a5..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 @@ -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; @@ -114,7 +115,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); @@ -136,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); @@ -178,7 +180,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); @@ -224,7 +226,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,