You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "yandrey321 (via GitHub)" <gi...@apache.org> on 2026/07/22 16:46:36 UTC
[PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
yandrey321 opened a new pull request, #10842:
URL: https://github.com/apache/ozone/pull/10842
## What changes were proposed in this pull request?
SCM refreshes a datanode's ports on re-registration (SCMNodeManager)
1. When a datanode re-registers with a changed port set — e.g. it restarted with DataStream enabled and now advertises the RATIS_DATASTREAM port — SCM updates its stored NodeManager record instead of keeping the stale one. Previously a port-only change (same IP/hostname/version) was ignored, so SCM never learned the new port and every new pipeline stayed port-less.
2. New PipelineManager.closeNonStreamablePipelines(): closes+deletes OPEN pipelines whose persisted node snapshot lacks a port the currently-registered nodes now expose BackgroundPipelineCreator then builds fresh, streaming-capable replacements. Runs automatically as part of the pipeline scrubber: scrubAndCloseNonStreamablePipelines() (scrubs, then closes non-streamable) is wired into the BackgroundPipelineScrubber.
Why close-and-recreate rather than heal in place: a pipeline created before DataStream persists a stale dataStreamAddress (the gRPC port) in its Raft group configuration. ServerState.setRaftConf() only re-adds peers to the gRPC RPC, not the datastream RPC, so the group can never stream in place — the group must be recreated.
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-12991
## How was this patch tested?
TestSCMNodeManager: testRegisterRefreshesPortsOnPortChange (ports changed → node record updated) and testRegisterKeepsPortsWhenUnchanged (unchanged → no update).
TestPipelineManagerImpl: coverage for closeNonStreamablePipelines() — closes a portless OPEN pipeline once nodes expose the new port, and is a no-op when port sets already match.
Integration — TestOzoneFileSystemDataStreamEnablement (MiniOzoneCluster, datanodes start with DataStream disabled, client with streaming enabled):
testDataStreamFallbackAndPortRefresh: writes fall back gracefully while a pipeline is port-less; SCM refreshes the datanodes' ports on re-registration (validates piece 1).
testCloseNonStreamablePipelineThenStream: a portless legacy pipeline survives a rolling restart + SCM restart; closeNonStreamablePipelines() retires it; a fresh streaming-capable pipeline forms and a streaming write succeeds end-to-end (validates piece 2).
testBatchWritesAcrossStreamingEnablement: full lifecycle over a batch of files — several files written while disabled (all succeed via fallback), enable + close legacy pipeline, then several more files that all stream over a RATIS_DATASTREAM-capable pipeline; asserts no write fails and the post-enablement writes take the streaming path.
when part 1 (https://github.com/apache/ozone/pull/10828) is missing:
```
mvn -pl hadoop-ozone/integration-test test \
-Dtest='TestOzoneFileSystemDataStreamEnablement#testBatchWritesAcrossStreamingEnablement' \
-DskipShade -DskipRecon \
-Ddevelocity.cache.local.enabled=false -Ddevelocity.cache.remote.enabled=false -Dscan=false
Tests run: 1, Failures: 0, Errors: 1 <<< FAILURE!
java.io.IOException: RATIS_DATASTREAM port is missing for datanode <dn>
in pipeline <id>; datastream is disabled for this pipeline
at org.apache.hadoop.ozone.client.io.KeyDataStreamOutput.handleWrite(KeyDataStreamOutput.java:212)
...
Caused by: java.io.IOException: RATIS_DATASTREAM port is missing ...
at org.apache.hadoop.hdds.scm.storage.BlockDataStreamOutput.setupStream(BlockDataStreamOutput.java:200)
at org.apache.hadoop.hdds.scm.storage.BlockDataStreamOutput.<init>(BlockDataStreamOutput.java:173)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3669304068
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -562,6 +558,68 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
|| !left.getHostName().equals(right.getHostName()));
}
+ /**
+ * Close (and delete) OPEN pipelines that predate a datanode capability the
+ * registered node now advertises but the pipeline's stored node snapshot
+ * lacks — in practice the RATIS_DATASTREAM port after Ratis DataStream was
+ * enabled. Such pipelines cannot serve streaming even after the datanodes
+ * restart, because the Raft group's persisted configuration still carries the
+ * stale datastream address; only a freshly created pipeline is
+ * streaming-capable. BackgroundPipelineCreator recreates replacements from
+ * the now-capable nodes (HDDS-12991).
+ */
+ void scrubAndCloseNonStreamablePipelines() {
+ try {
+ scrubPipelines();
+ } catch (IOException e) {
+ LOG.error("Unexpected error during pipeline scrubbing", e);
+ }
+ closeNonStreamablePipelines();
+ }
+
+ @Override
+ public void closeNonStreamablePipelines() {
+ for (Pipeline pipeline : getPipelines()) {
+ if (!pipeline.isOpen() || !nodesExposeNewPorts(pipeline)) {
+ continue;
+ }
+ try {
+ final PipelineID id = pipeline.getId();
+ LOG.info("Closing non-streamable pipeline {} so a streaming-capable "
+ + "pipeline can replace it", id);
+ closePipeline(id);
+ deletePipeline(id);
+ } catch (IOException e) {
+ LOG.error("Failed to close non-streamable pipeline {}",
+ pipeline.getId(), e);
+ }
+ }
+ }
+
+ /**
+ * Whether any registered node of the pipeline exposes a port name that the
+ * pipeline's stored copy of that node lacks (e.g. RATIS_DATASTREAM added
+ * after the pipeline was created).
+ */
+ private boolean nodesExposeNewPorts(Pipeline pipeline) {
+ for (DatanodeDetails stored : pipeline.getNodes()) {
+ final DatanodeDetails current = nodeManager.getNode(stored.getID());
+ if (current != null && exposesNewPorts(stored, current)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean exposesNewPorts(DatanodeDetails stored,
+ DatanodeDetails current) {
+ final Set<DatanodeDetails.Port.Name> storedNames = stored.getPorts().stream()
+ .map(DatanodeDetails.Port::getName).collect(Collectors.toSet());
+ return current.getPorts().stream()
+ .map(DatanodeDetails.Port::getName)
+ .anyMatch(name -> !storedNames.contains(name));
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3669302514
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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() {
Review Comment:
Fixed
##########
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();
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "chihsuan (via GitHub)" <gi...@apache.org>.
chihsuan commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3682796369
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java:
##########
@@ -474,6 +474,11 @@ public RegisteredCommand register(
"oldVersion = {}, newVersion = {}.",
datanodeDetails, oldNode.getVersion(), datanodeDetails.getVersion());
nodeStateManager.updateNode(datanodeDetails, layoutInfo);
+ } else if (oldNode.portsChanged(datanodeDetails)) {
+ // Refresh the stored node when its port set changes
+ LOG.info("Updating ports for registered datanode {}: {} -> {}",
+ datanodeDetails, oldNode.getPorts(), datanodeDetails.getPorts());
+ nodeStateManager.updateNode(datanodeDetails, layoutInfo);
Review Comment:
I may be missing some context here, but does calling `updateNode` for a port-only change replace the entire `DatanodeEntry` and reset state such as the container set, storage reports, and health status?
If so, would it be safer to update only the ports while preserving the existing runtime state?
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java:
##########
@@ -0,0 +1,362 @@
+/*
+ * 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;
Review Comment:
nit: PipelineManager is unused now.
```
./hadoop-ozone/dev-support/checks/checkstyle.sh
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java
53: Unused import - org.apache.hadoop.hdds.scm.pipeline.PipelineManager.
```
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java:
##########
@@ -0,0 +1,362 @@
+/*
+ * 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.scm.pipeline.PipelineManagerImpl;
+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):
+ * <ul>
+ * <li>{@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.</li>
+ * <li>{@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.</li>
+ * </ul>
+ */
+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<Pipeline> 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<? extends DatanodeDetails> 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<Pipeline> 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 PipelineManagerImpl pipelineManager =
+ (PipelineManagerImpl) cluster.getStorageContainerManager().getPipelineManager();
+ final List<Pipeline> reloaded = openRatisThreePipelines();
+ assertFalse(reloaded.isEmpty());
+ reloaded.forEach(p -> assertFalse(allNodesHaveDatastreamPort(p),
+ "reloaded pipeline should still be portless"));
+
+ // Close the pipeline(s) exposing the new datastream port; a fresh
+ // streaming-capable pipeline is created in their place by
+ // BackgroundPipelineCreator.
+ pipelineManager.scrubAndClosePipelinesExposingNewPorts();
Review Comment:
SCM still has `DataStream` disabled here, so this call returns without closing the legacy pipelines. Could we enable `DataStream` for SCM before this call and assert that the old pipeline IDs are removed?
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -562,6 +559,66 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
|| !left.getHostName().equals(right.getHostName()));
}
+ /**
+ * Scrub pipelines, then close (and delete) OPEN RATIS pipelines whose
+ * registered nodes now advertise the RATIS_DATASTREAM port their stored node
+ * snapshot lacks.
+ */
+ public void scrubAndClosePipelinesExposingNewPorts() {
+ try {
+ scrubPipelines();
+ } catch (IOException e) {
+ LOG.error("Unexpected error during pipeline scrubbing", e);
+ }
+ closePipelinesExposingNewPorts();
+ }
+
+ void closePipelinesExposingNewPorts() {
+ if (!isDataStreamEnabled()) {
+ return;
+ }
+ for (Pipeline pipeline : getPipelines()) {
Review Comment:
Could one scrubber pass close to all legacy OPEN RATIS pipelines after an upgrade, since they may all satisfy this condition at roughly the same time?
I understand SCM can create replacement pipelines on demand, but could closing them all together cause a burst of retries and pipeline creation?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3677143118
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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,
Review Comment:
Recreating EC pipelines when data streaming is enabled is unwanted side effect of this change, so we need to limit the scope to Ratis only pipelines and use-case when data streaming feature flag is enabled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Recreate pipelines after enabling ratis write streaming [ozone]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on PR #10842:
URL: https://github.com/apache/ozone/pull/10842#issuecomment-5144645590
> ... The comments inlined are for you info. If you are going to update this, please fix them. ...
@yandrey321 , Since you are updating this, could you also address my comments?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3669405775
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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)) {
Review Comment:
Its a valid concern and the question is how the pipelines are going to be re-distributed after restart.
@szetszwo what is your opinion?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3669411304
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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,
Review Comment:
the same concern applies for all port changes. pipeline cannot pick up in-place port configuration changes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on PR #10842:
URL: https://github.com/apache/ozone/pull/10842#issuecomment-5062936763
@szetszwo
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3647238329
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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).
+ */
Review Comment:
This kind of long javadoc repeated multiple times talking about the same thing through out this PR. It will be very maintain it for the future changes. Please review and revise the AI generated javadocs/comments before submitting a PR.
##########
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();
Review Comment:
The implementation is closing all the pipelines with new port but specific to Streaming. Please update the name and javadoc.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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() {
Review Comment:
The implementation is closing all the pipelines with new port but specific to Streaming. Please change the name to reflect it.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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);
Review Comment:
Again, the implementation is closing all the pipelines with new port. The log message is very misleading.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -562,6 +558,68 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
|| !left.getHostName().equals(right.getHostName()));
}
+ /**
+ * Close (and delete) OPEN pipelines that predate a datanode capability the
+ * registered node now advertises but the pipeline's stored node snapshot
+ * lacks — in practice the RATIS_DATASTREAM port after Ratis DataStream was
+ * enabled. Such pipelines cannot serve streaming even after the datanodes
+ * restart, because the Raft group's persisted configuration still carries the
+ * stale datastream address; only a freshly created pipeline is
+ * streaming-capable. BackgroundPipelineCreator recreates replacements from
+ * the now-capable nodes (HDDS-12991).
+ */
+ void scrubAndCloseNonStreamablePipelines() {
+ try {
+ scrubPipelines();
+ } catch (IOException e) {
+ LOG.error("Unexpected error during pipeline scrubbing", e);
+ }
+ closeNonStreamablePipelines();
+ }
+
+ @Override
+ public void closeNonStreamablePipelines() {
+ for (Pipeline pipeline : getPipelines()) {
+ if (!pipeline.isOpen() || !nodesExposeNewPorts(pipeline)) {
+ continue;
+ }
+ try {
+ final PipelineID id = pipeline.getId();
+ LOG.info("Closing non-streamable pipeline {} so a streaming-capable "
+ + "pipeline can replace it", id);
+ closePipeline(id);
+ deletePipeline(id);
+ } catch (IOException e) {
+ LOG.error("Failed to close non-streamable pipeline {}",
+ pipeline.getId(), e);
+ }
+ }
+ }
+
+ /**
+ * Whether any registered node of the pipeline exposes a port name that the
+ * pipeline's stored copy of that node lacks (e.g. RATIS_DATASTREAM added
+ * after the pipeline was created).
+ */
+ private boolean nodesExposeNewPorts(Pipeline pipeline) {
+ for (DatanodeDetails stored : pipeline.getNodes()) {
+ final DatanodeDetails current = nodeManager.getNode(stored.getID());
+ if (current != null && exposesNewPorts(stored, current)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean exposesNewPorts(DatanodeDetails stored,
+ DatanodeDetails current) {
+ final Set<DatanodeDetails.Port.Name> storedNames = stored.getPorts().stream()
+ .map(DatanodeDetails.Port::getName).collect(Collectors.toSet());
+ return current.getPorts().stream()
+ .map(DatanodeDetails.Port::getName)
+ .anyMatch(name -> !storedNames.contains(name));
Review Comment:
Move it to DatanodeDetails and make it non-static.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3685176931
##########
hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java:
##########
@@ -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<String> portValues(DatanodeDetails datanodeDetails) {
+ return datanodeDetails.getPorts().stream()
+ .map(port -> port.getName() + "=" + port.getValue())
+ .collect(Collectors.toSet());
Review Comment:
This method is inefficient. Below are not needed:
- getPorts() copies the ports list.
- Create a Set of Strings.
However, the ports list itself is inefficient. Let's improve it in HDDS-16043.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -182,13 +183,9 @@ public static PipelineManagerImpl newPipelineManager(
.setServiceName("BackgroundPipelineScrubber")
.setIntervalInMillis(scrubberIntervalInMillis)
.setWaitTimeInMillis(safeModeWaitMs)
- .setPeriodicalTask(() -> {
- try {
- pipelineManager.scrubPipelines();
- } catch (IOException e) {
- LOG.error("Unexpected error during pipeline scrubbing", e);
- }
- }).build();
+ .setPeriodicalTask(
+ pipelineManager::scrubAndClosePipelinesExposingNewPorts)
Review Comment:
Use a single line.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -562,6 +559,66 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
|| !left.getHostName().equals(right.getHostName()));
}
+ /**
+ * Scrub pipelines, then close (and delete) OPEN RATIS pipelines whose
+ * registered nodes now advertise the RATIS_DATASTREAM port their stored node
+ * snapshot lacks.
+ */
+ public void scrubAndClosePipelinesExposingNewPorts() {
+ try {
+ scrubPipelines();
+ } catch (IOException e) {
+ LOG.error("Unexpected error during pipeline scrubbing", e);
+ }
+ closePipelinesExposingNewPorts();
+ }
+
+ void closePipelinesExposingNewPorts() {
Review Comment:
The code is checking only RATIS_DATASTREAM port. It is better call it closePipelinesMissingDataStreamPort, which is consistent with nodesMissingDataStreamPort.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -562,6 +559,66 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
|| !left.getHostName().equals(right.getHostName()));
}
+ /**
+ * Scrub pipelines, then close (and delete) OPEN RATIS pipelines whose
+ * registered nodes now advertise the RATIS_DATASTREAM port their stored node
+ * snapshot lacks.
+ */
+ public void scrubAndClosePipelinesExposingNewPorts() {
+ try {
+ scrubPipelines();
+ } catch (IOException e) {
+ LOG.error("Unexpected error during pipeline scrubbing", e);
+ }
+ closePipelinesExposingNewPorts();
+ }
+
+ void closePipelinesExposingNewPorts() {
+ if (!isDataStreamEnabled()) {
+ return;
+ }
+ for (Pipeline pipeline : getPipelines()) {
+ if (!pipeline.isOpen()
+ || pipeline.getType() != ReplicationType.RATIS
+ || !nodesMissingDataStreamPort(pipeline)) {
+ continue;
+ }
+ try {
+ final PipelineID id = pipeline.getId();
+ LOG.info("Closing RATIS pipeline {} whose nodes now advertise the "
+ + "datastream port so a datastream-capable pipeline can replace it",
+ id);
Review Comment:
Log messages should be shorter, simpler, and concise; see below.
<img width="562" height="369" alt="Image" src="https://github.com/user-attachments/assets/8d53c239-1492-47ba-a929-e1fa2a1cadb1" />
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java:
##########
@@ -474,6 +474,11 @@ public RegisteredCommand register(
"oldVersion = {}, newVersion = {}.",
datanodeDetails, oldNode.getVersion(), datanodeDetails.getVersion());
nodeStateManager.updateNode(datanodeDetails, layoutInfo);
+ } else if (oldNode.portsChanged(datanodeDetails)) {
+ // Refresh the stored node when its port set changes
+ LOG.info("Updating ports for registered datanode {}: {} -> {}",
+ datanodeDetails, oldNode.getPorts(), datanodeDetails.getPorts());
+ nodeStateManager.updateNode(datanodeDetails, layoutInfo);
Review Comment:
@chihsuan , the port changed case should be handled similar to the version changed case. It seems good to refresh everything.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -562,6 +559,66 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
|| !left.getHostName().equals(right.getHostName()));
}
+ /**
+ * Scrub pipelines, then close (and delete) OPEN RATIS pipelines whose
+ * registered nodes now advertise the RATIS_DATASTREAM port their stored node
+ * snapshot lacks.
+ */
+ public void scrubAndClosePipelinesExposingNewPorts() {
+ try {
+ scrubPipelines();
+ } catch (IOException e) {
+ LOG.error("Unexpected error during pipeline scrubbing", e);
+ }
+ closePipelinesExposingNewPorts();
+ }
+
+ void closePipelinesExposingNewPorts() {
+ if (!isDataStreamEnabled()) {
+ return;
+ }
+ for (Pipeline pipeline : getPipelines()) {
Review Comment:
This is a one time thing happened when changing DataStream from disabled to enabled. For simplicity, let's just close all the existing pipelines and recreate new pipelines.
In practice, the maintenance window will be longer. So, it seems fine.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on PR #10842:
URL: https://github.com/apache/ozone/pull/10842#issuecomment-5134725412
@chihsuan , please see if you have further comments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3685485785
##########
hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java:
##########
@@ -0,0 +1,362 @@
+/*
+ * 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;
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3669308233
##########
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):
+ * <ul>
+ * <li>{@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.</li>
+ * <li>{@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.</li>
+ * </ul>
+ */
+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<Pipeline> 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<? extends DatanodeDetails> 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)
Review Comment:
its an attempt to have a failsafe for slow CI environments.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3684650769
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java:
##########
@@ -474,6 +474,11 @@ public RegisteredCommand register(
"oldVersion = {}, newVersion = {}.",
datanodeDetails, oldNode.getVersion(), datanodeDetails.getVersion());
nodeStateManager.updateNode(datanodeDetails, layoutInfo);
+ } else if (oldNode.portsChanged(datanodeDetails)) {
+ // Refresh the stored node when its port set changes
+ LOG.info("Updating ports for registered datanode {}: {} -> {}",
+ datanodeDetails, oldNode.getPorts(), datanodeDetails.getPorts());
+ nodeStateManager.updateNode(datanodeDetails, layoutInfo);
Review Comment:
Is there a reason why we dont want to update stale properties here and update ports only?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on PR #10842:
URL: https://github.com/apache/ozone/pull/10842#issuecomment-5137767399
CI passed in my private branch: https://github.com/yandrey321/ozone/actions/runs/30587999129
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3677155545
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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)) {
Review Comment:
In order to address this concern we need to change a way how DNs are selected for a new pipeline creation. There is a statistical chance that new pipeline would pick up a node where streaming port is not enabled yet, so it would be recreated again.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on PR #10842:
URL: https://github.com/apache/ozone/pull/10842#issuecomment-5137741290
@szetszwo could you please start CI test cycle for this PR and merge it when CI gets green status?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3685493472
##########
hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/protocol/DatanodeDetails.java:
##########
@@ -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<String> portValues(DatanodeDetails datanodeDetails) {
+ return datanodeDetails.getPorts().stream()
+ .map(port -> port.getName() + "=" + port.getValue())
+ .collect(Collectors.toSet());
Review Comment:
I'll fix it as part of [HDDS-16043](https://issues.apache.org/jira/browse/HDDS-16043)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on PR #10842:
URL: https://github.com/apache/ozone/pull/10842#issuecomment-5144934296
> > ... The comments inlined are for you info. If you are going to update this, please fix them. ...
>
> @yandrey321 , Since you are updating this, could you also address my comments?
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3669300882
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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).
+ */
Review Comment:
fixed.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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);
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3668161646
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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,
Review Comment:
As far as I understand its a generic issue with ports update, after port updates the pipeline cannot be updated in-place so we have to recreate it in order to pick up configuration change.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3676056055
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java:
##########
@@ -117,6 +117,14 @@ void addContainerToPipeline(PipelineID pipelineID, ContainerID containerID)
void closeStalePipelines(DatanodeDetails datanodeDetails);
+ /**
+ * Close OPEN pipelines whose datanodes now expose a port name that the
+ * pipeline's stored node snapshot lacks, so fresh pipelines advertising the
+ * new port are created in their place. A pipeline's cannot pick up
+ * a newly advertised port in place, so it must be recreated.
+ */
+ void closePipelinesExposingNewPorts();
Review Comment:
Since this method only called internally in PipelineManagerImpl, let's keep it private/package private and don't add it to the interface.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java:
##########
@@ -487,6 +492,22 @@ public RegisteredCommand register(
.build();
}
+ /**
+ * Whether the datanode's exposed ports changed between two registrations.
+ * Compared as a set of name=value entries, since
+ * {@link DatanodeDetails.Port#equals} ignores the port value.
+ */
+ private static boolean portsChanged(DatanodeDetails oldNode,
Review Comment:
portsChanged and exposesNewPorts are two very similar methods. Let's keep only one and consistently using it.
- Since portsChanged is more general. Let's keep it, move it to DatanodeDetails and make it non-static.
- Remove exposesNewPorts since it is harder to use (non-commutative: a.exposesNewPorts(b) and b.exposesNewPorts(a) could be different.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Recreate pipelines after enabling ratis write streaming [ozone]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo merged PR #10842:
URL: https://github.com/apache/ozone/pull/10842
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991 recreate pipelines after enabling ratis write streaming [ozone]
Posted by "chihsuan (via GitHub)" <gi...@apache.org>.
chihsuan commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3644758438
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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,
Review Comment:
I wonder if we could narrow this check to RATIS pipelines and the `RATIS_DATASTREAM` port specifically. Older pipeline snapshots may legitimately lack ports added by later versions, such as HTTP, HTTPS, or CLIENT_RPC. In that case, checking any new port could retire unrelated pipelines, including EC pipelines.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -562,6 +558,68 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail
|| !left.getHostName().equals(right.getHostName()));
}
+ /**
+ * Close (and delete) OPEN pipelines that predate a datanode capability the
+ * registered node now advertises but the pipeline's stored node snapshot
+ * lacks — in practice the RATIS_DATASTREAM port after Ratis DataStream was
+ * enabled. Such pipelines cannot serve streaming even after the datanodes
+ * restart, because the Raft group's persisted configuration still carries the
+ * stale datastream address; only a freshly created pipeline is
+ * streaming-capable. BackgroundPipelineCreator recreates replacements from
+ * the now-capable nodes (HDDS-12991).
+ */
+ void scrubAndCloseNonStreamablePipelines() {
+ try {
+ scrubPipelines();
+ } catch (IOException e) {
+ LOG.error("Unexpected error during pipeline scrubbing", e);
+ }
+ closeNonStreamablePipelines();
+ }
+
+ @Override
+ public void closeNonStreamablePipelines() {
+ for (Pipeline pipeline : getPipelines()) {
+ if (!pipeline.isOpen() || !nodesExposeNewPorts(pipeline)) {
+ continue;
+ }
+ try {
+ final PipelineID id = pipeline.getId();
+ LOG.info("Closing non-streamable pipeline {} so a streaming-capable "
+ + "pipeline can replace it", id);
+ closePipeline(id);
+ deletePipeline(id);
+ } catch (IOException e) {
+ LOG.error("Failed to close non-streamable pipeline {}",
+ pipeline.getId(), e);
+ }
+ }
+ }
+
+ /**
+ * Whether any registered node of the pipeline exposes a port name that the
+ * pipeline's stored copy of that node lacks (e.g. RATIS_DATASTREAM added
+ * after the pipeline was created).
+ */
+ private boolean nodesExposeNewPorts(Pipeline pipeline) {
+ for (DatanodeDetails stored : pipeline.getNodes()) {
+ final DatanodeDetails current = nodeManager.getNode(stored.getID());
+ if (current != null && exposesNewPorts(stored, current)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static boolean exposesNewPorts(DatanodeDetails stored,
+ DatanodeDetails current) {
+ final Set<DatanodeDetails.Port.Name> storedNames = stored.getPorts().stream()
+ .map(DatanodeDetails.Port::getName).collect(Collectors.toSet());
+ return current.getPorts().stream()
+ .map(DatanodeDetails.Port::getName)
+ .anyMatch(name -> !storedNames.contains(name));
Review Comment:
`SCMNodeManager.portsChanged` handles `name=value` changes, while this check only compares names. Should we also compare the `RATIS_DATASTREAM` port value here?
##########
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):
+ * <ul>
+ * <li>{@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.</li>
+ * <li>{@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.</li>
+ * </ul>
+ */
+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<Pipeline> 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<? extends DatanodeDetails> 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)
Review Comment:
Would it make sense to rely on the repository’s five-minute default timeout here? This test waits up to 30 seconds for port registration both before and after the SCM restart, then up to 60 seconds for a replacement pipeline.
With cluster startup and restarts as well, the 70-second outer timeout may fire first and hide which operation was actually slow.
The same concern also applies to lines 225 and 318.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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)) {
Review Comment:
One concern during a rolling restart: this may close the pipeline as soon as the first member advertises the new port. A replacement could still include datanodes that have not restarted yet, leading to another close-and-recreate cycle later. Would it make sense if we defer retirement until SCM can ensure the replacement uses only DataStream-capable nodes?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org