From 917170c6be95fc42e5e0213dfb7beb389cf2eb13 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 22 Jul 2026 10:47:40 -0400 Subject: [PATCH] HDDS-12991 recreate pipelines after enabling ratis write streaming --- .../hadoop/hdds/scm/node/SCMNodeManager.java | 24 ++ .../hdds/scm/pipeline/PipelineManager.java | 9 + .../scm/pipeline/PipelineManagerImpl.java | 72 +++- .../hdds/scm/node/TestSCMNodeManager.java | 77 ++++ .../scm/pipeline/MockPipelineManager.java | 5 + .../scm/pipeline/TestPipelineManagerImpl.java | 113 ++++++ ...stOzoneFileSystemDataStreamEnablement.java | 360 ++++++++++++++++++ 7 files changed, 653 insertions(+), 7 deletions(-) create mode 100644 hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java index cbcfc5537913..97bc9f5705ce 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java +++ b/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 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, diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java index e0d9de1f10b5..fa98650e6522 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java +++ b/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(); diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java index 8003d8d52e96..42a0ea4221f6 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java +++ b/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 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. */ diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java index a0f1c1bf3623..78814b53e81a 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java +++ b/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 { diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java index ef6f187b040c..473fd0b0ea27 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java +++ b/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() { diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java index 2f1d5ede5d7f..b9bc62a84434 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java +++ b/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 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 registered = nodeManager.getAllNodes(); + final List idsA = new ArrayList<>(); + final List 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 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 registered = nodeManager.getAllNodes(); + final List 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(); + } + } } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java new file mode 100644 index 000000000000..30bd04597b42 --- /dev/null +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java @@ -0,0 +1,360 @@ +/* + * 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.fs.ozone; + +import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_HEARTBEAT_INTERVAL; +import static org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name.RATIS_DATASTREAM; +import static org.apache.hadoop.hdds.protocol.proto.HddsProtos.ReplicationFactor.THREE; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_DEADNODE_INTERVAL; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_PIPELINE_CREATION_INTERVAL; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_RATIS_PIPELINE_LIMIT; +import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_STALENODE_INTERVAL; +import static org.apache.hadoop.ozone.OzoneConfigKeys.HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_DATASTREAM_AUTO_THRESHOLD; +import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_FS_DATASTREAM_ENABLED; +import static org.apache.hadoop.ozone.OzoneConsts.OZONE_URI_SCHEME; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.util.List; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.BooleanSupplier; +import org.apache.hadoop.fs.CommonConfigurationKeysPublic; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FSDataOutputStream; +import org.apache.hadoop.fs.FileSystem; +import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hdds.client.RatisReplicationConfig; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.conf.StorageUnit; +import org.apache.hadoop.hdds.protocol.DatanodeDetails; +import org.apache.hadoop.hdds.scm.pipeline.Pipeline; +import org.apache.hadoop.hdds.scm.pipeline.PipelineManager; +import org.apache.hadoop.hdds.utils.IOUtils; +import org.apache.hadoop.ozone.ClientConfigForTesting; +import org.apache.hadoop.ozone.MiniOzoneCluster; +import org.apache.hadoop.ozone.TestDataUtil; +import org.apache.hadoop.ozone.client.OzoneBucket; +import org.apache.hadoop.ozone.client.OzoneClient; +import org.apache.hadoop.ozone.client.io.SelectorOutputStream; +import org.apache.hadoop.ozone.om.helpers.BucketLayout; +import org.apache.ozone.test.GenericTestUtils; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; + +/** + * End-to-end tests for enabling Ratis DataStream on a running cluster + * (HDDS-12991). The two tests are isolated (separate clusters): + *
    + *
  • {@link #testDataStreamFallbackAndPortRefresh()}: a streaming client + * gracefully falls back to the non-streaming path while a pipeline lacks the + * RATIS_DATASTREAM port, and SCM refreshes a datanode's ports when it + * re-registers with datastream enabled.
    • + *
  • {@link #testCloseNonStreamablePipelineThenStream()}: SCM closes a + * pipeline created before datastream (which can never stream in place) so a + * fresh streaming-capable pipeline replaces it, after which a streaming write + * succeeds end-to-end.
    • + *
+ */ +public class TestOzoneFileSystemDataStreamEnablement { + + // Small threshold/payload keep the writes fast while still selecting the + // streaming path (payload > threshold). + private static final int AUTO_THRESHOLD = 4 << 10; + private static final int WRITE_SIZE = 256 << 10; + + private MiniOzoneCluster cluster; + private OzoneClient client; + private OzoneBucket bucket; + private OzoneConfiguration conf; + + private void startClusterWithDatanodeStreamDisabled() throws Exception { + conf = new OzoneConfiguration(); + // Datanode side: datastream initially disabled, so pipelines are created + // without the RATIS_DATASTREAM port. + conf.setBoolean(HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, false); + // Client side: always attempt streaming writes. + conf.setBoolean(OZONE_FS_DATASTREAM_ENABLED, true); + conf.set(OZONE_FS_DATASTREAM_AUTO_THRESHOLD, AUTO_THRESHOLD + "B"); + conf.setInt(OZONE_SCM_RATIS_PIPELINE_LIMIT, 10); + // A long stale interval keeps the OPEN pipeline alive across the (no + // stop-wait) rolling restart, so the test drives pipeline closure itself. + conf.set(OZONE_SCM_STALENODE_INTERVAL, "5m"); + conf.set(OZONE_SCM_DEADNODE_INTERVAL, "10m"); + conf.set(HDDS_HEARTBEAT_INTERVAL, "1s"); + conf.set(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, "1s"); + // Recreate pipelines quickly after a close so the test does not wait the + // default two minutes for a fresh RATIS/THREE pipeline. + conf.set(OZONE_SCM_PIPELINE_CREATION_INTERVAL, "1s"); + + final int chunkSize = 16 << 10; + ClientConfigForTesting.newBuilder(StorageUnit.BYTES) + .setChunkSize(chunkSize) + .setStreamBufferFlushSize(32 << 10) + .setStreamBufferMaxSize(64 << 10) + .setDataStreamBufferFlushSize(64 << 10) + .setDataStreamMinPacketSize(chunkSize) + .setDataStreamWindowSize(5 * chunkSize) + .setBlockSize(1 << 20) + .applyTo(conf); + + cluster = MiniOzoneCluster.newBuilder(conf).setNumDatanodes(3).build(); + cluster.waitForClusterToBeReady(); + client = cluster.newClient(); + bucket = TestDataUtil.createVolumeAndBucket(client, + BucketLayout.FILE_SYSTEM_OPTIMIZED); + } + + @AfterEach + public void teardown() { + IOUtils.closeQuietly(client); + if (cluster != null) { + cluster.shutdown(); + } + } + + private FileSystem fs() throws IOException { + final String rootPath = String.format("%s://%s.%s/", + OZONE_URI_SCHEME, bucket.getName(), bucket.getVolumeName()); + conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, rootPath); + return FileSystem.get(conf); + } + + /** Write {@code data} and return the underlying stream selected by the FS. */ + private static Class writeAndGetUnderlying(FileSystem fs, Path path, + byte[] data) throws IOException { + final FSDataOutputStream out = fs.create(path, true); + out.write(data); + final SelectorOutputStream selector = + (SelectorOutputStream) out.getWrappedStream(); + out.close(); + return selector.getUnderlying().getClass(); + } + + private static void assertRoundTrips(FileSystem fs, Path path, byte[] expected) + throws IOException { + final byte[] read = new byte[expected.length]; + try (FSDataInputStream in = fs.open(path)) { + in.readFully(read); + } + assertArrayEquals(expected, read); + } + + private static byte[] randomBytes() { + final byte[] bytes = new byte[WRITE_SIZE]; + ThreadLocalRandom.current().nextBytes(bytes); + return bytes; + } + + private List openRatisThreePipelines() { + return cluster.getStorageContainerManager().getPipelineManager() + .getPipelines(RatisReplicationConfig.getInstance(THREE), + Pipeline.PipelineState.OPEN); + } + + private static boolean allNodesHaveDatastreamPort(Pipeline pipeline) { + return pipeline.getNodes().stream() + .allMatch(n -> n.hasPort(RATIS_DATASTREAM)); + } + + /** + * Enable datastream on every datanode via a rolling restart. {@code false} + * (no stop-wait) keeps each restart short; combined with the long stale + * interval the OPEN pipeline survives, so its node snapshot stays portless. + */ + private void rollingRestartEnablingDataStream() throws Exception { + for (int i = 0; i < cluster.getHddsDatanodes().size(); i++) { + cluster.getHddsDatanodes().get(i).getConf() + .setBoolean(HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, true); + cluster.restartHddsDatanode(i, false); + } + cluster.waitForClusterToBeReady(); + } + + /** Poll until SCM's node records all expose RATIS_DATASTREAM (validates D). */ + private void waitForAllRegisteredNodesToHaveDatastreamPort() + throws InterruptedException, TimeoutException { + final BooleanSupplier ready = () -> { + final List nodes = cluster + .getStorageContainerManager().getScmNodeManager().getAllNodes(); + return nodes.size() == cluster.getHddsDatanodes().size() + && nodes.stream().allMatch(n -> n.hasPort(RATIS_DATASTREAM)); + }; + GenericTestUtils.waitFor(ready, 500, 30_000); + } + + /** Poll until an OPEN RATIS/THREE pipeline exposes RATIS_DATASTREAM ports. */ + private void waitForStreamablePipeline() + throws InterruptedException, TimeoutException { + final BooleanSupplier ready = () -> openRatisThreePipelines().stream() + .anyMatch(TestOzoneFileSystemDataStreamEnablement + ::allNodesHaveDatastreamPort); + GenericTestUtils.waitFor(ready, 500, 60_000); + } + + /** + * While the datanodes have datastream disabled, a streaming client write + * falls back to the non-streaming path and succeeds. After a rolling restart + * enables datastream, SCM refreshes the datanodes' ports; the pre-existing + * pipeline is still portless, so writes keep falling back gracefully until it + * is replaced. + */ + @Test + @Timeout(value = 50, unit = TimeUnit.SECONDS) + public void testDataStreamFallbackAndPortRefresh() throws Exception { + startClusterWithDatanodeStreamDisabled(); + + try (FileSystem fs = fs()) { + final byte[] data = randomBytes(); + + // Datanode datastream disabled -> streaming falls back, still writes. + final Path before = new Path("/before-enable.dat"); + assertEquals(CapableOzoneFSOutputStream.class, + writeAndGetUnderlying(fs, before, data)); + assertRoundTrips(fs, before, data); + + rollingRestartEnablingDataStream(); + + // SCM's node records now expose the RATIS_DATASTREAM port. + waitForAllRegisteredNodesToHaveDatastreamPort(); + + // The legacy pipeline is still portless, so a streaming write keeps + // falling back gracefully and still succeeds. + final Path after = new Path("/after-enable.dat"); + assertEquals(CapableOzoneFSOutputStream.class, + writeAndGetUnderlying(fs, after, data)); + assertRoundTrips(fs, after, data); + } + } + + /** + * A pipeline created while datastream was disabled keeps a portless node + * snapshot and a stale datastream address in its Raft group, so it can + * never serve streaming even after the datanodes restart. SCM closes it so a + * fresh, streaming-capable pipeline is created; a streaming write then + * succeeds over the new pipeline (HDDS-12991). + */ + @Test + @Timeout(value = 70, unit = TimeUnit.SECONDS) + public void testCloseNonStreamablePipelineThenStream() throws Exception { + startClusterWithDatanodeStreamDisabled(); + + try (FileSystem fs = fs()) { + final byte[] data = randomBytes(); + + // Create a portless OPEN pipeline via a fallback write. + final Path p1 = new Path("/legacy.dat"); + assertEquals(CapableOzoneFSOutputStream.class, + writeAndGetUnderlying(fs, p1, data)); + final List before = openRatisThreePipelines(); + assertFalse(before.isEmpty()); + before.forEach(p -> assertFalse(allNodesHaveDatastreamPort(p), + "pipeline should be portless before enabling datastream")); + + rollingRestartEnablingDataStream(); + waitForAllRegisteredNodesToHaveDatastreamPort(); + + // SCM restart reloads the persisted (still portless) pipeline while the + // datanodes are registered with the port (no re-registration event fires). + cluster.restartStorageContainerManager(true); + waitForAllRegisteredNodesToHaveDatastreamPort(); + + final PipelineManager pipelineManager = + cluster.getStorageContainerManager().getPipelineManager(); + final List reloaded = openRatisThreePipelines(); + assertFalse(reloaded.isEmpty()); + reloaded.forEach(p -> assertFalse(allNodesHaveDatastreamPort(p), + "reloaded pipeline should still be portless")); + + // Close the non-streamable pipeline(s); a fresh streaming-capable + // pipeline is created in their place by BackgroundPipelineCreator. + pipelineManager.closeNonStreamablePipelines(); + waitForStreamablePipeline(); + + // The new pipeline actually serves a streaming write end-to-end. + final Path p2 = new Path("/after-recreate.dat"); + assertEquals(CapableOzoneFSDataStreamOutput.class, + writeAndGetUnderlying(fs, p2, data)); + assertRoundTrips(fs, p2, data); + } + } + + /** + * Full lifecycle over a batch of files: write several files while datastream + * is disabled (each must still succeed via the non-streaming fallback), then + * enable datastream (rolling restart + SCM restart + close the non-streamable + * pipeline), then write several more files that must all succeed over a + * streaming-capable pipeline. Asserts that none of the writes fail, the + * post-enablement writes take the streaming path, and an OPEN pipeline + * exposing the RATIS_DATASTREAM port serves them. + * + *

This requires the part-1 client fallback (HDDS-12991 part 1): without it + * the initial streaming-disabled writes fail instead of falling back, so this + * test also demonstrates that part 1 is a prerequisite for part 2. + */ + @Test + @Timeout(value = 120, unit = TimeUnit.SECONDS) + public void testBatchWritesAcrossStreamingEnablement() throws Exception { + startClusterWithDatanodeStreamDisabled(); + + final int fileCount = 5; + try (FileSystem fs = fs()) { + // Phase 1: datastream disabled -> every write falls back, none fail. + for (int i = 0; i < fileCount; i++) { + final byte[] data = randomBytes(); + final Path p = new Path("/disabled-" + i + ".dat"); + assertEquals(CapableOzoneFSOutputStream.class, + writeAndGetUnderlying(fs, p, data), + "write while datastream disabled must fall back to non-streaming"); + assertRoundTrips(fs, p, data); + } + + // Enable datastream on the datanodes and replace the legacy pipeline. + rollingRestartEnablingDataStream(); + waitForAllRegisteredNodesToHaveDatastreamPort(); + cluster.restartStorageContainerManager(true); + waitForAllRegisteredNodesToHaveDatastreamPort(); + cluster.getStorageContainerManager().getPipelineManager() + .closeNonStreamablePipelines(); + waitForStreamablePipeline(); + + // Phase 2: datastream enabled -> every write streams, none fail. + for (int i = 0; i < fileCount; i++) { + final byte[] data = randomBytes(); + final Path p = new Path("/enabled-" + i + ".dat"); + assertEquals(CapableOzoneFSDataStreamOutput.class, + writeAndGetUnderlying(fs, p, data), + "write after enabling datastream must use the streaming path"); + assertRoundTrips(fs, p, data); + } + + // The post-enablement writes are served by a streaming-capable pipeline. + assertTrue(openRatisThreePipelines().stream() + .anyMatch(TestOzoneFileSystemDataStreamEnablement + ::allNodesHaveDatastreamPort), + "an OPEN pipeline should expose the RATIS_DATASTREAM port"); + } + } +}