HDDS-12991 recreate pipelines after enabling ratis write streaming

hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java

@@ -384,6 +384,19 @@ public synchronized boolean hasPort(Port.Name name) {
     return false;
   }
 
+  /**
+   * Whether this datanode exposes a port name that {@code other} lacks.
+   *
+   * @param other a previously recorded snapshot of this node
+   * @return true if this node advertises a port name not present in
+   *     {@code other}
+   */
+  public boolean exposesNewPorts(DatanodeDetails other) {
+    return getPorts().stream()
+        .map(Port::getName)
+        .anyMatch(name -> !other.hasPort(name));
+  }
+
   /**
    * Helper method to get the Ratis port.
    * 

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java

@@ -474,6 +474,11 @@ public RegisteredCommand register(
               "oldVersion = {}, newVersion = {}.",
               datanodeDetails, oldNode.getVersion(), datanodeDetails.getVersion());
           nodeStateManager.updateNode(datanodeDetails, layoutInfo);
+        } else if (portsChanged(oldNode, datanodeDetails)) {
+          // Refresh the stored node when its port set changes
+          LOG.info("Updating ports for registered datanode {}: {} -> {}",
+              datanodeDetails, oldNode.getPorts(), datanodeDetails.getPorts());
+          nodeStateManager.updateNode(datanodeDetails, layoutInfo);
         }
       } catch (NodeNotFoundException e) {
         LOG.error("Cannot find datanode {} from nodeStateManager",
@@ -487,6 +492,22 @@ public RegisteredCommand register(
         .build();
   }
 
+  /**
+   * Whether the datanode's exposed ports changed between two registrations.
+   * Compared as a set of name=value entries, since
+   * {@link DatanodeDetails.Port#equals} ignores the port value.
+   */
+  private static boolean portsChanged(DatanodeDetails oldNode,
+      DatanodeDetails newNode) {
+    return !portValues(oldNode).equals(portValues(newNode));
+  }
+
+  private static Set<String> portValues(DatanodeDetails datanodeDetails) {
+    return datanodeDetails.getPorts().stream()
+        .map(port -> port.getName() + "=" + port.getValue())
+        .collect(Collectors.toSet());
+  }
+
   /**
    * Add an entry to the dnsToUuidMap, which maps hostname / IP to the DNs
    * running on that host. As each address can have many DNs running on it,

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java

@@ -117,6 +117,14 @@ void addContainerToPipeline(PipelineID pipelineID, ContainerID containerID)
 
   void closeStalePipelines(DatanodeDetails datanodeDetails);
 
+  /**
+   * Close OPEN pipelines whose datanodes now expose a port name that the
+   * pipeline's stored node snapshot lacks, so fresh pipelines advertising the
+   * new port are created in their place. A pipeline's cannot pick up
+   * a newly advertised port in place, so it must be recreated.
+   */
+  void closePipelinesExposingNewPorts();
+
   void scrubPipelines() throws IOException;
 
   void startPipelineCreator();

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java

@@ -182,13 +182,9 @@ public static PipelineManagerImpl newPipelineManager(
             .setServiceName("BackgroundPipelineScrubber")
             .setIntervalInMillis(scrubberIntervalInMillis)
             .setWaitTimeInMillis(safeModeWaitMs)
-            .setPeriodicalTask(() -> {
-              try {
-                pipelineManager.scrubPipelines();
-              } catch (IOException e) {
-                LOG.error("Unexpected error during pipeline scrubbing", e);
-              }
-            }).build();
+            .setPeriodicalTask(
+                pipelineManager::scrubAndClosePipelinesExposingNewPorts)
+            .build();
 
     pipelineManager.setBackgroundPipelineScrubber(backgroundPipelineScrubber);
     serviceManager.register(backgroundPipelineScrubber);
@@ -562,6 +558,52 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
         ||  !left.getHostName().equals(right.getHostName()));
   }
 
