From 917170c6be95fc42e5e0213dfb7beb389cf2eb13 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 22 Jul 2026 10:47:40 -0400 Subject: [PATCH 1/7] 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"); + } + } +} From c149cea3f7450b0e6d98fb95bd0bac3a843f09e7 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Tue, 28 Jul 2026 14:01:37 -0400 Subject: [PATCH 2/7] Addressed review comments --- .../hadoop/hdds/protocol/DatanodeDetails.java | 13 +++++++ .../hdds/scm/pipeline/PipelineManager.java | 11 +++--- .../scm/pipeline/PipelineManagerImpl.java | 35 ++++++------------- .../scm/pipeline/MockPipelineManager.java | 2 +- .../scm/pipeline/TestPipelineManagerImpl.java | 20 +++++------ ...stOzoneFileSystemDataStreamEnablement.java | 9 ++--- 6 files changed, 44 insertions(+), 46 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java index abd00e706837..dbff786a7278 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java +++ b/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. * 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 fa98650e6522..8b41d63fd903 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 @@ -118,13 +118,12 @@ 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). + * 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 closeNonStreamablePipelines(); + void closePipelinesExposingNewPorts(); void scrubPipelines() throws IOException; 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 42a0ea4221f6..e084034d824a 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 @@ -183,7 +183,7 @@ public static PipelineManagerImpl newPipelineManager( .setIntervalInMillis(scrubberIntervalInMillis) .setWaitTimeInMillis(safeModeWaitMs) .setPeriodicalTask( - pipelineManager::scrubAndCloseNonStreamablePipelines) + pipelineManager::scrubAndClosePipelinesExposingNewPorts) .build(); pipelineManager.setBackgroundPipelineScrubber(backgroundPipelineScrubber); @@ -559,38 +559,32 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail } /** - * 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). + * Scrub pipelines, then close (and delete) OPEN pipelines whose registered + * nodes now expose a port name the pipeline's stored node snapshot lacks. */ - void scrubAndCloseNonStreamablePipelines() { + void scrubAndClosePipelinesExposingNewPorts() { try { scrubPipelines(); } catch (IOException e) { LOG.error("Unexpected error during pipeline scrubbing", e); } - closeNonStreamablePipelines(); + closePipelinesExposingNewPorts(); } @Override - public void closeNonStreamablePipelines() { + public void closePipelinesExposingNewPorts() { 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); + 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 non-streamable pipeline {}", + LOG.error("Failed to close pipeline {} exposing a new port", pipeline.getId(), e); } } @@ -604,22 +598,13 @@ public void closeNonStreamablePipelines() { private boolean nodesExposeNewPorts(Pipeline pipeline) { for (DatanodeDetails stored : pipeline.getNodes()) { final DatanodeDetails current = nodeManager.getNode(stored.getID()); - if (current != null && exposesNewPorts(stored, current)) { + if (current != null && current.exposesNewPorts(stored)) { 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/pipeline/MockPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java index 473fd0b0ea27..af2da1c6de97 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 @@ -250,7 +250,7 @@ public void closeStalePipelines(DatanodeDetails datanodeDetails) { } @Override - public void closeNonStreamablePipelines() { + public void closePipelinesExposingNewPorts() { } 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 b9bc62a84434..24a7fa5c67d9 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 @@ -1021,7 +1021,7 @@ private static boolean exists(PipelineManagerImpl pipelineManager, } @Test - public void testCloseNonStreamablePipelines() throws Exception { + public void testClosePipelinesExposingNewPorts() throws Exception { try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { // Registered datanodes (MockNodeManager) expose all ports incl datastream. final List registered = nodeManager.getAllNodes(); @@ -1046,10 +1046,10 @@ public void testCloseNonStreamablePipelines() throws Exception { // ALLOCATED (non-open) portless -> skipped. final Pipeline allocated = addPipeline(pipelineManager, ALLOCATED, idsB); - pipelineManager.closeNonStreamablePipelines(); + pipelineManager.closePipelinesExposingNewPorts(); assertFalse(exists(pipelineManager, stale.getId()), - "legacy portless OPEN pipeline should be closed and deleted"); + "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())); @@ -1057,7 +1057,7 @@ public void testCloseNonStreamablePipelines() throws Exception { } @Test - public void testCloseNonStreamablePipelinesSwallowsError() throws Exception { + public void testClosePipelinesExposingNewPortsSwallowsError() throws Exception { try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { final List registered = nodeManager.getAllNodes(); final List nodes = new ArrayList<>(); @@ -1069,17 +1069,17 @@ public void testCloseNonStreamablePipelinesSwallowsError() throws Exception { 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(); + spy.closePipelinesExposingNewPorts(); 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. + // 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.scrubAndCloseNonStreamablePipelines(); + pipelineManager.scrubAndClosePipelinesExposingNewPorts(); } } @@ -1089,8 +1089,8 @@ public void testScrubAndCloseSwallowsScrubError() throws Exception { 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(); + spy.scrubAndClosePipelinesExposingNewPorts(); + verify(spy).closePipelinesExposingNewPorts(); } } } 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 index 30bd04597b42..2e835f11b71d 100644 --- 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 @@ -288,9 +288,10 @@ public void testCloseNonStreamablePipelineThenStream() throws Exception { 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(); + // Close the pipeline(s) exposing the new datastream port; a fresh + // streaming-capable pipeline is created in their place by + // BackgroundPipelineCreator. + pipelineManager.closePipelinesExposingNewPorts(); waitForStreamablePipeline(); // The new pipeline actually serves a streaming write end-to-end. @@ -337,7 +338,7 @@ public void testBatchWritesAcrossStreamingEnablement() throws Exception { cluster.restartStorageContainerManager(true); waitForAllRegisteredNodesToHaveDatastreamPort(); cluster.getStorageContainerManager().getPipelineManager() - .closeNonStreamablePipelines(); + .closePipelinesExposingNewPorts(); waitForStreamablePipeline(); // Phase 2: datastream enabled -> every write streams, none fail. From 2ba6beaedebcebb9fd65e945c008f2bf45d4d454 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Tue, 28 Jul 2026 14:16:32 -0400 Subject: [PATCH 3/7] cleaned up comments --- .../java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java | 5 +---- .../apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java | 3 +-- 2 files changed, 2 insertions(+), 6 deletions(-) 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 97bc9f5705ce..b11ab003850e 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 @@ -475,10 +475,7 @@ public RegisteredCommand register( 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). + // 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); 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 e084034d824a..3b25b904d8a8 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 @@ -592,8 +592,7 @@ public void closePipelinesExposingNewPorts() { /** * 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). + * pipeline's stored copy of that node lacks */ private boolean nodesExposeNewPorts(Pipeline pipeline) { for (DatanodeDetails stored : pipeline.getNodes()) { From b00f7b8e5e9225968222f68f1e69a63a6b4b2929 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Tue, 28 Jul 2026 14:24:45 -0400 Subject: [PATCH 4/7] fixed checkstyle --- .../java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java | 2 +- .../apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java index dbff786a7278..c7e38be2850b 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java @@ -385,7 +385,7 @@ public synchronized boolean hasPort(Port.Name name) { } /** - * Whether this datanode exposes a port name that {@code other} lacks + * 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 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 3b25b904d8a8..706b2f3492de 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 @@ -592,7 +592,7 @@ public void closePipelinesExposingNewPorts() { /** * Whether any registered node of the pipeline exposes a port name that the - * pipeline's stored copy of that node lacks + * pipeline's stored copy of that node lacks. */ private boolean nodesExposeNewPorts(Pipeline pipeline) { for (DatanodeDetails stored : pipeline.getNodes()) { From d7de70dbe98f0ece90e95cb4318438870063b706 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 29 Jul 2026 14:22:32 -0400 Subject: [PATCH 5/7] limit the scrubbing to ratis pipelines when data streaming is enabled and DN exposes the data streaming port --- .../hadoop/hdds/protocol/DatanodeDetails.java | 13 ---- .../hdds/scm/pipeline/PipelineManager.java | 9 ++- .../scm/pipeline/PipelineManagerImpl.java | 38 +++++++--- .../scm/pipeline/TestPipelineManagerImpl.java | 73 +++++++++++++++++++ 4 files changed, 105 insertions(+), 28 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java index c7e38be2850b..abd00e706837 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java @@ -384,19 +384,6 @@ 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. * 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 8b41d63fd903..bac08a136f12 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 @@ -118,10 +118,11 @@ 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. + * When datastream is enabled, close OPEN RATIS pipelines whose datanodes now + * advertise the RATIS_DATASTREAM port that the pipeline's stored node + * snapshot lacks, so datastream-capable pipelines are created in their place. + * A pipeline cannot pick up a newly advertised port in place, so it must be + * recreated. */ void closePipelinesExposingNewPorts(); 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 706b2f3492de..d55ad7256998 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 @@ -63,6 +63,7 @@ import org.apache.hadoop.hdds.utils.db.Table; import org.apache.hadoop.metrics2.util.MBeans; import org.apache.hadoop.ozone.ClientVersion; +import org.apache.hadoop.ozone.OzoneConfigKeys; import org.apache.hadoop.util.Time; import org.apache.ratis.protocol.exceptions.NotLeaderException; import org.slf4j.Logger; @@ -559,8 +560,9 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail } /** - * Scrub pipelines, then close (and delete) OPEN pipelines whose registered - * nodes now expose a port name the pipeline's stored node snapshot lacks. + * Scrub pipelines, then close (and delete) OPEN RATIS pipelines whose + * registered nodes now advertise the RATIS_DATASTREAM port their stored node + * snapshot lacks. */ void scrubAndClosePipelinesExposingNewPorts() { try { @@ -573,31 +575,45 @@ void scrubAndClosePipelinesExposingNewPorts() { @Override public void closePipelinesExposingNewPorts() { + if (!isDataStreamEnabled()) { + return; + } for (Pipeline pipeline : getPipelines()) { - if (!pipeline.isOpen() || !nodesExposeNewPorts(pipeline)) { + if (!pipeline.isOpen() + || pipeline.getType() != ReplicationType.RATIS + || !nodesMissingDataStreamPort(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); + LOG.info("Closing RATIS pipeline {} whose nodes now advertise the " + + "datastream port so a datastream-capable pipeline can replace it", + id); closePipeline(id); deletePipeline(id); } catch (IOException e) { - LOG.error("Failed to close pipeline {} exposing a new port", - pipeline.getId(), e); + LOG.error("Failed to close RATIS pipeline {} missing the datastream " + + "port", pipeline.getId(), e); } } } + private boolean isDataStreamEnabled() { + return conf.getBoolean( + OzoneConfigKeys.HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, + OzoneConfigKeys.HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED_DEFAULT); + } + /** - * Whether any registered node of the pipeline exposes a port name that the - * pipeline's stored copy of that node lacks. + * Whether any registered node of the pipeline now advertises the + * RATIS_DATASTREAM port that the pipeline's stored node snapshot lacks. */ - private boolean nodesExposeNewPorts(Pipeline pipeline) { + private boolean nodesMissingDataStreamPort(Pipeline pipeline) { for (DatanodeDetails stored : pipeline.getNodes()) { final DatanodeDetails current = nodeManager.getNode(stored.getID()); - if (current != null && current.exposesNewPorts(stored)) { + if (current != null + && current.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM) + && !stored.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM)) { return true; } } 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 24a7fa5c67d9..cee7b85fe766 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 @@ -23,6 +23,7 @@ import static org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_PIPELINE_DESTROY_TIMEOUT; import static org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState.ALLOCATED; import static org.apache.hadoop.hdds.scm.pipeline.Pipeline.PipelineState.OPEN; +import static org.apache.hadoop.ozone.OzoneConfigKeys.HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED; import static org.apache.ozone.test.MetricsAsserts.getLongCounter; import static org.apache.ozone.test.MetricsAsserts.getMetrics; import static org.apache.ratis.util.Preconditions.assertInstanceOf; @@ -1022,6 +1023,7 @@ private static boolean exists(PipelineManagerImpl pipelineManager, @Test public void testClosePipelinesExposingNewPorts() throws Exception { + conf.setBoolean(HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, true); try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { // Registered datanodes (MockNodeManager) expose all ports incl datastream. final List registered = nodeManager.getAllNodes(); @@ -1056,8 +1058,79 @@ public void testClosePipelinesExposingNewPorts() throws Exception { } } + @Test + public void testClosePipelinesExposingNewPortsSkippedWhenDataStreamDisabled() + throws Exception { + // Datastream disabled (default): even a portless RATIS pipeline is kept. + 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 portless = addPipeline(pipelineManager, OPEN, nodes); + + pipelineManager.closePipelinesExposingNewPorts(); + + assertTrue(exists(pipelineManager, portless.getId()), + "portless pipeline must be kept while datastream is disabled"); + } + } + + @Test + public void testClosePipelinesExposingNewPortsSkipsEcPipeline() + throws Exception { + conf.setBoolean(HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, true); + try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { + final List registered = nodeManager.getAllNodes(); + final List nodes = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + nodes.add(portlessDatanode(registered.get(i).getID())); + } + final Pipeline ec = Pipeline.newBuilder() + .setReplicationConfig(new ECReplicationConfig(3, 2)) + .setNodes(nodes) + .setState(OPEN) + .setId(PipelineID.randomId()) + .build(); + pipelineManager.getStateManager().addPipeline( + ec.getProtobufMessage(ClientVersion.CURRENT_VERSION)); + + pipelineManager.closePipelinesExposingNewPorts(); + + assertTrue(exists(pipelineManager, ec.getId()), + "EC pipeline must not be closed by datastream port scrubbing"); + } + } + + @Test + public void testClosePipelinesExposingNewPortsKeepsNotYetRestartedNodes() + throws Exception { + conf.setBoolean(HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, true); + try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { + // Nodes are registered and healthy but still lack the datastream port + // (they have not restarted yet during a rolling enablement). + final List nodes = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + final DatanodeDetails portless = portlessDatanode(DatanodeID.randomID()); + nodeManager.register(new DatanodeInfo(portless, + NodeStatus.inServiceHealthy(), null, + HddsTestUtils.ROLL_INTERVAL_MS_DEFAULT), null, null); + nodes.add(portless); + } + final Pipeline pending = addPipeline(pipelineManager, OPEN, nodes); + + pipelineManager.closePipelinesExposingNewPorts(); + + assertTrue(exists(pipelineManager, pending.getId()), + "pipeline whose registered nodes have not yet advertised the " + + "datastream port must be kept"); + } + } + @Test public void testClosePipelinesExposingNewPortsSwallowsError() throws Exception { + conf.setBoolean(HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, true); try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { final List registered = nodeManager.getAllNodes(); final List nodes = new ArrayList<>(); From 34e9db616aec6a9f3d906af9154a47b438145a22 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 29 Jul 2026 14:36:21 -0400 Subject: [PATCH 6/7] fixed review comments --- .../hadoop/hdds/protocol/DatanodeDetails.java | 21 ++++++++++++++++++- .../hadoop/hdds/scm/node/SCMNodeManager.java | 18 +--------------- 2 files changed, 21 insertions(+), 18 deletions(-) diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java index abd00e706837..4abe44a3040b 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java @@ -35,6 +35,7 @@ import java.util.Objects; import java.util.Set; import java.util.UUID; +import java.util.stream.Collectors; import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.hdds.DatanodeVersion; import org.apache.hadoop.hdds.HddsUtils; @@ -384,9 +385,27 @@ public synchronized boolean hasPort(Port.Name name) { return false; } + /** + * Whether this datanode's exposed ports differ from {@code other}'s. + * Compared as a set of name=value entries, since {@link Port#equals} + * ignores the port value. + * + * @param other another snapshot of this datanode + * @return true if the two port sets are not identical + */ + public boolean portsChanged(DatanodeDetails other) { + return !portValues(this).equals(portValues(other)); + } + + private static Set portValues(DatanodeDetails datanodeDetails) { + return datanodeDetails.getPorts().stream() + .map(port -> port.getName() + "=" + port.getValue()) + .collect(Collectors.toSet()); + } + /** * Helper method to get the Ratis port. - * + * * @return Port */ public Port getRatisPort() { 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 b11ab003850e..fe886bf0c20e 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,7 +474,7 @@ public RegisteredCommand register( "oldVersion = {}, newVersion = {}.", datanodeDetails, oldNode.getVersion(), datanodeDetails.getVersion()); nodeStateManager.updateNode(datanodeDetails, layoutInfo); - } else if (portsChanged(oldNode, datanodeDetails)) { + } else if (oldNode.portsChanged(datanodeDetails)) { // Refresh the stored node when its port set changes LOG.info("Updating ports for registered datanode {}: {} -> {}", datanodeDetails, oldNode.getPorts(), datanodeDetails.getPorts()); @@ -492,22 +492,6 @@ 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, From 400b1df340cf6d08e9b95d8941bd0372bf35d429 Mon Sep 17 00:00:00 2001 From: Andrey Yarovoy Date: Wed, 29 Jul 2026 15:46:36 -0400 Subject: [PATCH 7/7] fixed review comments --- .../hadoop/hdds/scm/pipeline/PipelineManager.java | 9 --------- .../hadoop/hdds/scm/pipeline/PipelineManagerImpl.java | 5 ++--- .../hadoop/hdds/scm/pipeline/MockPipelineManager.java | 5 ----- .../TestOzoneFileSystemDataStreamEnablement.java | 11 ++++++----- 4 files changed, 8 insertions(+), 22 deletions(-) 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 bac08a136f12..e0d9de1f10b5 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,15 +117,6 @@ void addContainerToPipeline(PipelineID pipelineID, ContainerID containerID) void closeStalePipelines(DatanodeDetails datanodeDetails); - /** - * When datastream is enabled, close OPEN RATIS pipelines whose datanodes now - * advertise the RATIS_DATASTREAM port that the pipeline's stored node - * snapshot lacks, so datastream-capable pipelines are created in their place. - * A pipeline cannot pick up a newly advertised port in place, so it must be - * recreated. - */ - void closePipelinesExposingNewPorts(); - 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 d55ad7256998..2b30a1a2f667 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 @@ -564,7 +564,7 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail * registered nodes now advertise the RATIS_DATASTREAM port their stored node * snapshot lacks. */ - void scrubAndClosePipelinesExposingNewPorts() { + public void scrubAndClosePipelinesExposingNewPorts() { try { scrubPipelines(); } catch (IOException e) { @@ -573,8 +573,7 @@ void scrubAndClosePipelinesExposingNewPorts() { closePipelinesExposingNewPorts(); } - @Override - public void closePipelinesExposingNewPorts() { + void closePipelinesExposingNewPorts() { if (!isDataStreamEnabled()) { return; } 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 af2da1c6de97..ef6f187b040c 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,11 +249,6 @@ public void closeStalePipelines(DatanodeDetails datanodeDetails) { } - @Override - public void closePipelinesExposingNewPorts() { - - } - @Override public void scrubPipelines() { 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 index 2e835f11b71d..7d6c5289e126 100644 --- 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 @@ -51,6 +51,7 @@ 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.scm.pipeline.PipelineManagerImpl; import org.apache.hadoop.hdds.utils.IOUtils; import org.apache.hadoop.ozone.ClientConfigForTesting; import org.apache.hadoop.ozone.MiniOzoneCluster; @@ -281,8 +282,8 @@ public void testCloseNonStreamablePipelineThenStream() throws Exception { cluster.restartStorageContainerManager(true); waitForAllRegisteredNodesToHaveDatastreamPort(); - final PipelineManager pipelineManager = - cluster.getStorageContainerManager().getPipelineManager(); + final PipelineManagerImpl pipelineManager = + (PipelineManagerImpl) cluster.getStorageContainerManager().getPipelineManager(); final List reloaded = openRatisThreePipelines(); assertFalse(reloaded.isEmpty()); reloaded.forEach(p -> assertFalse(allNodesHaveDatastreamPort(p), @@ -291,7 +292,7 @@ public void testCloseNonStreamablePipelineThenStream() throws Exception { // Close the pipeline(s) exposing the new datastream port; a fresh // streaming-capable pipeline is created in their place by // BackgroundPipelineCreator. - pipelineManager.closePipelinesExposingNewPorts(); + pipelineManager.scrubAndClosePipelinesExposingNewPorts(); waitForStreamablePipeline(); // The new pipeline actually serves a streaming write end-to-end. @@ -337,8 +338,8 @@ public void testBatchWritesAcrossStreamingEnablement() throws Exception { waitForAllRegisteredNodesToHaveDatastreamPort(); cluster.restartStorageContainerManager(true); waitForAllRegisteredNodesToHaveDatastreamPort(); - cluster.getStorageContainerManager().getPipelineManager() - .closePipelinesExposingNewPorts(); + ((PipelineManagerImpl) cluster.getStorageContainerManager().getPipelineManager()) + .scrubAndClosePipelinesExposingNewPorts(); waitForStreamablePipeline(); // Phase 2: datastream enabled -> every write streams, none fail.