You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@ozone.apache.org by "yandrey321 (via GitHub)" <gi...@apache.org> on 2026/07/13 23:13:06 UTC
[PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
yandrey321 opened a new pull request, #10746:
URL: https://github.com/apache/ozone/pull/10746
## What changes were proposed in this pull request?
### Root cause
A datanode advertises its RATIS_DATASTREAM port in DatanodeDetails only when streaming is enabled at start-up, and a pipeline persists a snapshot of its members' DatanodeDetails. On a cluster that predates streaming:
1. When a datanode re-registers after streaming is enabled, SCMNodeManager.register() refreshed the stored node record only on an IP/hostname or version change, so the newly-exposed RATIS_DATASTREAM port was dropped (HDDS-15799).
2. DatanodeDetails.getPort(RATIS_DATASTREAM) silently falls back to the RATIS/gRPC port, so a streaming client could target the wrong port (HDDS-8687).
3. A pipeline created before streaming keeps a portless member snapshot. Even after its datanodes restart with streaming enabled, that pipeline cannot serve a stream: its Ratis group's persisted configuration was built with a datastream address that resolves to the gRPC port, and the group configuration is not rebuilt on restart. Such a pipeline has to be recreated to become streaming-capable.
### Changes
1. SCM refreshes a datanode's ports on re-registration — SCMNodeManager. When a registered datanode re-registers with a changed port set (e.g. now exposing RATIS_DATASTREAM), SCM updates its stored node record, comparing ports as a name→value set. This reuses the existing NodeStateManager.updateNode path; node state is rebuilt from registrations, so there is no schema or upgrade impact.
2. The streaming client degrades gracefully instead of failing — hadoop-hdds/client, hadoop-ozone/client, hadoop-ozone/common, ozonefs-common. A new StreamNotSupportedException is thrown by BlockDataStreamOutput.setupStream when a pipeline node lacks the RATIS_DATASTREAM port (using the explicit hasPort check, without changing getPort fallback semantics). KeyDataStreamOutput propagates it recognizably, and SelectorOutputStream recovers on the first flush by closing the streaming stream and replaying the buffered data to a non-streaming fallback stream, wired in BasicOzoneFileSystem/BasicRootedOzoneFileSystem. Writes to a not-yet-streaming pipeline therefore keep succeeding.
3. SCM replaces pipelines that cannot stream — PipelineManager. A new closeNonStreamablePipelines(), run from the background pipeline scrubber, closes and deletes OPEN pipelines whose registered datanodes now expose a port the pipeline's stored snapshot lacks. BackgroundPipelineCreator then builds fresh, streaming-capable pipelines from the now-capable nodes. Closing (rather than editing the snapshot) is required because a pre-streaming Ratis group cannot be made streamable in place. It reuses the existing replicated closePipeline/deletePipeline operations, so no new HA/Ratis operation is introduced.
Net effect: after streaming is enabled and the datanodes restart, writes keep succeeding via the fallback while SCM replaces the legacy pipelines; once the new pipelines exist, streaming writes flow over them — no metadata reset required.
Co-authored with Cloude Code
## What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-12991
https://issues.apache.org/jira/browse/HDDS-15799
https://issues.apache.org/jira/browse/HDDS-8687
## How was this patch tested?
- Unit: TestSCMNodeManager (port refresh on re-registration; no refresh when unchanged); TestSelectorOutputStream (fallback on first flush for write(int) and write(byte[]), buffered-data replay, failing-close handling, no-fallback propagation, hasCapability); TestPipelineManagerImpl (closeNonStreamablePipelines closes a legacy pipeline while leaving streaming-capable, unregistered-node, and non-open pipelines untouched; the scrub-and-close background task). New non-test code has 100% line and branch coverage.
- Integration: TestOzoneFileSystemWithStreamingDisabledDatanode (a streaming write falls back and round-trips over both o3fs and ofs); new TestOzoneFileSystemDataStreamEnablement: (a) after a rolling restart enables datastream, SCM refreshes the node ports and writes keep succeeding via fallback; (b) a pipeline that predates datastream is closed and a fresh one is created, over which a streaming write then succeeds end-to-end. Both integration tests use bounded `@Timeouts`.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #10746:
URL: https://github.com/apache/ozone/pull/10746#discussion_r3620679674
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java:
##########
@@ -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));
Review Comment:
Curious that if we know that we are fixing for `DatanodeDetails.Port.Name.RATIS_DATASTREAM` missing issue, can we only look for the case in which the newNode has the data stream port while the old node does not, thus we need to do an update?
It does not look wrong that we compare entire port set, but the side effect is I do not know if there is any port that do not need to be updated.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "szetszwo (via GitHub)" <gi...@apache.org>.
szetszwo commented on PR #10746:
URL: https://github.com/apache/ozone/pull/10746#issuecomment-5025037145
@yandrey321 , thanks for working on this!
This AI generated PR has the usual problems of AI generated PRs:
- the size is too big; see https://github.com/apache/ozone/pull/10673#issuecomment-4938876287
- it has many unnecessary changes;
- the comments are too long.
Could you clean them up.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "jojochuang (via GitHub)" <gi...@apache.org>.
jojochuang commented on PR #10746:
URL: https://github.com/apache/ozone/pull/10746#issuecomment-5024744559
@ss77892
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "peterxcli (via GitHub)" <gi...@apache.org>.
peterxcli commented on PR #10746:
URL: https://github.com/apache/ozone/pull/10746#issuecomment-4970975987
might related to https://github.com/apache/ozone/pull/10743/changes#r3573417136
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #10746:
URL: https://github.com/apache/ozone/pull/10746#discussion_r3620679674
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java:
##########
@@ -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));
Review Comment:
Curious that if we know that we are fixing for `DatanodeDetails.Port.Name.RATIS_DATASTREAM` missing issue, can we only look for the case in which the newNode has the data stream port while the old node does not, thus we need to do an update?
It does not look wrong that we compare entire port set, but the side effect is I do not know if there are any port that do not want to be updated.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on PR #10746:
URL: https://github.com/apache/ozone/pull/10746#issuecomment-4984683051
> might relate to https://github.com/apache/ozone/pull/10743/changes#r3573417136
this one would be the separate work item.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 commented on PR #10746:
URL: https://github.com/apache/ozone/pull/10746#issuecomment-5036338741
@szetszwo @jojochuang @amaliujia Please review the first subset of these changes: https://github.com/apache/ozone/pull/10828
I'll add another PR for the server side part of this PR.
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "yandrey321 (via GitHub)" <gi...@apache.org>.
yandrey321 closed pull request #10746: HDDS-12991. Support enabling Ratis DataStream on an existing cluster
URL: https://github.com/apache/ozone/pull/10746
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org
Re: [PR] HDDS-12991. Support enabling Ratis DataStream on an existing cluster [ozone]
Posted by "amaliujia (via GitHub)" <gi...@apache.org>.
amaliujia commented on code in PR #10746:
URL: https://github.com/apache/ozone/pull/10746#discussion_r3620679674
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/node/SCMNodeManager.java:
##########
@@ -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));
Review Comment:
Curious that if we know that we are fixing for `DatanodeDetails.Port.Name.RATIS_DATASTREAM` missing issue, can we only look for the case in which the newNode has the data stream port while the old node does not, thus we need to do an update?
--
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: issues-unsubscribe@ozone.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@ozone.apache.org
For additional commands, e-mail: issues-help@ozone.apache.org