HDDS-15614. Remove Datanode download/pull container replication (apache#10584)

Co-authored-by: Doroszlai, Attila <adoroszlai@apache.org>

Author: errose28

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java

@@ -70,17 +70,14 @@
 import org.apache.hadoop.ozone.container.ec.reconstruction.ECReconstructionCoordinator;
 import org.apache.hadoop.ozone.container.ec.reconstruction.ECReconstructionMetrics;
 import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer;
-import org.apache.hadoop.ozone.container.replication.ContainerImporter;
 import org.apache.hadoop.ozone.container.replication.ContainerReplicator;
-import org.apache.hadoop.ozone.container.replication.DownloadAndImportReplicator;
 import org.apache.hadoop.ozone.container.replication.GrpcContainerUploader;
 import org.apache.hadoop.ozone.container.replication.MeasuredReplicator;
 import org.apache.hadoop.ozone.container.replication.OnDemandContainerReplicationSource;
 import org.apache.hadoop.ozone.container.replication.PushReplicator;
 import org.apache.hadoop.ozone.container.replication.ReplicationServer.ReplicationConfig;
 import org.apache.hadoop.ozone.container.replication.ReplicationSupervisor;
 import org.apache.hadoop.ozone.container.replication.ReplicationSupervisorMetrics;
-import org.apache.hadoop.ozone.container.replication.SimpleContainerDownloader;
 import org.apache.hadoop.ozone.container.upgrade.DataNodeUpgradeFinalizer;
 import org.apache.hadoop.ozone.container.upgrade.VersionedDatanodeFeatures;
 import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
@@ -125,7 +122,6 @@ public class DatanodeStateMachine implements Closeable {
    * constructor in a non-thread-safe way - see HDDS-3116.
    */
   private final ReadWriteLock constructionLock = new ReentrantReadWriteLock();
-  private final MeasuredReplicator pullReplicatorWithMetrics;
   private final MeasuredReplicator pushReplicatorWithMetrics;
   private final ReplicationSupervisorMetrics replicationSupervisorMetrics;
   private final NettyMetrics nettyMetrics;
@@ -195,21 +191,11 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
     }
     nextHB = new AtomicLong(Time.monotonicNow());
 
-    ContainerImporter importer = new ContainerImporter(conf,
-        container.getContainerSet(),
-        container.getController(),
-        container.getVolumeSet(),
-        volumeChoosingPolicy);
-    ContainerReplicator pullReplicator = new DownloadAndImportReplicator(
-        conf, container.getContainerSet(),
-        importer,
-        new SimpleContainerDownloader(conf, certClient));
     ContainerReplicator pushReplicator = new PushReplicator(conf,
         new OnDemandContainerReplicationSource(container.getController()),
         new GrpcContainerUploader(conf, certClient, container.getController())
     );
 
-    pullReplicatorWithMetrics = new MeasuredReplicator(pullReplicator, "pull");
     pushReplicatorWithMetrics = new MeasuredReplicator(pushReplicator, "push");
 
     ReplicationConfig replicationConfig =
@@ -266,8 +252,7 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
             dnConf.getCommandQueueLimit(), threadNamePrefix))
         .addHandler(new DeleteBlocksCommandHandler(getContainer(),
             conf, dnConf, threadNamePrefix))
