HDDS-12991 recreate pipelines after enabling ratis write streaming#10842
HDDS-12991 recreate pipelines after enabling ratis write streaming#10842yandrey321 wants to merge 1 commit into
Conversation
chihsuan
left a comment
chihsuan
left a comment
There was a problem hiding this comment.
Thanks for working on this! I left a few comments around the pipeline retirement condition but the overall approach looks reasonable.
Also, the three integration tests each start their own MiniOzoneCluster, which probably would add noticeable CI cost. Would you consider combining these into one end-to-end lifecycle test?
| .map(DatanodeDetails.Port::getName).collect(Collectors.toSet()); | ||
| return current.getPorts().stream() | ||
| .map(DatanodeDetails.Port::getName) | ||
| .anyMatch(name -> !storedNames.contains(name)); |
There was a problem hiding this comment.
SCMNodeManager.portsChanged handles name=value changes, while this check only compares names. Should we also compare the RATIS_DATASTREAM port value here?
| return false; | ||
| } | ||
|
|
||
| private static boolean exposesNewPorts(DatanodeDetails stored, |
There was a problem hiding this comment.
I wonder if we could narrow this check to RATIS pipelines and the RATIS_DATASTREAM port specifically. Older pipeline snapshots may legitimately lack ports added by later versions, such as HTTP, HTTPS, or CLIENT_RPC. In that case, checking any new port could retire unrelated pipelines, including EC pipelines.
| private boolean nodesExposeNewPorts(Pipeline pipeline) { | ||
| for (DatanodeDetails stored : pipeline.getNodes()) { | ||
| final DatanodeDetails current = nodeManager.getNode(stored.getID()); | ||
| if (current != null && exposesNewPorts(stored, current)) { |
There was a problem hiding this comment.
One concern during a rolling restart: this may close the pipeline as soon as the first member advertises the new port. A replacement could still include datanodes that have not restarted yet, leading to another close-and-recreate cycle later. Would it make sense if we defer retirement until SCM can ensure the replacement uses only DataStream-capable nodes?
| * succeeds over the new pipeline (HDDS-12991). | ||
| */ | ||
| @Test | ||
| @Timeout(value = 70, unit = TimeUnit.SECONDS) |
There was a problem hiding this comment.
Would it make sense to rely on the repository’s five-minute default timeout here? This test waits up to 30 seconds for port registration both before and after the SCM restart, then up to 60 seconds for a replacement pipeline.
With cluster startup and restarts as well, the 70-second outer timeout may fire first and hide which operation was actually slow.
The same concern also applies to lines 225 and 318.
2 participants
What changes were proposed in this pull request?
SCM refreshes a datanode's ports on re-registration (SCMNodeManager)
When a datanode re-registers with a changed port set — e.g. it restarted with DataStream enabled and now advertises the RATIS_DATASTREAM port — SCM updates its stored NodeManager record instead of keeping the stale one. Previously a port-only change (same IP/hostname/version) was ignored, so SCM never learned the new port and every new pipeline stayed port-less.
New PipelineManager.closeNonStreamablePipelines(): closes+deletes OPEN pipelines whose persisted node snapshot lacks a port the currently-registered nodes now expose BackgroundPipelineCreator then builds fresh, streaming-capable replacements. Runs automatically as part of the pipeline scrubber: scrubAndCloseNonStreamablePipelines() (scrubs, then closes non-streamable) is wired into the BackgroundPipelineScrubber.
Why close-and-recreate rather than heal in place: a pipeline created before DataStream persists a stale dataStreamAddress (the gRPC port) in its Raft group configuration. ServerState.setRaftConf() only re-adds peers to the gRPC RPC, not the datastream RPC, so the group can never stream in place — the group must be recreated.
What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-12991
How was this patch tested?
TestSCMNodeManager: testRegisterRefreshesPortsOnPortChange (ports changed → node record updated) and testRegisterKeepsPortsWhenUnchanged (unchanged → no update).
TestPipelineManagerImpl: coverage for closeNonStreamablePipelines() — closes a portless OPEN pipeline once nodes expose the new port, and is a no-op when port sets already match.
Integration — TestOzoneFileSystemDataStreamEnablement (MiniOzoneCluster, datanodes start with DataStream disabled, client with streaming enabled):
testDataStreamFallbackAndPortRefresh: writes fall back gracefully while a pipeline is port-less; SCM refreshes the datanodes' ports on re-registration (validates piece 1).
testCloseNonStreamablePipelineThenStream: a portless legacy pipeline survives a rolling restart + SCM restart; closeNonStreamablePipelines() retires it; a fresh streaming-capable pipeline forms and a streaming write succeeds end-to-end (validates piece 2).
testBatchWritesAcrossStreamingEnablement: full lifecycle over a batch of files — several files written while disabled (all succeed via fallback), enable + close legacy pipeline, then several more files that all stream over a RATIS_DATASTREAM-capable pipeline; asserts no write fails and the post-enablement writes take the streaming path.
when part 1 (#10828) is missing: