HDDS-12991 recreate pipelines after enabling ratis write streaming

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

@@ -474,6 +474,14 @@ 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 (e.g. a datanode
+          // restarts with Ratis DataStream enabled and now exposes the
+          // RATIS_DATASTREAM port). Otherwise the stale record would keep
+          // streaming clients from reaching the datastream port (HDDS-15799).
+          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 +495,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,15 @@ void addContainerToPipeline(PipelineID pipelineID, ContainerID containerID)
 
   void closeStalePipelines(DatanodeDetails datanodeDetails);
 
+  /**
+   * Close OPEN pipelines whose datanodes now expose a port (the
+   * RATIS_DATASTREAM port after Ratis DataStream was enabled) that the
+   * pipeline's stored node snapshot lacks, so streaming-capable pipelines are
+   * created in their place. A pre-datastream Raft group cannot be made
+   * streamable in place, so it must be recreated (HDDS-12991).
+   */
+  void closeNonStreamablePipelines();
+
   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::scrubAndCloseNonStreamablePipelines)
+            .build();
 
     pipelineManager.setBackgroundPipelineScrubber(backgroundPipelineScrubber);
     serviceManager.register(backgroundPipelineScrubber);
@@ -562,6 +558,68 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
         ||  !left.getHostName().equals(right.getHostName()));
   }
 
+  /**
+   * Close (and delete) OPEN pipelines that predate a datanode capability the
+   * registered node now advertises but the pipeline's stored node snapshot
+   * lacks — in practice the RATIS_DATASTREAM port after Ratis DataStream was
+   * enabled. Such pipelines cannot serve streaming even after the datanodes
+   * restart, because the Raft group's persisted configuration still carries the
+   * stale datastream address; only a freshly created pipeline is
+   * streaming-capable. BackgroundPipelineCreator recreates replacements from
+   * the now-capable nodes (HDDS-12991).
+   */
+  void scrubAndCloseNonStreamablePipelines() {
+    try {
+      scrubPipelines();
+    } catch (IOException e) {
+      LOG.error("Unexpected error during pipeline scrubbing", e);
+    }
+    closeNonStreamablePipelines();
+  }
+
+  @Override
+  public void closeNonStreamablePipelines() {
+    for (Pipeline pipeline : getPipelines()) {
+      if (!pipeline.isOpen() || !nodesExposeNewPorts(pipeline)) {
+        continue;
+      }
+      try {
+        final PipelineID id = pipeline.getId();
+        LOG.info("Closing non-streamable pipeline {} so a streaming-capable "
+            + "pipeline can replace it", id);
+        closePipeline(id);
+        deletePipeline(id);
+      } catch (IOException e) {
+        LOG.error("Failed to close non-streamable pipeline {}",
+            pipeline.getId(), e);
+      }
+    }
+  }
+
+  /**
+   * Whether any registered node of the pipeline exposes a port name that the
+   * pipeline's stored copy of that node lacks (e.g. RATIS_DATASTREAM added
+   * after the pipeline was created).
+   */
+  private boolean nodesExposeNewPorts(Pipeline pipeline) {
+    for (DatanodeDetails stored : pipeline.getNodes()) {
+      final DatanodeDetails current = nodeManager.getNode(stored.getID());
+      if (current != null && exposesNewPorts(stored, current)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  private static boolean exposesNewPorts(DatanodeDetails stored,
+      DatanodeDetails current) {
+    final Set<DatanodeDetails.Port.Name> storedNames = stored.getPorts().stream()
+        .map(DatanodeDetails.Port::getName).collect(Collectors.toSet());
+    return current.getPorts().stream()
+        .map(DatanodeDetails.Port::getName)
+        .anyMatch(name -> !storedNames.contains(name));
+  }
+
   /**
    * 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 closeNonStreamablePipelines() {
+
+  }
+
   @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 testCloseNonStreamablePipelines() 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.closeNonStreamablePipelines();
+
+      assertFalse(exists(pipelineManager, stale.getId()),
+          "legacy portless OPEN pipeline should be closed and deleted");
+      assertTrue(exists(pipelineManager, portful.getId()));
+      assertTrue(exists(pipelineManager, unreg.getId()));
+      assertTrue(exists(pipelineManager, allocated.getId()));
+    }
+  }
+
+  @Test
+  public void testCloseNonStreamablePipelinesSwallowsError() 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.closeNonStreamablePipelines();
+      assertTrue(exists(pipelineManager, stale.getId()));
+    }
+  }
+
+  @Test
+  public void testScrubAndCloseWiring() throws Exception {
+    // The background task scrubs then closes non-streamable pipelines; on an
+    // empty manager both are no-ops and must not throw.
+    try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) {
+      pipelineManager.scrubAndCloseNonStreamablePipelines();
+    }
+  }
+
+  @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.scrubAndCloseNonStreamablePipelines();
+      verify(spy).closeNonStreamablePipelines();
+    }
+  }
 }