HDDS-12991. Support enabling Ratis DataStream on an existing cluster

[yandrey321]

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.

[peterxcli]

might relate to https://github.com/apache/ozone/pull/10743/changes#r3573417136