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-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..5e3cec54d650 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); @@ -224,6 +231,12 @@ private int writeToDataStreamOutput(BlockDataStreamOutputEntry current, current.write(b, off, writeLen); offset += writeLen; } + } catch (StreamNotSupportedException e) { + // The pipeline cannot stream (datanodes lack the RATIS_DATASTREAM port). + // Surface it immediately instead of routing through handleException(), + // whose retries would each hit the same missing port and only delay the + // caller's fall back to the non-streaming write path (HDDS-12991). + throw e; } catch (IOException ioe) { // for the current iteration, totalDataWritten - currentPos gives the // amount of data already written to the buffer 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..dab6a5fbcc11 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,75 @@ 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 closeFailure) { + // The StreamNotSupportedException is recovered (not rethrown), so a + // failure closing the discarded streaming output would be lost if + // only attached as suppressed. Log it here so it is surfaced. + LOG.warn("Failed to close the unsupported streaming output while " + + "falling back to the non-streaming output stream.", closeFailure); + } + 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 +192,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 +219,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..fc68410ee80e 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(); + + // Streaming output that cannot stream: always throws on write. Created + // inside the selector so its ownership transfers to the + // SelectorOutputStream (which closes it), rather than being an open + // local this method would have to close. + 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 { + // No fallback: closing re-attempts the flush and throws again. Close + // defensively so the stream is not left open. + try { + out.close(); + } catch (IOException ignored) { + // expected on close + } + } + } + + /** + * 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(); + // Streaming output that both throws on write and fails to close, created + // inside the selector so ownership transfers to the SelectorOutputStream. + 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 throws on close's flush attempt too; close defensively. + try { + failing.close(); + } catch (IOException ignored) { + // expected on close + } + } + } + + 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/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 7ade8339232c..7437b7a999c6 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 @@ -312,8 +312,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 6be8ebbddb6d..1af85ad2a721 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,