+  /**
+   * Scrub pipelines, then close (and delete) OPEN pipelines whose registered
+   * nodes now expose a port name the pipeline's stored node snapshot lacks.
+   */
+  void scrubAndClosePipelinesExposingNewPorts() {
+    try {
+      scrubPipelines();
+    } catch (IOException e) {
+      LOG.error("Unexpected error during pipeline scrubbing", e);
+    }
+    closePipelinesExposingNewPorts();
+  }
+
+  @Override
+  public void closePipelinesExposingNewPorts() {
+    for (Pipeline pipeline : getPipelines()) {
+      if (!pipeline.isOpen() || !nodesExposeNewPorts(pipeline)) {
+        continue;
+      }
+      try {
+        final PipelineID id = pipeline.getId();
+        LOG.info("Closing pipeline {} whose nodes expose a new port so a "
+            + "pipeline advertising it can replace it", id);
+        closePipeline(id);
+        deletePipeline(id);
+      } catch (IOException e) {
+        LOG.error("Failed to close pipeline {} exposing a new port",
+            pipeline.getId(), e);
+      }
+    }
+  }
+
+  /**
+   * Whether any registered node of the pipeline exposes a port name that the
+   * pipeline's stored copy of that node lacks.
+   */
+  private boolean nodesExposeNewPorts(Pipeline pipeline) {
+    for (DatanodeDetails stored : pipeline.getNodes()) {
+      final DatanodeDetails current = nodeManager.getNode(stored.getID());
+      if (current != null && current.exposesNewPorts(stored)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   /**
    * Scrub pipelines.
    */

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java

@@ -43,6 +43,7 @@
 import static org.apache.ozone.test.MetricsAsserts.getMetrics;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -63,6 +64,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import java.util.UUID;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
@@ -344,6 +346,81 @@ private DatanodeDetails registerWithCapacity(SCMNodeManager nodeManager,
     return cmd.getDatanode();
   }
 
+  private static DatanodeDetails.Builder datanodeWithoutDatastream(UUID uuid) {
+    return DatanodeDetails.newBuilder()
+        .setUuid(uuid)
+        .setHostName("host-" + uuid)
+        .setIpAddress("127.0.0.1")
+        .addPort(DatanodeDetails.newPort(
+            DatanodeDetails.Port.Name.STANDALONE, 9859))
+        .addPort(DatanodeDetails.newPort(
+            DatanodeDetails.Port.Name.RATIS, 9858));
+  }
+
+  private void registerNode(SCMNodeManager nodeManager, DatanodeDetails dn) {
+    StorageReportProto storageReport = HddsTestUtils.createStorageReport(
+        dn.getID(), dn.getNetworkFullPath(), Long.MAX_VALUE);
+    MetadataStorageReportProto metadataStorageReport =
+        HddsTestUtils.createMetadataStorageReport(
+            dn.getNetworkFullPath(), Long.MAX_VALUE);
+    RegisteredCommand cmd = nodeManager.register(dn,
+        HddsTestUtils.createNodeReport(Arrays.asList(storageReport),
+            Arrays.asList(metadataStorageReport)),
+        getRandomPipelineReports(), UpgradeUtils.defaultLayoutVersionProto());
+    assertEquals(success, cmd.getError());
+  }
+
+  /**
+   * A datanode that re-registers with the same identity but now exposes the
+   * RATIS_DATASTREAM port (e.g. Ratis DataStream was enabled) must have its
+   * stored record refreshed so the new port is visible (HDDS-15799).
+   */
+  @Test
+  public void testRegisterRefreshesPortsOnPortChange()
+      throws IOException, AuthenticationException {
+    try (SCMNodeManager nodeManager = createNodeManager(getConf())) {
+      final UUID uuid = UUID.randomUUID();
+
+      // First registration: streaming disabled, no RATIS_DATASTREAM port.
+      registerNode(nodeManager, datanodeWithoutDatastream(uuid).build());
+      DatanodeDetails stored = nodeManager.getNode(
+          datanodeWithoutDatastream(uuid).build().getID());
+      assertFalse(stored.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM));
+
+      // Re-registration (same id/ip/host/version) now exposing the port.
+      registerNode(nodeManager, datanodeWithoutDatastream(uuid)
+          .addPort(DatanodeDetails.newPort(
+              DatanodeDetails.Port.Name.RATIS_DATASTREAM, 9855))
+          .build());
+      stored = nodeManager.getNode(
+          datanodeWithoutDatastream(uuid).build().getID());
+      assertTrue(stored.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM),
+          "stored node should be refreshed with the RATIS_DATASTREAM port");
+    }
+  }
+
+  /**
+   * Re-registering a datanode with an unchanged port set must not disturb the
+   * stored record (the port-refresh branch is skipped).
+   */
+  @Test
+  public void testRegisterKeepsPortsWhenUnchanged()
+      throws IOException, AuthenticationException {
+    try (SCMNodeManager nodeManager = createNodeManager(getConf())) {
+      final UUID uuid = UUID.randomUUID();
+
+      registerNode(nodeManager, datanodeWithoutDatastream(uuid).build());
+      // Re-register with the identical port set.
+      registerNode(nodeManager, datanodeWithoutDatastream(uuid).build());
+
+      final DatanodeDetails stored = nodeManager.getNode(
+          datanodeWithoutDatastream(uuid).build().getID());
+      assertTrue(stored.hasPort(DatanodeDetails.Port.Name.STANDALONE));
+      assertTrue(stored.hasPort(DatanodeDetails.Port.Name.RATIS));
+      assertFalse(stored.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM));
+    }
+  }
+
   private void assertPipelineClosedAfterLayoutHeartbeat(
       DatanodeDetails originalNode1, DatanodeDetails originalNode2,
       SCMNodeManager nodeManager, LayoutVersionProto layout) throws Exception {

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java

@@ -249,6 +249,11 @@ public void closeStalePipelines(DatanodeDetails datanodeDetails) {
 
   }
 
+  @Override
+  public void closePipelinesExposingNewPorts() {
+
+  }
+
   @Override
   public void scrubPipelines() {
 

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java

@@ -88,6 +88,7 @@
 import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub;
 import org.apache.hadoop.hdds.scm.ha.SCMServiceManager;
 import org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition;
+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.pipeline.choose.algorithms.HealthyPipelineChoosePolicy;
@@ -100,6 +101,7 @@
 import org.apache.hadoop.hdds.utils.db.DBStoreBuilder;
 import org.apache.hadoop.hdds.utils.db.Table;
 import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.apache.hadoop.ozone.ClientVersion;
 import org.apache.hadoop.ozone.container.common.SCMTestUtils;
 import org.apache.ozone.test.GenericTestUtils;
 import org.apache.ozone.test.GenericTestUtils.LogCapturer;
@@ -980,4 +982,115 @@ private static void assertFailsNotLeader(CheckedRunnable<?> block) {
     assertEquals(ResultCodes.SCM_NOT_LEADER, e.getResult());
     assertInstanceOf(NotLeaderException.class, e.getCause());
   }
+
+  private static DatanodeDetails portlessDatanode(DatanodeID id) {
+    return DatanodeDetails.newBuilder()
+        .setID(id)
+        .setHostName("host-" + id)
+        .setIpAddress("127.0.0.1")
+        .addPort(DatanodeDetails.newPort(
+            DatanodeDetails.Port.Name.STANDALONE, 9859))
+        .addPort(DatanodeDetails.newPort(
+            DatanodeDetails.Port.Name.RATIS, 9858))
+        .build();
+  }
+
+  private Pipeline addPipeline(PipelineManagerImpl pipelineManager,
+      Pipeline.PipelineState state, List<DatanodeDetails> nodes)
+      throws IOException {
+    final Pipeline pipeline = Pipeline.newBuilder()
+        .setReplicationConfig(
+            RatisReplicationConfig.getInstance(ReplicationFactor.THREE))
+        .setNodes(nodes)
+        .setState(state)
+        .setId(PipelineID.randomId())
+        .build();
+    pipelineManager.getStateManager().addPipeline(
+        pipeline.getProtobufMessage(ClientVersion.CURRENT_VERSION));
+    return pipeline;
+  }
+
+  private static boolean exists(PipelineManagerImpl pipelineManager,
+      PipelineID id) {
+    try {
+      pipelineManager.getPipeline(id);
+      return true;
+    } catch (PipelineNotFoundException e) {
+      return false;
+    }
+  }
+
+  @Test
+  public void testClosePipelinesExposingNewPorts() throws Exception {
+    try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) {
+      // Registered datanodes (MockNodeManager) expose all ports incl datastream.
+      final List<DatanodeInfo> registered = nodeManager.getAllNodes();
+      final List<DatanodeDetails> idsA = new ArrayList<>();
+      final List<DatanodeDetails> idsB = new ArrayList<>();
+      for (int i = 0; i < 3; i++) {
+        idsA.add(portlessDatanode(registered.get(i).getID()));
+        idsB.add(portlessDatanode(registered.get(i + 3).getID()));
+      }
+
+      // OPEN, registered nodes, portless -> legacy pipeline, must be closed.
+      final Pipeline stale = addPipeline(pipelineManager, OPEN, idsA);
+      // OPEN, registered nodes carrying all ports -> not stale, kept.
+      final Pipeline portful = addPipeline(pipelineManager, OPEN,
+          new ArrayList<>(registered.subList(6, 9)));
+      // OPEN, but nodes are NOT registered -> cannot heal, left alone.
+      final List<DatanodeDetails> unregistered = new ArrayList<>();
+      for (int i = 0; i < 3; i++) {
+        unregistered.add(portlessDatanode(DatanodeID.randomID()));
+      }
+      final Pipeline unreg = addPipeline(pipelineManager, OPEN, unregistered);
+      // ALLOCATED (non-open) portless -> skipped.
+      final Pipeline allocated = addPipeline(pipelineManager, ALLOCATED, idsB);
+
+      pipelineManager.closePipelinesExposingNewPorts();
+
+      assertFalse(exists(pipelineManager, stale.getId()),
+          "OPEN pipeline whose nodes expose new ports should be closed and deleted");
+      assertTrue(exists(pipelineManager, portful.getId()));
+      assertTrue(exists(pipelineManager, unreg.getId()));
+      assertTrue(exists(pipelineManager, allocated.getId()));
+    }
+  }
+
+  @Test
+  public void testClosePipelinesExposingNewPortsSwallowsError() throws Exception {
+    try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) {
+      final List<DatanodeInfo> registered = nodeManager.getAllNodes();
+      final List<DatanodeDetails> nodes = new ArrayList<>();
+      for (int i = 0; i < 3; i++) {
+        nodes.add(portlessDatanode(registered.get(i).getID()));
+      }
+      final Pipeline stale = addPipeline(pipelineManager, OPEN, nodes);
+
+      final PipelineManagerImpl spy = spy(pipelineManager);
+      doThrow(new IOException("boom")).when(spy).closePipeline(stale.getId());
+      // The close failure is logged and swallowed; the loop does not throw.
+      spy.closePipelinesExposingNewPorts();
+      assertTrue(exists(pipelineManager, stale.getId()));
+    }
+  }
+
+  @Test
+  public void testScrubAndCloseWiring() throws Exception {
+    // The background task scrubs then closes pipelines exposing new ports; on
+    // an empty manager both are no-ops and must not throw.
+    try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) {
+      pipelineManager.scrubAndClosePipelinesExposingNewPorts();
+    }
+  }
+
+  @Test
+  public void testScrubAndCloseSwallowsScrubError() throws Exception {
+    try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) {
+      final PipelineManagerImpl spy = spy(pipelineManager);
+      doThrow(new IOException("boom")).when(spy).scrubPipelines();
+      // Scrub failure is logged and swallowed; the close pass still runs.
+      spy.scrubAndClosePipelinesExposingNewPorts();
+      verify(spy).closePipelinesExposingNewPorts();
+    }
+  }
 }