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: [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]
