Copilot commented on code in PR #10746:
URL: https://github.com/apache/ozone/pull/10746#discussion_r3617661777
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/pipeline/PipelineManagerImpl.java:
##########
@@ -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<DatanodeDetails.Port.Name> storedNames =
stored.getPorts().stream()
+ .map(DatanodeDetails.Port::getName).collect(Collectors.toSet());
+ return current.getPorts().stream()
+ .map(DatanodeDetails.Port::getName)
+ .anyMatch(name -> !storedNames.contains(name));
+ }
Review Comment:
closeNonStreamablePipelines currently treats *any* newly-advertised datanode
port name (vs the pipeline snapshot) as a reason to close and delete an OPEN
pipeline. This is broader than the documented intent (RATIS_DATASTREAM
enablement) and could cause unexpected pipeline churn if other ports are added
in future releases.
##########
hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/client/io/SelectorOutputStream.java:
##########
@@ -78,38 +85,98 @@ 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 suppressed) {
+ e.addSuppressed(suppressed);
+ }
Review Comment:
In SelectorOutputStream fallback handling, IOException from closing the
unsupported streaming output is added as suppressed to
StreamNotSupportedException, but the warning is logged *before* the close
attempt. Since the exception is recovered (not rethrown), the suppressed close
failure is never surfaced, making this hard to debug.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
