[
https://issues.apache.org/jira/browse/NIFI-16125?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Xinyu Wang updated NIFI-16125:
------------------------------
Description:
During flow initialization the Flow Controller performs a one-time validation
pass over every component before scheduled/enabled components are started. This
pass runs synchronously on the startup thread, one component at a time. For
large flows (tens of thousands of components) it adds a measurable, avoidable
delay to every startup and cluster (re)join, spent on a single CPU core while
the rest sit idle.
*Current behavior (2.10.0)*
The initial pass is driven by \{{TriggerValidationTask.run()}}, which iterates
each component collection sequentially:
{code:java}
for (final ComponentNode node : flowManager.getAllControllerServices())
validationTrigger.trigger(node);
for (final ComponentNode node : flowManager.getAllReportingTasks())
validationTrigger.trigger(node);
for (final ComponentNode node : flowManager.getAllFlowAnalysisRules())
validationTrigger.trigger(node);
for (final ComponentNode node : flowManager.getAllParameterProviders())
validationTrigger.trigger(node);
for (final ComponentNode node : flowManager.getRootGroup().findAllProcessors())
validationTrigger.trigger(node);
for (final ComponentNode node : flowManager.getAllFlowRegistryClients())
validationTrigger.trigger(node);
for (final ConnectorNode connector : flowManager.getAllConnectors())
connector.validateComponents(validationTrigger);
{code}
It is invoked synchronously on the startup thread from
\{{FlowController.initializeFlow()}}:
{code:java}
new TriggerValidationTask(flowManager, triggerIfValidating).run(); // runs on
the calling (main) thread
LOG.info("Performed initial validation of all components in {} milliseconds",
millis);
{code}
The \{{triggerIfValidating}} trigger used here takes the synchronous path,
\{{validationTrigger.trigger(component)}} → \{{component.performValidation()}},
which blocks the calling thread per component.
A 5-thread validation pool already exists:
{code:java}
this.validationThreadPool = new FlowEngine(5, "Validate Components", true);
{code}
…but it is only used for the periodic re-validation task
(\{{scheduleWithFixedDelay}}, every 5s). The one-time initial sweep does not
use it.
*Proposed change*
Fan the initial sweep out over the existing {{validationThreadPool}} instead of
validating one component at a time, then barrier before returning (subsequent
startup steps assume components have reached a terminal validation state).
was:
During flow initialization the Flow Controller performs a one-time validation
pass over every component before scheduled/enabled components are started. This
pass runs synchronously on the startup thread, one component at a time. For
large flows (tens of thousands of components) it adds a measurable, avoidable
delay to every startup and cluster (re)join, spent on a single CPU core while
the rest sit idle.
*Proposed change*
Fan the initial sweep out over the existing \{{validationThreadPool}} instead
of validating one component at a time, then barrier before returning
(subsequent startup steps assume components have reached a terminal validation
state).
> Parallelize the initial component validation sweep during flow startup
> ----------------------------------------------------------------------
>
> Key: NIFI-16125
> URL: https://issues.apache.org/jira/browse/NIFI-16125
> Project: Apache NiFi
> Issue Type: Improvement
> Components: Core Framework
> Affects Versions: 2.5.0, 2.10.0
> Reporter: Xinyu Wang
> Priority: Minor
>
> During flow initialization the Flow Controller performs a one-time validation
> pass over every component before scheduled/enabled components are started.
> This pass runs synchronously on the startup thread, one component at a time.
> For large flows (tens of thousands of components) it adds a measurable,
> avoidable delay to every startup and cluster (re)join, spent on a single CPU
> core while the rest sit idle.
> *Current behavior (2.10.0)*
> The initial pass is driven by \{{TriggerValidationTask.run()}}, which
> iterates each component collection sequentially:
> {code:java}
> for (final ComponentNode node : flowManager.getAllControllerServices())
> validationTrigger.trigger(node);
> for (final ComponentNode node : flowManager.getAllReportingTasks())
> validationTrigger.trigger(node);
> for (final ComponentNode node : flowManager.getAllFlowAnalysisRules())
> validationTrigger.trigger(node);
> for (final ComponentNode node : flowManager.getAllParameterProviders())
> validationTrigger.trigger(node);
> for (final ComponentNode node :
> flowManager.getRootGroup().findAllProcessors())
> validationTrigger.trigger(node);
> for (final ComponentNode node : flowManager.getAllFlowRegistryClients())
> validationTrigger.trigger(node);
> for (final ConnectorNode connector : flowManager.getAllConnectors())
> connector.validateComponents(validationTrigger);
> {code}
> It is invoked synchronously on the startup thread from
> \{{FlowController.initializeFlow()}}:
> {code:java}
> new TriggerValidationTask(flowManager, triggerIfValidating).run(); // runs
> on the calling (main) thread
> LOG.info("Performed initial validation of all components in {} milliseconds",
> millis);
> {code}
> The \{{triggerIfValidating}} trigger used here takes the synchronous path,
> \{{validationTrigger.trigger(component)}} →
> \{{component.performValidation()}}, which blocks the calling thread per
> component.
> A 5-thread validation pool already exists:
> {code:java}
> this.validationThreadPool = new FlowEngine(5, "Validate Components", true);
> {code}
> …but it is only used for the periodic re-validation task
> (\{{scheduleWithFixedDelay}}, every 5s). The one-time initial sweep does not
> use it.
> *Proposed change*
> Fan the initial sweep out over the existing {{validationThreadPool}} instead
> of validating one component at a time, then barrier before returning
> (subsequent startup steps assume components have reached a terminal
> validation state).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
