[
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.
h3. h3. 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.
h3. h3. 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).
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()}}.
> 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.
> h3. h3. 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.
> h3. h3. 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)
