HDDS-15417. Handle mixed datanode versions on the replication path

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/ComponentVersion.java

@@ -80,4 +80,26 @@ default boolean isSupportedBy(ComponentVersion other) {
   default Optional<? extends UpgradeAction> 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;
+  }
 }

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();
   }

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;
   }
 }

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);
   }
 
 }

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<ReplicationTask> captureSubmittedTask() {
+    ArgumentCaptor<ReplicationTask> 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<ReplicationTask> 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());
+  }
 }

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;

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);

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;