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