HDDS-12991. Support enabling Ratis DataStream on an existing cluster

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java

@@ -196,7 +196,8 @@ public BlockDataStreamOutput(
   private DataStreamOutput setupStream(Pipeline pipeline) throws IOException {
     for (DatanodeDetails dn : pipeline.getNodes()) {
       if (!dn.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM)) {
-        throw new IOException("RATIS_DATASTREAM port is missing for datanode "
+        throw new StreamNotSupportedException(
+            "RATIS_DATASTREAM port is missing for datanode "
             + dn + " in pipeline " + pipeline.getId()
             + "; datastream is disabled for this pipeline");
       }

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamNotSupportedException.java

@@ -0,0 +1,35 @@
+/*
+ * 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.hdds.scm.storage;
+
+import java.io.IOException;
+
+/**
+ * Thrown when a Ratis DataStream write cannot proceed because the target
+ * pipeline does not support streaming (e.g. its datanodes were created before
+ * DataStream was enabled and therefore lack the RATIS_DATASTREAM port).
+ *
+ * <p>Callers may catch this to fall back to the non-streaming write path
+ * instead of failing the write (HDDS-12991).
+ */
+public class StreamNotSupportedException extends IOException {
+
+  public StreamNotSupportedException(String message) {
+    super(message);
+  }
+}

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() {