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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