-        .addHandler(new ReplicateContainerCommandHandler(conf, supervisor,
-            pullReplicatorWithMetrics, pushReplicatorWithMetrics))
+        .addHandler(new ReplicateContainerCommandHandler(supervisor, pushReplicatorWithMetrics))
         .addHandler(reconstructECContainersCommandHandler)
         .addHandler(new DeleteContainerCommandHandler(
             dnConf.getContainerDeleteThreads(), clock,
@@ -652,7 +637,7 @@ public EnumCounters<SCMCommandProto.Type> getQueuedCommandCount() {
    */
   public synchronized void stopDaemon() {
     try {
-      IOUtils.close(LOG, pushReplicatorWithMetrics, pullReplicatorWithMetrics);
+      IOUtils.close(LOG, pushReplicatorWithMetrics);
       supervisor.stop();
       context.setShutdownGracefully();
       context.setState(DatanodeStates.SHUTDOWN);

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/commandhandler/ReplicateContainerCommandHandler.java

@@ -18,9 +18,6 @@
 package org.apache.hadoop.ozone.container.common.statemachine.commandhandler;
 
 import com.google.common.base.Preconditions;
-import java.util.List;
-import org.apache.hadoop.hdds.conf.ConfigurationSource;
-import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto.Type;
 import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager;
@@ -31,32 +28,20 @@
 import org.apache.hadoop.ozone.container.replication.ReplicationTask;
 import org.apache.hadoop.ozone.protocol.commands.ReplicateContainerCommand;
 import org.apache.hadoop.ozone.protocol.commands.SCMCommand;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
- * Command handler to copy containers from sources.
+ * Command handler to push containers to a target datanode.
  */
 public class ReplicateContainerCommandHandler implements CommandHandler {
 
-  static final Logger LOG =
-      LoggerFactory.getLogger(ReplicateContainerCommandHandler.class);
-
   private ReplicationSupervisor supervisor;
 
-  private ContainerReplicator downloadReplicator;
-
   private ContainerReplicator pushReplicator;
 
   private static final String METRIC_NAME = ReplicationTask.METRIC_NAME;
 
-  public ReplicateContainerCommandHandler(
-      ConfigurationSource conf,
-      ReplicationSupervisor supervisor,
-      ContainerReplicator downloadReplicator,
-      ContainerReplicator pushReplicator) {
+  public ReplicateContainerCommandHandler(ReplicationSupervisor supervisor, ContainerReplicator pushReplicator) {
     this.supervisor = supervisor;
-    this.downloadReplicator = downloadReplicator;
     this.pushReplicator = pushReplicator;
   }
 
@@ -70,20 +55,13 @@ public void handle(SCMCommand<?> command, OzoneContainer container,
 
     final ReplicateContainerCommand replicateCommand =
         (ReplicateContainerCommand) command;
-    final List<DatanodeDetails> sourceDatanodes =
-        replicateCommand.getSourceDatanodes();
     final long containerID = replicateCommand.getContainerID();
-    final DatanodeDetails target = replicateCommand.getTargetDatanode();
-
-    Preconditions.checkArgument(!sourceDatanodes.isEmpty() || target != null,
-        "Replication command is received for container %s "
-            + "without source or target datanodes.", containerID);
 
-    ContainerReplicator replicator =
-        replicateCommand.getTargetDatanode() == null ?
-            downloadReplicator : pushReplicator;
+    Preconditions.checkArgument(replicateCommand.getTargetDatanode() != null,
+        "Replication command received for container %s without a target datanode.",
+        containerID);
 
-    ReplicationTask task = new ReplicationTask(replicateCommand, replicator);
+    ReplicationTask task = new ReplicationTask(replicateCommand, pushReplicator);
     supervisor.addTask(task);
   }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/OzoneContainer.java

@@ -240,7 +240,6 @@ public OzoneContainer(HddsDatanodeService hddsDatanodeService,
         context);
 
     replicationServer = new ReplicationServer(
-        controller,
         conf.getObject(ReplicationConfig.class),
         secConf,
         certClient,

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/AbstractReplicationTask.java

@@ -42,8 +42,6 @@ public abstract class AbstractReplicationTask {
 
   private ReplicationCommandPriority priority = NORMAL;
 
-  private boolean shouldOnlyRunOnInServiceDatanodes = true;
-
   protected AbstractReplicationTask(long containerID,
       long deadlineMsSinceEpoch, long term) {
     this(containerID, deadlineMsSinceEpoch, term,
@@ -115,24 +113,6 @@ public ReplicationCommandPriority getPriority() {
     return priority;
   }
 
-  /**
-   * Returns true if the task should only run on in service datanodes. False
-   * otherwise.
-   */
-  public boolean shouldOnlyRunOnInServiceDatanodes() {
-    return shouldOnlyRunOnInServiceDatanodes;
-  }
-
-  /**
-   * Set whether the task should only run on in service datanodes. Passing false
-   * allows the task to run on out of service datanodes as well.
-   * @param runOnInServiceOnly
-   */
-  protected void setShouldOnlyRunOnInServiceDatanodes(
-      boolean runOnInServiceOnly) {
-    this.shouldOnlyRunOnInServiceDatanodes = runOnInServiceOnly;
-  }
-
   /**
    * Hook for subclasses to provide info about the command.
    * @return string representation of the command

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java

@@ -104,7 +104,7 @@ protected GrpcReplicationClient createReplicationClient(
       throws IOException {
     return new GrpcReplicationClient(target.getIpAddress(),
         target.getPort(Port.Name.REPLICATION).getValue(),
-        securityConfig, certClient, compression);
+        securityConfig, certClient);
   }
 
   /**