yandrey321 commented on code in PR #10842:
URL: https://github.com/apache/ozone/pull/10842#discussion_r3677143118
##########
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,
Review Comment:
Recreating EC pipelines when data streaming is enabled is unwanted side
effect of this change, so we need to limit the scope to Ratis only pipelines
and use-case when data streaming feature flag is enabled.
--
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]
