diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java index 1ababc7a1d76..fa3cf9b2b3a7 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java @@ -196,7 +196,8 @@ public BlockDataStreamOutput( private DataStreamOutput setupStream(Pipeline pipeline) throws IOException { for (DatanodeDetails dn : pipeline.getNodes()) { if (!dn.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM)) { - throw new IOException("RATIS_DATASTREAM port is missing for datanode " + throw new StreamNotSupportedException( + "RATIS_DATASTREAM port is missing for datanode " + dn + " in pipeline " + pipeline.getId() + "; datastream is disabled for this pipeline"); } diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamNotSupportedException.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamNotSupportedException.java new file mode 100644 index 000000000000..1f94b4e22bdf --- /dev/null +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/StreamNotSupportedException.java @@ -0,0 +1,35 @@ +/* + * 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.hdds.scm.storage; + +import java.io.IOException; + +/** + * Thrown when a Ratis DataStream write cannot proceed because the target + * pipeline does not support streaming (e.g. its datanodes were created before + * DataStream was enabled and therefore lack the RATIS_DATASTREAM port). + * + *

Callers may catch this to fall back to the non-streaming write path + * instead of failing the write (HDDS-12991). + */ +public class StreamNotSupportedException extends IOException { + + public StreamNotSupportedException(String message) { + super(message); + } +} diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java index cbcfc5537913..97bc9f5705ce 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java @@ -474,6 +474,14 @@ public RegisteredCommand register( "oldVersion = {}, newVersion = {}.", datanodeDetails, oldNode.getVersion(), datanodeDetails.getVersion()); nodeStateManager.updateNode(datanodeDetails, layoutInfo); + } else if (portsChanged(oldNode, datanodeDetails)) { + // Refresh the stored node when its port set changes (e.g. a datanode + // restarts with Ratis DataStream enabled and now exposes the + // RATIS_DATASTREAM port). Otherwise the stale record would keep + // streaming clients from reaching the datastream port (HDDS-15799). + LOG.info("Updating ports for registered datanode {}: {} -> {}", + datanodeDetails, oldNode.getPorts(), datanodeDetails.getPorts()); + nodeStateManager.updateNode(datanodeDetails, layoutInfo); } } catch (NodeNotFoundException e) { LOG.error("Cannot find datanode {} from nodeStateManager", @@ -487,6 +495,22 @@ public RegisteredCommand register( .build(); } + /** + * Whether the datanode's exposed ports changed between two registrations. + * Compared as a set of name=value entries, since + * {@link DatanodeDetails.Port#equals} ignores the port value. + */ + private static boolean portsChanged(DatanodeDetails oldNode, + DatanodeDetails newNode) { + return !portValues(oldNode).equals(portValues(newNode)); + } + + private static Set portValues(DatanodeDetails datanodeDetails) { + return datanodeDetails.getPorts().stream() + .map(port -> port.getName() + "=" + port.getValue()) + .collect(Collectors.toSet()); + } + /** * Add an entry to the dnsToUuidMap, which maps hostname / IP to the DNs * running on that host. As each address can have many DNs running on it, diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java index e0d9de1f10b5..fa98650e6522 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManager.java @@ -117,6 +117,15 @@ void addContainerToPipeline(PipelineID pipelineID, ContainerID containerID) void closeStalePipelines(DatanodeDetails datanodeDetails); + /** + * Close OPEN pipelines whose datanodes now expose a port (the + * RATIS_DATASTREAM port after Ratis DataStream was enabled) that the + * pipeline's stored node snapshot lacks, so streaming-capable pipelines are + * created in their place. A pre-datastream Raft group cannot be made + * streamable in place, so it must be recreated (HDDS-12991). + */ + void closeNonStreamablePipelines(); + void scrubPipelines() throws IOException; void startPipelineCreator(); diff --git a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java index 8003d8d52e96..42a0ea4221f6 100644 --- a/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java +++ b/hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java @@ -182,13 +182,9 @@ public static PipelineManagerImpl newPipelineManager( .setServiceName("BackgroundPipelineScrubber") .setIntervalInMillis(scrubberIntervalInMillis) .setWaitTimeInMillis(safeModeWaitMs) - .setPeriodicalTask(() -> { - try { - pipelineManager.scrubPipelines(); - } catch (IOException e) { - LOG.error("Unexpected error during pipeline scrubbing", e); - } - }).build(); + .setPeriodicalTask( + pipelineManager::scrubAndCloseNonStreamablePipelines) + .build(); pipelineManager.setBackgroundPipelineScrubber(backgroundPipelineScrubber); serviceManager.register(backgroundPipelineScrubber); @@ -562,6 +558,68 @@ static boolean sameIdDifferentHostOrAddress(DatanodeDetails left, DatanodeDetail || !left.getHostName().equals(right.getHostName())); } + /** + * Close (and delete) OPEN pipelines that predate a datanode capability the + * registered node now advertises but the pipeline's stored node snapshot + * lacks — in practice the RATIS_DATASTREAM port after Ratis DataStream was + * enabled. Such pipelines cannot serve streaming even after the datanodes + * restart, because the Raft group's persisted configuration still carries the + * stale datastream address; only a freshly created pipeline is + * streaming-capable. BackgroundPipelineCreator recreates replacements from + * the now-capable nodes (HDDS-12991). + */ + void scrubAndCloseNonStreamablePipelines() { + try { + scrubPipelines(); + } catch (IOException e) { + LOG.error("Unexpected error during pipeline scrubbing", e); + } + closeNonStreamablePipelines(); + } + + @Override + public void closeNonStreamablePipelines() { + for (Pipeline pipeline : getPipelines()) { + if (!pipeline.isOpen() || !nodesExposeNewPorts(pipeline)) { + continue; + } + try { + final PipelineID id = pipeline.getId(); + LOG.info("Closing non-streamable pipeline {} so a streaming-capable " + + "pipeline can replace it", id); + closePipeline(id); + deletePipeline(id); + } catch (IOException e) { + LOG.error("Failed to close non-streamable pipeline {}", + pipeline.getId(), e); + } + } + } + + /** + * Whether any registered node of the pipeline exposes a port name that the + * pipeline's stored copy of that node lacks (e.g. RATIS_DATASTREAM added + * after the pipeline was created). + */ + private boolean nodesExposeNewPorts(Pipeline pipeline) { + for (DatanodeDetails stored : pipeline.getNodes()) { + final DatanodeDetails current = nodeManager.getNode(stored.getID()); + if (current != null && exposesNewPorts(stored, current)) { + return true; + } + } + return false; + } + + private static boolean exposesNewPorts(DatanodeDetails stored, + DatanodeDetails current) { + final Set storedNames = stored.getPorts().stream() + .map(DatanodeDetails.Port::getName).collect(Collectors.toSet()); + return current.getPorts().stream() + .map(DatanodeDetails.Port::getName) + .anyMatch(name -> !storedNames.contains(name)); + } + /** * Scrub pipelines. */ diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java index a0f1c1bf3623..78814b53e81a 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestSCMNodeManager.java @@ -43,6 +43,7 @@ import static org.apache.ozone.test.MetricsAsserts.getMetrics; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotEquals; import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertThrows; @@ -63,6 +64,7 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.UUID; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -344,6 +346,81 @@ private DatanodeDetails registerWithCapacity(SCMNodeManager nodeManager, return cmd.getDatanode(); } + private static DatanodeDetails.Builder datanodeWithoutDatastream(UUID uuid) { + return DatanodeDetails.newBuilder() + .setUuid(uuid) + .setHostName("host-" + uuid) + .setIpAddress("127.0.0.1") + .addPort(DatanodeDetails.newPort( + DatanodeDetails.Port.Name.STANDALONE, 9859)) + .addPort(DatanodeDetails.newPort( + DatanodeDetails.Port.Name.RATIS, 9858)); + } + + private void registerNode(SCMNodeManager nodeManager, DatanodeDetails dn) { + StorageReportProto storageReport = HddsTestUtils.createStorageReport( + dn.getID(), dn.getNetworkFullPath(), Long.MAX_VALUE); + MetadataStorageReportProto metadataStorageReport = + HddsTestUtils.createMetadataStorageReport( + dn.getNetworkFullPath(), Long.MAX_VALUE); + RegisteredCommand cmd = nodeManager.register(dn, + HddsTestUtils.createNodeReport(Arrays.asList(storageReport), + Arrays.asList(metadataStorageReport)), + getRandomPipelineReports(), UpgradeUtils.defaultLayoutVersionProto()); + assertEquals(success, cmd.getError()); + } + + /** + * A datanode that re-registers with the same identity but now exposes the + * RATIS_DATASTREAM port (e.g. Ratis DataStream was enabled) must have its + * stored record refreshed so the new port is visible (HDDS-15799). + */ + @Test + public void testRegisterRefreshesPortsOnPortChange() + throws IOException, AuthenticationException { + try (SCMNodeManager nodeManager = createNodeManager(getConf())) { + final UUID uuid = UUID.randomUUID(); + + // First registration: streaming disabled, no RATIS_DATASTREAM port. + registerNode(nodeManager, datanodeWithoutDatastream(uuid).build()); + DatanodeDetails stored = nodeManager.getNode( + datanodeWithoutDatastream(uuid).build().getID()); + assertFalse(stored.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM)); + + // Re-registration (same id/ip/host/version) now exposing the port. + registerNode(nodeManager, datanodeWithoutDatastream(uuid) + .addPort(DatanodeDetails.newPort( + DatanodeDetails.Port.Name.RATIS_DATASTREAM, 9855)) + .build()); + stored = nodeManager.getNode( + datanodeWithoutDatastream(uuid).build().getID()); + assertTrue(stored.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM), + "stored node should be refreshed with the RATIS_DATASTREAM port"); + } + } + + /** + * Re-registering a datanode with an unchanged port set must not disturb the + * stored record (the port-refresh branch is skipped). + */ + @Test + public void testRegisterKeepsPortsWhenUnchanged() + throws IOException, AuthenticationException { + try (SCMNodeManager nodeManager = createNodeManager(getConf())) { + final UUID uuid = UUID.randomUUID(); + + registerNode(nodeManager, datanodeWithoutDatastream(uuid).build()); + // Re-register with the identical port set. + registerNode(nodeManager, datanodeWithoutDatastream(uuid).build()); + + final DatanodeDetails stored = nodeManager.getNode( + datanodeWithoutDatastream(uuid).build().getID()); + assertTrue(stored.hasPort(DatanodeDetails.Port.Name.STANDALONE)); + assertTrue(stored.hasPort(DatanodeDetails.Port.Name.RATIS)); + assertFalse(stored.hasPort(DatanodeDetails.Port.Name.RATIS_DATASTREAM)); + } + } + private void assertPipelineClosedAfterLayoutHeartbeat( DatanodeDetails originalNode1, DatanodeDetails originalNode2, SCMNodeManager nodeManager, LayoutVersionProto layout) throws Exception { diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java index ef6f187b040c..473fd0b0ea27 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/MockPipelineManager.java @@ -249,6 +249,11 @@ public void closeStalePipelines(DatanodeDetails datanodeDetails) { } + @Override + public void closeNonStreamablePipelines() { + + } + @Override public void scrubPipelines() { diff --git a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java index 2f1d5ede5d7f..b9bc62a84434 100644 --- a/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java +++ b/hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/pipeline/TestPipelineManagerImpl.java @@ -88,6 +88,7 @@ import org.apache.hadoop.hdds.scm.ha.SCMHAManagerStub; import org.apache.hadoop.hdds.scm.ha.SCMServiceManager; import org.apache.hadoop.hdds.scm.metadata.SCMDBDefinition; +import org.apache.hadoop.hdds.scm.node.DatanodeInfo; import org.apache.hadoop.hdds.scm.node.NodeManager; import org.apache.hadoop.hdds.scm.node.NodeStatus; import org.apache.hadoop.hdds.scm.pipeline.choose.algorithms.HealthyPipelineChoosePolicy; @@ -100,6 +101,7 @@ import org.apache.hadoop.hdds.utils.db.DBStoreBuilder; import org.apache.hadoop.hdds.utils.db.Table; import org.apache.hadoop.metrics2.MetricsRecordBuilder; +import org.apache.hadoop.ozone.ClientVersion; import org.apache.hadoop.ozone.container.common.SCMTestUtils; import org.apache.ozone.test.GenericTestUtils; import org.apache.ozone.test.GenericTestUtils.LogCapturer; @@ -980,4 +982,115 @@ private static void assertFailsNotLeader(CheckedRunnable block) { assertEquals(ResultCodes.SCM_NOT_LEADER, e.getResult()); assertInstanceOf(NotLeaderException.class, e.getCause()); } + + private static DatanodeDetails portlessDatanode(DatanodeID id) { + return DatanodeDetails.newBuilder() + .setID(id) + .setHostName("host-" + id) + .setIpAddress("127.0.0.1") + .addPort(DatanodeDetails.newPort( + DatanodeDetails.Port.Name.STANDALONE, 9859)) + .addPort(DatanodeDetails.newPort( + DatanodeDetails.Port.Name.RATIS, 9858)) + .build(); + } + + private Pipeline addPipeline(PipelineManagerImpl pipelineManager, + Pipeline.PipelineState state, List nodes) + throws IOException { + final Pipeline pipeline = Pipeline.newBuilder() + .setReplicationConfig( + RatisReplicationConfig.getInstance(ReplicationFactor.THREE)) + .setNodes(nodes) + .setState(state) + .setId(PipelineID.randomId()) + .build(); + pipelineManager.getStateManager().addPipeline( + pipeline.getProtobufMessage(ClientVersion.CURRENT_VERSION)); + return pipeline; + } + + private static boolean exists(PipelineManagerImpl pipelineManager, + PipelineID id) { + try { + pipelineManager.getPipeline(id); + return true; + } catch (PipelineNotFoundException e) { + return false; + } + } + + @Test + public void testCloseNonStreamablePipelines() throws Exception { + try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { + // Registered datanodes (MockNodeManager) expose all ports incl datastream. + final List registered = nodeManager.getAllNodes(); + final List idsA = new ArrayList<>(); + final List idsB = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + idsA.add(portlessDatanode(registered.get(i).getID())); + idsB.add(portlessDatanode(registered.get(i + 3).getID())); + } + + // OPEN, registered nodes, portless -> legacy pipeline, must be closed. + final Pipeline stale = addPipeline(pipelineManager, OPEN, idsA); + // OPEN, registered nodes carrying all ports -> not stale, kept. + final Pipeline portful = addPipeline(pipelineManager, OPEN, + new ArrayList<>(registered.subList(6, 9))); + // OPEN, but nodes are NOT registered -> cannot heal, left alone. + final List unregistered = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + unregistered.add(portlessDatanode(DatanodeID.randomID())); + } + final Pipeline unreg = addPipeline(pipelineManager, OPEN, unregistered); + // ALLOCATED (non-open) portless -> skipped. + final Pipeline allocated = addPipeline(pipelineManager, ALLOCATED, idsB); + + pipelineManager.closeNonStreamablePipelines(); + + assertFalse(exists(pipelineManager, stale.getId()), + "legacy portless OPEN pipeline should be closed and deleted"); + assertTrue(exists(pipelineManager, portful.getId())); + assertTrue(exists(pipelineManager, unreg.getId())); + assertTrue(exists(pipelineManager, allocated.getId())); + } + } + + @Test + public void testCloseNonStreamablePipelinesSwallowsError() throws Exception { + try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { + final List registered = nodeManager.getAllNodes(); + final List nodes = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + nodes.add(portlessDatanode(registered.get(i).getID())); + } + final Pipeline stale = addPipeline(pipelineManager, OPEN, nodes); + + final PipelineManagerImpl spy = spy(pipelineManager); + doThrow(new IOException("boom")).when(spy).closePipeline(stale.getId()); + // The close failure is logged and swallowed; the loop does not throw. + spy.closeNonStreamablePipelines(); + assertTrue(exists(pipelineManager, stale.getId())); + } + } + + @Test + public void testScrubAndCloseWiring() throws Exception { + // The background task scrubs then closes non-streamable pipelines; on an + // empty manager both are no-ops and must not throw. + try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { + pipelineManager.scrubAndCloseNonStreamablePipelines(); + } + } + + @Test + public void testScrubAndCloseSwallowsScrubError() throws Exception { + try (PipelineManagerImpl pipelineManager = createPipelineManager(true)) { + final PipelineManagerImpl spy = spy(pipelineManager); + doThrow(new IOException("boom")).when(spy).scrubPipelines(); + // Scrub failure is logged and swallowed; the close pass still runs. + spy.scrubAndCloseNonStreamablePipelines(); + verify(spy).closeNonStreamablePipelines(); + } + } } diff --git a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java index 119af2f04e5c..b9a5a2e170b1 100644 --- a/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java +++ b/hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/io/KeyDataStreamOutput.java @@ -40,6 +40,7 @@ import org.apache.hadoop.hdds.scm.pipeline.PipelineID; import org.apache.hadoop.hdds.scm.storage.AbstractDataStreamOutput; import org.apache.hadoop.hdds.scm.storage.BlockDataStreamOutput; +import org.apache.hadoop.hdds.scm.storage.StreamNotSupportedException; import org.apache.hadoop.ozone.om.helpers.OmKeyInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfoGroup; @@ -207,6 +208,12 @@ private void handleWrite(ByteBuffer b, int off, long len, boolean retry) } len -= writtenLength; off += writtenLength; + } catch (StreamNotSupportedException e) { + // The pipeline cannot stream (e.g. datanodes created before DataStream + // was enabled). Surface it as-is so callers can fall back to the + // non-streaming write path (HDDS-12991). + markStreamClosed(); + throw e; } catch (Exception e) { markStreamClosed(); throw new IOException(e); diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/client/io/SelectorOutputStream.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/client/io/SelectorOutputStream.java index 0618d4a354b6..ca055c5b6be0 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/client/io/SelectorOutputStream.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/client/io/SelectorOutputStream.java @@ -23,7 +23,11 @@ import java.util.Objects; import org.apache.hadoop.fs.StreamCapabilities; import org.apache.hadoop.fs.Syncable; +import org.apache.hadoop.hdds.scm.storage.StreamNotSupportedException; +import org.apache.ratis.util.function.CheckedConsumer; import org.apache.ratis.util.function.CheckedFunction; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * An {@link OutputStream} first write data to a buffer up to the capacity. @@ -39,6 +43,9 @@ public class SelectorOutputStream extends OutputStream implements Syncable, StreamCapabilities { + private static final Logger LOG = + LoggerFactory.getLogger(SelectorOutputStream.class); + private final ByteArrayBuffer buffer; private final Underlying underlying; @@ -78,19 +85,26 @@ void write(byte[] src, int srcOffset, int length) { offset += length; } - OUT selectAndClose( - int outstandingBytes, boolean force, - CheckedFunction selector) - throws IOException { + /** Whether an {@link OutputStream} must be selected to hold the bytes. */ + boolean requiresSelection(int outstandingBytes, boolean force) { assertRemaining(0); - final int required = offset + outstandingBytes; - if (force || required > array.length) { - final OUT out = selector.apply(required); + return force || offset + outstandingBytes > array.length; + } + + /** The total number of bytes the selected stream must accept. */ + int required(int outstandingBytes) { + return offset + outstandingBytes; + } + + /** Write the buffered bytes to {@code out}; the buffer is not released. */ + void writePrefixTo(OutputStream out) throws IOException { + if (offset > 0) { out.write(array, 0, offset); - array = null; - return out; } - return null; + } + + void release() { + array = null; } } @@ -98,18 +112,71 @@ OUT selectAndClose( final class Underlying { /** Select an {@link OutputStream} by the number of bytes. */ private final CheckedFunction selector; + /** Selector used when the primary selection cannot stream; may be null. */ + private final CheckedFunction fallbackSelector; private OUT out; + /** No byte has been accepted by {@link #out} yet, so the buffered prefix is + * still intact and a {@link StreamNotSupportedException} can be recovered + * by switching to the fallback (non-streaming) output. */ + private boolean started = false; - private Underlying(CheckedFunction selector) { + private Underlying(CheckedFunction selector, + CheckedFunction fallbackSelector) { this.selector = selector; + this.fallbackSelector = fallbackSelector; } + /** Select an {@link OutputStream} if the buffer must be flushed. */ private OUT select(int outstandingBytes, boolean force) throws IOException { - if (out == null) { - out = buffer.selectAndClose(outstandingBytes, force, selector); + if (out == null && buffer.requiresSelection(outstandingBytes, force)) { + out = selector.apply(buffer.required(outstandingBytes)); } return out; } + + /** + * Flush the buffered prefix and the given payload to the selected stream. + * On the first flush, a {@link StreamNotSupportedException} (e.g. the chosen + * streaming output cannot be used because the pipeline lacks the + * RATIS_DATASTREAM port) is recovered by switching to the non-streaming + * fallback and replaying the buffered prefix and the payload. Once the + * first flush succeeds, the buffer is released and later writes go directly + * to the selected stream. + */ + private void firstFlush(CheckedConsumer payload) + throws IOException { + try { + buffer.writePrefixTo(out); + payload.accept(out); + } catch (StreamNotSupportedException e) { + if (fallbackSelector == null) { + throw e; + } + // The selected (streaming) output cannot be used for this pipeline + // (e.g. datanodes without the RATIS_DATASTREAM port). Fall back to the + // non-streaming output; the buffered prefix is still intact. + LOG.warn("Streaming output is not supported for this write; " + + "falling back to the non-streaming output stream.", e); + try { + out.close(); + } catch (IOException suppressed) { + e.addSuppressed(suppressed); + } + out = fallbackSelector.apply(buffer.required(0)); + buffer.writePrefixTo(out); + payload.accept(out); + } + started = true; + buffer.release(); + } + + /** Ensure the buffered prefix has been flushed to the selected stream. + * The caller selects with {@code force}, so {@link #out} is already set. */ + private void ensureStarted() throws IOException { + if (!started) { + firstFlush(ignored -> { }); + } + } } /** @@ -121,8 +188,24 @@ private OUT select(int outstandingBytes, boolean force) throws IOException { */ public SelectorOutputStream(int selectionThreshold, CheckedFunction selector) { + this(selectionThreshold, selector, null); + } + + /** + * Same as {@link #SelectorOutputStream(int, CheckedFunction)} but with a + * fallback selector used when the primary selection throws + * {@link StreamNotSupportedException} (e.g. the chosen streaming output cannot + * be used because the pipeline lacks the RATIS_DATASTREAM port). The buffered + * data is re-written to the fallback stream. + * + * @param fallbackSelector produces the non-streaming output; {@code null} + * disables the fallback (original behavior). + */ + public SelectorOutputStream(int selectionThreshold, + CheckedFunction selector, + CheckedFunction fallbackSelector) { this.buffer = new ByteArrayBuffer(selectionThreshold); - this.underlying = new Underlying(selector); + this.underlying = new Underlying(selector, fallbackSelector); } public OUT getUnderlying() { @@ -132,26 +215,32 @@ public OUT getUnderlying() { @Override public void write(int b) throws IOException { final OUT out = underlying.select(1, false); - if (out != null) { - out.write(b); - } else { + if (out == null) { buffer.write((byte) b); + } else if (!underlying.started) { + underlying.firstFlush(o -> o.write(b)); + } else { + out.write(b); } } @Override public void write(@Nonnull byte[] array, int off, int len) throws IOException { - final OUT selected = underlying.select(len, false); - if (selected != null) { - selected.write(array, off, len); - } else { + final OUT out = underlying.select(len, false); + if (out == null) { buffer.write(array, off, len); + } else if (!underlying.started) { + underlying.firstFlush(o -> o.write(array, off, len)); + } else { + out.write(array, off, len); } } private OUT select() throws IOException { - return underlying.select(0, true); + final OUT out = underlying.select(0, true); + underlying.ensureStarted(); + return out; } @Override diff --git a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/client/io/TestSelectorOutputStream.java b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/client/io/TestSelectorOutputStream.java index 22ed243edb43..ad2237a45a09 100644 --- a/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/client/io/TestSelectorOutputStream.java +++ b/hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/client/io/TestSelectorOutputStream.java @@ -18,15 +18,21 @@ package org.apache.hadoop.ozone.client.io; import static org.assertj.core.api.Assertions.assertThat; +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.assertSame; import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; import java.io.ByteArrayOutputStream; import java.io.IOException; import java.io.OutputStream; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.Supplier; +import org.apache.hadoop.fs.StreamCapabilities; import org.apache.hadoop.fs.Syncable; +import org.apache.hadoop.hdds.scm.storage.StreamNotSupportedException; import org.apache.ratis.util.MemoizedSupplier; import org.apache.ratis.util.function.CheckedConsumer; import org.apache.ratis.util.function.CheckedFunction; @@ -157,4 +163,254 @@ void testHSyncNonSyncable() { LOG.info("thrown", thrown); assertThat(thrown).hasMessageContaining("not Syncable"); } + + /** + * When the selected (streaming) output throws + * {@link StreamNotSupportedException} on its first write, the stream must + * fall back to the non-streaming selector and preserve all buffered data + * (HDDS-12991). + */ + @Test + void testFallbackOnStreamNotSupported() throws Exception { + final int threshold = 10; + final AtomicBoolean streamingClosed = new AtomicBoolean(false); + final ByteArrayOutputStream fallbackOut = new ByteArrayOutputStream(); + + // The streaming selection cannot stream: it always throws on write and + // records that it was closed. Created inside the selector so it is owned + // (and closed) by the SelectorOutputStream under test. + final CheckedFunction selector = + byteWritten -> byteWritten <= threshold + ? new ByteArrayOutputStream() + : new OutputStream() { + @Override + public void write(int b) throws IOException { + throw new StreamNotSupportedException("no RATIS_DATASTREAM port"); + } + + @Override + public void write(byte[] b, int off, int len) + throws IOException { + throw new StreamNotSupportedException("no RATIS_DATASTREAM port"); + } + + @Override + public void close() { + streamingClosed.set(true); + } + }; + final CheckedFunction fallbackSelector = + byteWritten -> fallbackOut; + + final SelectorOutputStream out = new SelectorOutputStream<>( + threshold, selector, fallbackSelector); + + final byte[] data = new byte[threshold + 5]; + for (int i = 0; i < data.length; i++) { + data[i] = (byte) i; + out.write(data[i]); + } + out.close(); + + // The streaming attempt was made and closed, and every byte (buffered and + // subsequent) landed in the non-streaming fallback. + assertTrue(streamingClosed.get()); + assertArrayEquals(data, fallbackOut.toByteArray()); + } + + /** + * Without a fallback selector, StreamNotSupportedException propagates. + */ + @Test + void testNoFallbackPropagates() { + final int threshold = 10; + final CheckedFunction selector = + byteWritten -> new OutputStream() { + @Override + public void write(int b) throws IOException { + throw new StreamNotSupportedException("no RATIS_DATASTREAM port"); + } + + @Override + public void write(byte[] b, int off, int len) + throws IOException { + throw new StreamNotSupportedException("no RATIS_DATASTREAM port"); + } + }; + final SelectorOutputStream out = + new SelectorOutputStream<>(threshold, selector); + try { + assertThrows(StreamNotSupportedException.class, () -> { + for (int i = 0; i < threshold + 5; i++) { + out.write(i); + } + }); + } finally { + // The selector always throws, so close() cannot flush the buffer either; + // release it quietly. + try { + out.close(); + } catch (IOException ignored) { + // expected: the underlying selection fails by design + } + } + } + + /** + * A {@code write(byte[], off, len)} that stays below the threshold is + * buffered and only flushed to the selected stream on close. + */ + @Test + void testArrayWriteBuffered() throws Exception { + final int threshold = 100; + final MemoizedSupplier below + = MemoizedSupplier.valueOf(ByteArrayOutputStream::new); + final MemoizedSupplier above + = MemoizedSupplier.valueOf(ByteArrayOutputStream::new); + final CheckedFunction selector + = byteWritten -> byteWritten <= threshold ? below.get() : above.get(); + + final byte[] data = newData(20); + try (SelectorOutputStream out = + new SelectorOutputStream<>(threshold, selector)) { + out.write(data, 0, 10); + out.write(data, 10, 10); + // Still buffered: no stream selected yet. + assertFalse(below.isInitialized()); + assertFalse(above.isInitialized()); + } + assertTrue(below.isInitialized()); + assertFalse(above.isInitialized()); + assertArrayEquals(data, + ((ByteArrayOutputStream) below.get()).toByteArray()); + } + + /** + * A {@code write(byte[], off, len)} above the threshold selects the stream + * and flushes the buffered prefix; a subsequent array write goes directly to + * the selected stream. + */ + @Test + void testArrayWriteSelectsThenDirect() throws Exception { + final int threshold = 10; + final ByteArrayOutputStream selected = new ByteArrayOutputStream(); + final CheckedFunction selector = + byteWritten -> selected; + + final byte[] data = newData(30); + final SelectorOutputStream out = + new SelectorOutputStream<>(threshold, selector); + out.write(data, 0, 5); // buffered (below threshold) + out.write(data, 5, 20); // exceeds threshold: select + firstFlush + assertSame(selected, out.getUnderlying()); + out.write(data, 25, 5); // started: direct write + out.close(); + + assertArrayEquals(data, selected.toByteArray()); + } + + /** + * Fallback triggered by a {@code write(byte[], off, len)} whose streaming + * stream both throws {@link StreamNotSupportedException} and fails to close; + * the close failure must not mask the fallback, which still receives the data. + */ + @Test + void testArrayWriteFallbackWithFailingClose() throws Exception { + final int threshold = 10; + final ByteArrayOutputStream fallbackOut = new ByteArrayOutputStream(); + // The streaming selection both throws on write and fails to close; created + // inside the selector so it is owned (and closed) by the stream under test. + final CheckedFunction selector = + byteWritten -> byteWritten <= threshold + ? new ByteArrayOutputStream() + : new OutputStream() { + @Override + public void write(int b) throws IOException { + throw new StreamNotSupportedException("no RATIS_DATASTREAM port"); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + throw new StreamNotSupportedException("no RATIS_DATASTREAM port"); + } + + @Override + public void close() throws IOException { + throw new IOException("close failed"); + } + }; + final CheckedFunction fallbackSelector = + byteWritten -> fallbackOut; + + final byte[] data = newData(threshold + 5); + try (SelectorOutputStream out = new SelectorOutputStream<>( + threshold, selector, fallbackSelector)) { + out.write(data, 0, data.length); + assertSame(fallbackOut, out.getUnderlying()); + } + assertArrayEquals(data, fallbackOut.toByteArray()); + } + + /** + * {@link SelectorOutputStream#hasCapability(String)} delegates to the selected + * stream when it is {@link StreamCapabilities}, returns false otherwise, and + * returns false when selection fails. + */ + @Test + void testHasCapability() throws Exception { + final int threshold = 10; + + // Underlying supports capabilities: delegate to it. + final CheckedFunction capable = + byteWritten -> new CapableOutputStream(true); + try (SelectorOutputStream out = + new SelectorOutputStream<>(threshold, capable)) { + assertTrue(out.hasCapability("hsync")); + } + + // Underlying is not StreamCapabilities: false. + try (SelectorOutputStream out = new SelectorOutputStream<>( + threshold, byteWritten -> new ByteArrayOutputStream())) { + assertFalse(out.hasCapability("hsync")); + } + + // Selection throws: swallowed, false. + final SelectorOutputStream failing = + new SelectorOutputStream<>(threshold, byteWritten -> { + throw new IOException("cannot select"); + }); + try { + assertFalse(failing.hasCapability("hsync")); + } finally { + // Selection always throws, so nothing is opened and close() rethrows it; + // release it quietly. + try { + failing.close(); + } catch (IOException ignored) { + // expected: the underlying selection fails by design + } + } + } + + private static byte[] newData(int length) { + final byte[] data = new byte[length]; + for (int i = 0; i < length; i++) { + data[i] = (byte) i; + } + return data; + } + + private static final class CapableOutputStream extends ByteArrayOutputStream + implements StreamCapabilities { + private final boolean capable; + + private CapableOutputStream(boolean capable) { + this.capable = capable; + } + + @Override + public boolean hasCapability(String capability) { + return capable; + } + } } diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java new file mode 100644 index 000000000000..03e2e2eb65cb --- /dev/null +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemDataStreamEnablement.java @@ -0,0 +1,302 @@ +/* + * 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 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): + *

+ */ +public class TestOzoneFileSystemDataStreamEnablement { + + // Small threshold/payload keep the writes fast while still selecting the + // streaming path (payload > threshold). + private static final int AUTO_THRESHOLD = 4 << 10; + private static final int WRITE_SIZE = 256 << 10; + + private MiniOzoneCluster cluster; + private OzoneClient client; + private OzoneBucket bucket; + private OzoneConfiguration conf; + + private void startClusterWithDatanodeStreamDisabled() throws Exception { + conf = new OzoneConfiguration(); + // Datanode side: datastream initially disabled, so pipelines are created + // without the RATIS_DATASTREAM port. + conf.setBoolean(HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, false); + // Client side: always attempt streaming writes. + conf.setBoolean(OZONE_FS_DATASTREAM_ENABLED, true); + conf.set(OZONE_FS_DATASTREAM_AUTO_THRESHOLD, AUTO_THRESHOLD + "B"); + conf.setInt(OZONE_SCM_RATIS_PIPELINE_LIMIT, 10); + // A long stale interval keeps the OPEN pipeline alive across the (no + // stop-wait) rolling restart, so the test drives pipeline closure itself. + conf.set(OZONE_SCM_STALENODE_INTERVAL, "5m"); + conf.set(OZONE_SCM_DEADNODE_INTERVAL, "10m"); + conf.set(HDDS_HEARTBEAT_INTERVAL, "1s"); + conf.set(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, "1s"); + // Recreate pipelines quickly after a close so the test does not wait the + // default two minutes for a fresh RATIS/THREE pipeline. + conf.set(OZONE_SCM_PIPELINE_CREATION_INTERVAL, "1s"); + + final int chunkSize = 16 << 10; + ClientConfigForTesting.newBuilder(StorageUnit.BYTES) + .setChunkSize(chunkSize) + .setStreamBufferFlushSize(32 << 10) + .setStreamBufferMaxSize(64 << 10) + .setDataStreamBufferFlushSize(64 << 10) + .setDataStreamMinPacketSize(chunkSize) + .setDataStreamWindowSize(5 * chunkSize) + .setBlockSize(1 << 20) + .applyTo(conf); + + cluster = MiniOzoneCluster.newBuilder(conf).setNumDatanodes(3).build(); + cluster.waitForClusterToBeReady(); + client = cluster.newClient(); + bucket = TestDataUtil.createVolumeAndBucket(client, + BucketLayout.FILE_SYSTEM_OPTIMIZED); + } + + @AfterEach + public void teardown() { + IOUtils.closeQuietly(client); + if (cluster != null) { + cluster.shutdown(); + } + } + + private FileSystem fs() throws IOException { + final String rootPath = String.format("%s://%s.%s/", + OZONE_URI_SCHEME, bucket.getName(), bucket.getVolumeName()); + conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, rootPath); + return FileSystem.get(conf); + } + + /** Write {@code data} and return the underlying stream selected by the FS. */ + private static Class writeAndGetUnderlying(FileSystem fs, Path path, + byte[] data) throws IOException { + final FSDataOutputStream out = fs.create(path, true); + out.write(data); + final SelectorOutputStream selector = + (SelectorOutputStream) out.getWrappedStream(); + out.close(); + return selector.getUnderlying().getClass(); + } + + private static void assertRoundTrips(FileSystem fs, Path path, byte[] expected) + throws IOException { + final byte[] read = new byte[expected.length]; + try (FSDataInputStream in = fs.open(path)) { + in.readFully(read); + } + assertArrayEquals(expected, read); + } + + private static byte[] randomBytes() { + final byte[] bytes = new byte[WRITE_SIZE]; + ThreadLocalRandom.current().nextBytes(bytes); + return bytes; + } + + private List openRatisThreePipelines() { + return cluster.getStorageContainerManager().getPipelineManager() + .getPipelines(RatisReplicationConfig.getInstance(THREE), + Pipeline.PipelineState.OPEN); + } + + private static boolean allNodesHaveDatastreamPort(Pipeline pipeline) { + return pipeline.getNodes().stream() + .allMatch(n -> n.hasPort(RATIS_DATASTREAM)); + } + + /** + * Enable datastream on every datanode via a rolling restart. {@code false} + * (no stop-wait) keeps each restart short; combined with the long stale + * interval the OPEN pipeline survives, so its node snapshot stays portless. + */ + private void rollingRestartEnablingDataStream() throws Exception { + for (int i = 0; i < cluster.getHddsDatanodes().size(); i++) { + cluster.getHddsDatanodes().get(i).getConf() + .setBoolean(HDDS_CONTAINER_RATIS_DATASTREAM_ENABLED, true); + cluster.restartHddsDatanode(i, false); + } + cluster.waitForClusterToBeReady(); + } + + /** Poll until SCM's node records all expose RATIS_DATASTREAM (validates D). */ + private void waitForAllRegisteredNodesToHaveDatastreamPort() + throws InterruptedException, TimeoutException { + final BooleanSupplier ready = () -> { + final List nodes = cluster + .getStorageContainerManager().getScmNodeManager().getAllNodes(); + return nodes.size() == cluster.getHddsDatanodes().size() + && nodes.stream().allMatch(n -> n.hasPort(RATIS_DATASTREAM)); + }; + GenericTestUtils.waitFor(ready, 500, 30_000); + } + + /** Poll until an OPEN RATIS/THREE pipeline exposes RATIS_DATASTREAM ports. */ + private void waitForStreamablePipeline() + throws InterruptedException, TimeoutException { + final BooleanSupplier ready = () -> openRatisThreePipelines().stream() + .anyMatch(TestOzoneFileSystemDataStreamEnablement + ::allNodesHaveDatastreamPort); + GenericTestUtils.waitFor(ready, 500, 60_000); + } + + /** + * While the datanodes have datastream disabled, a streaming client write + * falls back to the non-streaming path and succeeds. After a rolling restart + * enables datastream, SCM refreshes the datanodes' ports; the pre-existing + * pipeline is still portless, so writes keep falling back gracefully until it + * is replaced. + */ + @Test + @Timeout(value = 50, unit = TimeUnit.SECONDS) + public void testDataStreamFallbackAndPortRefresh() throws Exception { + startClusterWithDatanodeStreamDisabled(); + + try (FileSystem fs = fs()) { + final byte[] data = randomBytes(); + + // Datanode datastream disabled -> streaming falls back, still writes. + final Path before = new Path("/before-enable.dat"); + assertEquals(CapableOzoneFSOutputStream.class, + writeAndGetUnderlying(fs, before, data)); + assertRoundTrips(fs, before, data); + + rollingRestartEnablingDataStream(); + + // SCM's node records now expose the RATIS_DATASTREAM port. + waitForAllRegisteredNodesToHaveDatastreamPort(); + + // The legacy pipeline is still portless, so a streaming write keeps + // falling back gracefully and still succeeds. + final Path after = new Path("/after-enable.dat"); + assertEquals(CapableOzoneFSOutputStream.class, + writeAndGetUnderlying(fs, after, data)); + assertRoundTrips(fs, after, data); + } + } + + /** + * A pipeline created while datastream was disabled keeps a portless node + * snapshot and a stale datastream address in its Raft group, so it can + * never serve streaming even after the datanodes restart. SCM closes it so a + * fresh, streaming-capable pipeline is created; a streaming write then + * succeeds over the new pipeline (HDDS-12991). + */ + @Test + @Timeout(value = 70, unit = TimeUnit.SECONDS) + public void testCloseNonStreamablePipelineThenStream() throws Exception { + startClusterWithDatanodeStreamDisabled(); + + try (FileSystem fs = fs()) { + final byte[] data = randomBytes(); + + // Create a portless OPEN pipeline via a fallback write. + final Path p1 = new Path("/legacy.dat"); + assertEquals(CapableOzoneFSOutputStream.class, + writeAndGetUnderlying(fs, p1, data)); + final List before = openRatisThreePipelines(); + assertFalse(before.isEmpty()); + before.forEach(p -> assertFalse(allNodesHaveDatastreamPort(p), + "pipeline should be portless before enabling datastream")); + + rollingRestartEnablingDataStream(); + waitForAllRegisteredNodesToHaveDatastreamPort(); + + // SCM restart reloads the persisted (still portless) pipeline while the + // datanodes are registered with the port (no re-registration event fires). + cluster.restartStorageContainerManager(true); + waitForAllRegisteredNodesToHaveDatastreamPort(); + + final PipelineManager pipelineManager = + cluster.getStorageContainerManager().getPipelineManager(); + final List reloaded = openRatisThreePipelines(); + assertFalse(reloaded.isEmpty()); + reloaded.forEach(p -> assertFalse(allNodesHaveDatastreamPort(p), + "reloaded pipeline should still be portless")); + + // Close the non-streamable pipeline(s); a fresh streaming-capable + // pipeline is created in their place by BackgroundPipelineCreator. + pipelineManager.closeNonStreamablePipelines(); + waitForStreamablePipeline(); + + // The new pipeline actually serves a streaming write end-to-end. + final Path p2 = new Path("/after-recreate.dat"); + assertEquals(CapableOzoneFSDataStreamOutput.class, + writeAndGetUnderlying(fs, p2, data)); + assertRoundTrips(fs, p2, data); + } + } +} diff --git a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemWithStreamingDisabledDatanode.java b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemWithStreamingDisabledDatanode.java index cb59a3ae4da1..66a12d88ee60 100644 --- a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemWithStreamingDisabledDatanode.java +++ b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/fs/ozone/TestOzoneFileSystemWithStreamingDisabledDatanode.java @@ -21,13 +21,15 @@ 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_OFS_URI_SCHEME; import static org.apache.hadoop.ozone.OzoneConsts.OZONE_URI_SCHEME; -import static org.junit.jupiter.api.Assertions.assertThrows; -import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; import java.io.IOException; import java.util.concurrent.ThreadLocalRandom; 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; @@ -45,8 +47,10 @@ import org.junit.jupiter.api.Test; /** - * Verifies that when DataNode-side datastream is disabled, the client fails - * fast instead of falling back to the gRPC port for streaming writes. + * Verifies that when DataNode-side datastream is disabled (so pipelines carry + * no RATIS_DATASTREAM port), a streaming-enabled client does not fail. Instead + * of using the gRPC port for streaming, it gracefully falls back to the + * non-streaming write path and the data is written correctly (HDDS-12991). */ public class TestOzoneFileSystemWithStreamingDisabledDatanode { @@ -97,48 +101,42 @@ public static void teardown() { } @Test - public void testDatastreamWriteFailsFastWhenDatanodeStreamingDisabled() + public void testDatastreamWriteFallsBackWhenDatanodeStreamingDisabled() throws Exception { - final String rootPath = String.format("%s://%s.%s/", - OZONE_URI_SCHEME, bucket.getName(), bucket.getVolumeName()); - conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, rootPath); - final byte[] bytes = new byte[3 << 20]; ThreadLocalRandom.current().nextBytes(bytes); - try (FileSystem fs = FileSystem.get(conf)) { - final Path file = new Path("/streaming-disabled-dn.dat"); - - IOException ex = assertThrows(IOException.class, () -> { - try (FSDataOutputStream out = fs.create(file, true)) { - out.write(bytes); - } - }); - - final String msg = collectMessages(ex).toLowerCase(); - - // Expect a clear fail-fast error from our validation. - assertTrue( - msg.contains("ratis_datastream port is missing") - || msg.contains("datastream is disabled") - || msg.contains("ratis_datastream"), - () -> "Expected a clear fail-fast datastream error, but got: " + ex); - - // Ensure we did not fall back to the old gRPC/HTTP2 failure path. - assertTrue( - !msg.contains("http/2") && !msg.contains("timeout"), - () -> "Should not fall back to gRPC/HTTP2 path, but got: " + ex); - } + // o3fs (BasicOzoneFileSystem): bucket-scoped scheme. + final String o3fsRoot = String.format("%s://%s.%s/", + OZONE_URI_SCHEME, bucket.getName(), bucket.getVolumeName()); + assertFallbackRoundTrip(o3fsRoot, + new Path("/streaming-disabled-dn.dat"), bytes); + + // ofs (BasicRootedOzoneFileSystem): root-scoped scheme. + final String ofsRoot = String.format("%s://%s/", + OZONE_OFS_URI_SCHEME, conf.get(OZONE_OM_ADDRESS_KEY)); + final Path ofsFile = new Path(String.format("/%s/%s/streaming-disabled-ofs.dat", + bucket.getVolumeName(), bucket.getName())); + assertFallbackRoundTrip(ofsRoot, ofsFile, bytes); } - private static String collectMessages(Throwable t) { - StringBuilder sb = new StringBuilder(); - while (t != null) { - if (t.getMessage() != null) { - sb.append(t.getMessage()).append('\n'); + /** + * Write via the given filesystem root with client datastream enabled; the + * streaming path must detect the missing RATIS_DATASTREAM port, silently fall + * back to the non-streaming path, and the data must round-trip correctly. + */ + private void assertFallbackRoundTrip(String rootPath, Path file, byte[] bytes) + throws IOException { + conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY, rootPath); + try (FileSystem fs = FileSystem.get(conf)) { + try (FSDataOutputStream out = fs.create(file, true)) { + out.write(bytes); + } + final byte[] readBack = new byte[bytes.length]; + try (FSDataInputStream in = fs.open(file)) { + in.readFully(readBack); } - t = t.getCause(); + assertArrayEquals(bytes, readBack); } - return sb.toString(); } } diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java index 8e0f4e8a5dd2..4567f5c7de99 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicOzoneFileSystem.java @@ -311,8 +311,13 @@ private FSDataOutputStream createOutputStream(String key, short replication, final CheckedFunction selector = byteWritten -> selectOutputStream( key, replication, overwrite, recursive, byteWritten); + // Fallback used when the streaming output cannot be used for the target + // pipeline (no RATIS_DATASTREAM port): write via the non-streaming path. + final CheckedFunction fallback + = byteWritten -> createFSOutputStream(adapter.createFile( + key, replication, overwrite, recursive)); return new FSDataOutputStream(new SelectorOutputStream<>( - streamingAutoThreshold, selector), statistics); + streamingAutoThreshold, selector, fallback), statistics); } return new FSDataOutputStream(createFSOutputStream( diff --git a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java index 67f09313d0ea..7c1d53791088 100644 --- a/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java +++ b/hadoop-ozone/ozonefs-common/src/main/java/org/apache/hadoop/fs/ozone/BasicRootedOzoneFileSystem.java @@ -309,8 +309,13 @@ private FSDataOutputStream createOutputStream(String key, short replication, final CheckedFunction selector = byteWritten -> selectOutputStream( key, replication, overwrite, recursive, byteWritten); + // Fallback used when the streaming output cannot be used for the target + // pipeline (no RATIS_DATASTREAM port): write via the non-streaming path. + final CheckedFunction fallback + = byteWritten -> createFSOutputStream(adapter.createFile( + key, replication, overwrite, recursive)); return new FSDataOutputStream(new SelectorOutputStream<>( - streamingAutoThreshold, selector), statistics); + streamingAutoThreshold, selector, fallback), statistics); } return new FSDataOutputStream(createFSOutputStream( adapter.createFile(key,