HDDS-12991 failback to non-streaming write when pipeline does not have streaming port info

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");
       }

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).
+ *
+ * <p>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);
+  }
+}

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

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<OUT extends OutputStream>
     extends OutputStream implements Syncable, StreamCapabilities {
 
+  private static final Logger LOG =
+      LoggerFactory.getLogger(SelectorOutputStream.class);
+
   private final ByteArrayBuffer buffer;
   private final Underlying underlying;
 
@@ -78,38 +85,102 @@ void write(byte[] src, int srcOffset, int length) {
       offset += length;
     }
 
-    <OUT extends OutputStream> OUT selectAndClose(
-        int outstandingBytes, boolean force,
-        CheckedFunction<Integer, OUT, IOException> 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;
     }
   }
 
   /** To select the underlying {@link OutputStream}. */
   final class Underlying {
     /** Select an {@link OutputStream} by the number of bytes. */
     private final CheckedFunction<Integer, OUT, IOException> selector;
+    /** Selector used when the primary selection cannot stream; may be null. */
+    private final CheckedFunction<Integer, OUT, IOException> 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<Integer, OUT, IOException> selector) {
+    private Underlying(CheckedFunction<Integer, OUT, IOException> selector,
+        CheckedFunction<Integer, OUT, IOException> 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<OUT, IOException> 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<Integer, OUT, IOException> 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<Integer, OUT, IOException> selector,
+      CheckedFunction<Integer, OUT, IOException> 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