[
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).
Concretely:
- Add an optional \{{ExecutorService}} to \{{TriggerValidationTask}}. When
\{{null}}, keep the current serial behavior (so the periodic task is
unchanged); when provided, \{{submit()}} each component's validation and then
\{{Future.get()}} every future — a submit-all / get-all barrier.
- The barrier must run on the calling (startup) thread, not on a pool worker,
so waiting never consumes a validation worker. Since the periodic task is
scheduled only after this sweep, the pool is otherwise idle during
initialization — no self-starvation.
- Collect per-component failures (\{{ExecutionException}}) and log them without
aborting the remaining validations, preserving the current "validate
everything, log errors" semantics.
- In \{{FlowController.initializeFlow()}}, pass \{{validationThreadPool}} to
the initial-sweep instance only.
- Note the \{{getAllConnectors()}} case (added in 2.10):
\{{connector.validateComponents(...)}} has a different shape than the
per-\{{ComponentNode}} trigger and would need to be included in the
parallelized set (or handled alongside it).
h3. Correctness considerations
- Each \{{ComponentNode}} maintains its own validation state, so validating
distinct components concurrently is independent.
- Shared structures written during validation (e.g. \{{RuleViolationsManager}})
should be confirmed thread-safe / component-id-keyed as part of review.
- A barrier (wait for all futures) preserves the existing invariant that all
components are validated before \{{startProcessing()}}.
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.
*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).
> 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).
>
> Concretely:
> - Add an optional \{{ExecutorService}} to \{{TriggerValidationTask}}. When
> \{{null}}, keep the current serial behavior (so the periodic task is
> unchanged); when provided, \{{submit()}} each component's validation and then
> \{{Future.get()}} every future — a submit-all / get-all barrier.
> - The barrier must run on the calling (startup) thread, not on a pool worker,
> so waiting never consumes a validation worker. Since the periodic task is
> scheduled only after this sweep, the pool is otherwise idle during
> initialization — no self-starvation.
> - Collect per-component failures (\{{ExecutionException}}) and log them
> without aborting the remaining validations, preserving the current "validate
> everything, log errors" semantics.
> - In \{{FlowController.initializeFlow()}}, pass \{{validationThreadPool}} to
> the initial-sweep instance only.
> - Note the \{{getAllConnectors()}} case (added in 2.10):
> \{{connector.validateComponents(...)}} has a different shape than the
> per-\{{ComponentNode}} trigger and would need to be included in the
> parallelized set (or handled alongside it).
> h3. Correctness considerations
> - Each \{{ComponentNode}} maintains its own validation state, so validating
> distinct components concurrently is independent.
> - Shared structures written during validation (e.g.
> \{{RuleViolationsManager}}) should be confirmed thread-safe /
> component-id-keyed as part of review.
> - A barrier (wait for all futures) preserves the existing invariant that all
> components are validated before \{{startProcessing()}}.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
