diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/ParallelTriggerValidationTask.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/ParallelTriggerValidationTask.java new file mode 100644 index 000000000000..c455eb9fd4b2 --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/ParallelTriggerValidationTask.java @@ -0,0 +1,155 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.components.validation; + +import org.apache.nifi.components.connector.ConnectorNode; +import org.apache.nifi.controller.ComponentNode; +import org.apache.nifi.controller.flow.FlowManager; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; + +/** + * Triggers validation of every component and Connector by submitting each one to an {@link ExecutorService} and + * waiting for all of them to complete, instead of validating one at a time on the calling thread. Each + * {@link ComponentNode} tracks its own validation state independently, so validating distinct components + * concurrently is safe. + *

+ * This is intended for the one-time initial validation sweep performed when a flow is loaded during + * {@code FlowController} initialization, where the number of components can be large and validating them one at a + * time on a single thread can noticeably slow down startup. The periodic re-validation performed thereafter + * continues to use the serial {@link TriggerValidationTask}. + */ +public class ParallelTriggerValidationTask implements Runnable { + private static final Logger logger = LoggerFactory.getLogger(ParallelTriggerValidationTask.class); + + private final FlowManager flowManager; + private final ValidationTrigger validationTrigger; + private final ExecutorService executorService; + private volatile boolean completed = false; + + public ParallelTriggerValidationTask(final FlowManager flowManager, final ValidationTrigger validationTrigger, final ExecutorService executorService) { + this.flowManager = Objects.requireNonNull(flowManager, "FlowManager is required"); + this.validationTrigger = Objects.requireNonNull(validationTrigger, "ValidationTrigger is required"); + this.executorService = Objects.requireNonNull(executorService, "ExecutorService is required"); + } + + /** + * @return true if the most recent call to {@link #run()} triggered validation of every component + * and waited for it to complete; false if that call was interrupted, or the + * {@link ExecutorService} rejected work, before validation of every component could be triggered and + * completed. + */ + public boolean isValidationComplete() { + return completed; + } + + @Override + public void run() { + completed = false; + + try { + logger.debug("Triggering validation of all components in parallel"); + + final List nodes = ValidatableComponents.getComponentNodes(flowManager); + final List connectors = ValidatableComponents.getConnectors(flowManager); + + completed = triggerInParallel(nodes, connectors); + } catch (final Throwable t) { + logger.error("Encountered unexpected error when attempting to validate components", t); + } + } + + /** + * Submits validation of every given component and Connector to the executor and waits for all of it to + * complete before returning. + * + * @return true if every component and Connector was successfully submitted for validation and + * validation of all of them completed; false otherwise. + */ + private boolean triggerInParallel(final List nodes, final List connectors) { + final List> futures = new ArrayList<>(nodes.size() + connectors.size()); + boolean allSubmitted = true; + + for (final ComponentNode node : nodes) { + if (!submit(futures, () -> validationTrigger.trigger(node))) { + allSubmitted = false; + break; + } + } + + if (allSubmitted) { + for (final ConnectorNode connector : connectors) { + if (!submit(futures, () -> connector.validateComponents(validationTrigger))) { + allSubmitted = false; + break; + } + } + } + + final boolean allAwaited = awaitAll(futures); + return allSubmitted && allAwaited; + } + + private boolean submit(final List> futures, final Runnable task) { + try { + futures.add(executorService.submit(task)); + return true; + } catch (final RejectedExecutionException e) { + logger.warn("Validation thread pool rejected further work while triggering initial validation; not all components were submitted for validation"); + return false; + } + } + + /** + * Waits for every given Future to complete. If interrupted while waiting, does not block any further: + * the remaining, not-yet-started Futures are cancelled and this method returns immediately, so that shutdown + * is not delayed. + * + * @return true if every Future completed (whether successfully or with an exception); + * false if interrupted before all of them completed. + */ + private boolean awaitAll(final List> futures) { + for (int i = 0; i < futures.size(); i++) { + try { + futures.get(i).get(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + logger.warn("Interrupted while waiting for initial component validation to complete; {} of {} validation tasks had not yet finished", + futures.size() - i, futures.size()); + + for (int j = i; j < futures.size(); j++) { + futures.get(j).cancel(false); + } + + return false; + } catch (final ExecutionException e) { + logger.error("Failed to validate a component during initial validation", e.getCause()); + } + } + + return true; + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/TriggerValidationTask.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/TriggerValidationTask.java index d05033a8b4d3..cca8749afa65 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/TriggerValidationTask.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/TriggerValidationTask.java @@ -39,31 +39,11 @@ public void run() { try { logger.debug("Triggering validation of all components"); - for (final ComponentNode node : flowManager.getAllControllerServices()) { + for (final ComponentNode node : ValidatableComponents.getComponentNodes(flowManager)) { 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()) { + for (final ConnectorNode connector : ValidatableComponents.getConnectors(flowManager)) { connector.validateComponents(validationTrigger); } } catch (final Throwable t) { diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/ValidatableComponents.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/ValidatableComponents.java new file mode 100644 index 000000000000..ca16a0b22acf --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/components/validation/ValidatableComponents.java @@ -0,0 +1,51 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.components.validation; + +import org.apache.nifi.components.connector.ConnectorNode; +import org.apache.nifi.controller.ComponentNode; +import org.apache.nifi.controller.flow.FlowManager; + +import java.util.ArrayList; +import java.util.List; + +/** + * Gathers the components subject to the initial and periodic validation sweeps, so that + * {@link TriggerValidationTask} and {@link ParallelTriggerValidationTask} validate exactly the same set of + * components without duplicating the collection logic in both places. + */ +final class ValidatableComponents { + + private ValidatableComponents() { + } + + static List getComponentNodes(final FlowManager flowManager) { + final List nodes = new ArrayList<>(); + nodes.addAll(flowManager.getAllControllerServices()); + nodes.addAll(flowManager.getAllReportingTasks()); + nodes.addAll(flowManager.getAllFlowAnalysisRules()); + nodes.addAll(flowManager.getAllParameterProviders()); + nodes.addAll(flowManager.getRootGroup().findAllProcessors()); + nodes.addAll(flowManager.getAllFlowRegistryClients()); + return nodes; + } + + static List getConnectors(final FlowManager flowManager) { + return new ArrayList<>(flowManager.getAllConnectors()); + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java index 4953965ff887..ec02e4628b94 100644 --- a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java @@ -69,6 +69,7 @@ import org.apache.nifi.components.monitor.LongRunningTaskMonitor; import org.apache.nifi.components.state.StateManagerProvider; import org.apache.nifi.components.state.StateProvider; +import org.apache.nifi.components.validation.ParallelTriggerValidationTask; import org.apache.nifi.components.validation.StandardValidationTrigger; import org.apache.nifi.components.validation.StandardVerifiableComponentFactory; import org.apache.nifi.components.validation.TriggerValidationTask; @@ -1515,10 +1516,16 @@ public void trigger(final ComponentNode component) { if (flowAnalyzer != null) { new TriggerFlowAnalysisTask(flowAnalyzer, rootProcessGroupSupplier).run(); } - new TriggerValidationTask(flowManager, triggerIfValidating).run(); + final ParallelTriggerValidationTask initialValidationTask = new ParallelTriggerValidationTask(flowManager, triggerIfValidating, validationThreadPool); + initialValidationTask.run(); final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); - LOG.info("Performed initial validation of all components in {} milliseconds", millis); + if (initialValidationTask.isValidationComplete()) { + LOG.info("Performed initial validation of all components in {} milliseconds", millis); + } else { + LOG.warn("Initial validation of components did not complete for all components after {} milliseconds " + + "(interrupted or validation thread pool unavailable); some components may still report a VALIDATING status", millis); + } scheduleBackgroundFlowAnalysis(rootProcessGroupSupplier); // Trigger component validation to occur every 5 seconds. diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/validation/ParallelTriggerValidationTaskTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/validation/ParallelTriggerValidationTaskTest.java new file mode 100644 index 000000000000..9453e7fd9920 --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/validation/ParallelTriggerValidationTaskTest.java @@ -0,0 +1,205 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.components.validation; + +import org.apache.nifi.components.connector.ConnectorNode; +import org.apache.nifi.controller.ComponentNode; +import org.apache.nifi.controller.FlowAnalysisRuleNode; +import org.apache.nifi.controller.ParameterProviderNode; +import org.apache.nifi.controller.ProcessorNode; +import org.apache.nifi.controller.ReportingTaskNode; +import org.apache.nifi.controller.flow.FlowManager; +import org.apache.nifi.controller.service.ControllerServiceNode; +import org.apache.nifi.groups.ProcessGroup; +import org.apache.nifi.registry.flow.FlowRegistryClientNode; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class ParallelTriggerValidationTaskTest { + + private ExecutorService executorService; + + @AfterEach + void shutdownExecutor() { + if (executorService != null) { + executorService.shutdownNow(); + } + } + + @Test + void testConstructorRequiresExecutorService() { + final Fixture fixture = new Fixture(); + final ValidationTrigger trigger = recordingTrigger(new CopyOnWriteArrayList<>()); + + assertThrows(NullPointerException.class, () -> new ParallelTriggerValidationTask(fixture.flowManager, trigger, null)); + } + + @Test + void testRunTriggersEveryComponentAndCompletes() throws InterruptedException { + final Fixture fixture = new Fixture(); + final List triggered = new CopyOnWriteArrayList<>(); + final ValidationTrigger trigger = recordingTrigger(triggered); + executorService = Executors.newFixedThreadPool(4); + + final ParallelTriggerValidationTask task = new ParallelTriggerValidationTask(fixture.flowManager, trigger, executorService); + task.run(); + + assertTrue(task.isValidationComplete()); + assertEquals(new HashSet<>(fixture.expectedComponentNodes()), new HashSet<>(triggered)); + verify(fixture.connector, times(1)).validateComponents(trigger); + } + + @Test + void testRunReportsIncompleteWhenExecutorRejectsWork() { + final Fixture fixture = new Fixture(); + final List triggered = new CopyOnWriteArrayList<>(); + final ValidationTrigger trigger = recordingTrigger(triggered); + + executorService = Executors.newFixedThreadPool(2); + executorService.shutdown(); // no longer accepts new work; submit() will throw RejectedExecutionException + + final ParallelTriggerValidationTask task = new ParallelTriggerValidationTask(fixture.flowManager, trigger, executorService); + task.run(); + + assertFalse(task.isValidationComplete()); + } + + @Test + void testRunReportsIncompleteWhenInterruptedWhileAwaitingCompletion() throws InterruptedException { + final FlowManager flowManager = mock(FlowManager.class); + final ProcessGroup rootGroup = mock(ProcessGroup.class); + final ProcessorNode slowProcessor = mock(ProcessorNode.class); + + when(flowManager.getRootGroup()).thenReturn(rootGroup); + when(rootGroup.findAllProcessors()).thenReturn(List.of(slowProcessor)); + when(flowManager.getAllControllerServices()).thenReturn(Collections.emptySet()); + when(flowManager.getAllReportingTasks()).thenReturn(Collections.emptySet()); + when(flowManager.getAllFlowAnalysisRules()).thenReturn(Collections.emptySet()); + when(flowManager.getAllParameterProviders()).thenReturn(Collections.emptySet()); + when(flowManager.getAllFlowRegistryClients()).thenReturn(Collections.emptySet()); + when(flowManager.getAllConnectors()).thenReturn(Collections.emptyList()); + + final CountDownLatch validationStarted = new CountDownLatch(1); + final CountDownLatch releaseValidation = new CountDownLatch(1); + final ValidationTrigger trigger = new ValidationTrigger() { + @Override + public void triggerAsync(final ComponentNode component) { + } + + @Override + public void trigger(final ComponentNode component) { + validationStarted.countDown(); + try { + // Simulate a component validation that is still in progress when the calling + // thread is interrupted; the pool thread itself is not interrupted. + releaseValidation.await(); + } catch (final InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + }; + + executorService = Executors.newSingleThreadExecutor(); + final ParallelTriggerValidationTask task = new ParallelTriggerValidationTask(flowManager, trigger, executorService); + + final AtomicBoolean interruptStatusRestored = new AtomicBoolean(false); + final Thread runner = new Thread(() -> { + task.run(); + interruptStatusRestored.set(Thread.currentThread().isInterrupted()); + }); + + try { + runner.start(); + assertTrue(validationStarted.await(5, TimeUnit.SECONDS), "Validation did not start within the timeout"); + + runner.interrupt(); + runner.join(5_000); + + assertFalse(runner.isAlive(), "ParallelTriggerValidationTask did not return promptly after being interrupted"); + assertFalse(task.isValidationComplete()); + assertTrue(interruptStatusRestored.get(), "Interrupt status should be restored on the thread that called run()"); + } finally { + releaseValidation.countDown(); + } + } + + private static ValidationTrigger recordingTrigger(final List triggered) { + return new ValidationTrigger() { + @Override + public void triggerAsync(final ComponentNode component) { + triggered.add(component); + } + + @Override + public void trigger(final ComponentNode component) { + triggered.add(component); + } + }; + } + + /** + * Provides a FlowManager mock with exactly one component of each type that ParallelTriggerValidationTask is + * expected to validate, so tests can assert that every category is covered. + */ + private static final class Fixture { + private final FlowManager flowManager = mock(FlowManager.class); + private final ControllerServiceNode controllerService = mock(ControllerServiceNode.class); + private final ReportingTaskNode reportingTask = mock(ReportingTaskNode.class); + private final FlowAnalysisRuleNode flowAnalysisRule = mock(FlowAnalysisRuleNode.class); + private final ParameterProviderNode parameterProvider = mock(ParameterProviderNode.class); + private final FlowRegistryClientNode flowRegistryClient = mock(FlowRegistryClientNode.class); + private final ProcessorNode processor = mock(ProcessorNode.class); + private final ConnectorNode connector = mock(ConnectorNode.class); + private final ProcessGroup rootGroup = mock(ProcessGroup.class); + + private Fixture() { + when(flowManager.getAllControllerServices()).thenReturn(Set.of(controllerService)); + when(flowManager.getAllReportingTasks()).thenReturn(Set.of(reportingTask)); + when(flowManager.getAllFlowAnalysisRules()).thenReturn(Set.of(flowAnalysisRule)); + when(flowManager.getAllParameterProviders()).thenReturn(Set.of(parameterProvider)); + when(flowManager.getAllFlowRegistryClients()).thenReturn(Set.of(flowRegistryClient)); + when(flowManager.getAllConnectors()).thenReturn(List.of(connector)); + when(flowManager.getRootGroup()).thenReturn(rootGroup); + when(rootGroup.findAllProcessors()).thenReturn(List.of(processor)); + } + + private List expectedComponentNodes() { + return List.of(controllerService, reportingTask, flowAnalysisRule, parameterProvider, processor, flowRegistryClient); + } + } +} diff --git a/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/validation/TriggerValidationTaskTest.java b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/validation/TriggerValidationTaskTest.java new file mode 100644 index 000000000000..1fd231e83f03 --- /dev/null +++ b/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/components/validation/TriggerValidationTaskTest.java @@ -0,0 +1,102 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.components.validation; + +import org.apache.nifi.components.connector.ConnectorNode; +import org.apache.nifi.controller.ComponentNode; +import org.apache.nifi.controller.FlowAnalysisRuleNode; +import org.apache.nifi.controller.ParameterProviderNode; +import org.apache.nifi.controller.ProcessorNode; +import org.apache.nifi.controller.ReportingTaskNode; +import org.apache.nifi.controller.flow.FlowManager; +import org.apache.nifi.controller.service.ControllerServiceNode; +import org.apache.nifi.groups.ProcessGroup; +import org.apache.nifi.registry.flow.FlowRegistryClientNode; +import org.junit.jupiter.api.Test; + +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class TriggerValidationTaskTest { + + @Test + void testRunTriggersEveryComponentOnCallingThread() { + final Fixture fixture = new Fixture(); + final List triggered = new CopyOnWriteArrayList<>(); + final ValidationTrigger trigger = recordingTrigger(triggered); + + final TriggerValidationTask task = new TriggerValidationTask(fixture.flowManager, trigger); + task.run(); + + assertEquals(new HashSet<>(fixture.expectedComponentNodes()), new HashSet<>(triggered)); + verify(fixture.connector, times(1)).validateComponents(trigger); + } + + private static ValidationTrigger recordingTrigger(final List triggered) { + return new ValidationTrigger() { + @Override + public void triggerAsync(final ComponentNode component) { + triggered.add(component); + } + + @Override + public void trigger(final ComponentNode component) { + triggered.add(component); + } + }; + } + + /** + * Provides a FlowManager mock with exactly one component of each type that TriggerValidationTask is expected + * to validate, so tests can assert that every category is covered. + */ + private static final class Fixture { + private final FlowManager flowManager = mock(FlowManager.class); + private final ControllerServiceNode controllerService = mock(ControllerServiceNode.class); + private final ReportingTaskNode reportingTask = mock(ReportingTaskNode.class); + private final FlowAnalysisRuleNode flowAnalysisRule = mock(FlowAnalysisRuleNode.class); + private final ParameterProviderNode parameterProvider = mock(ParameterProviderNode.class); + private final FlowRegistryClientNode flowRegistryClient = mock(FlowRegistryClientNode.class); + private final ProcessorNode processor = mock(ProcessorNode.class); + private final ConnectorNode connector = mock(ConnectorNode.class); + private final ProcessGroup rootGroup = mock(ProcessGroup.class); + + private Fixture() { + when(flowManager.getAllControllerServices()).thenReturn(Set.of(controllerService)); + when(flowManager.getAllReportingTasks()).thenReturn(Set.of(reportingTask)); + when(flowManager.getAllFlowAnalysisRules()).thenReturn(Set.of(flowAnalysisRule)); + when(flowManager.getAllParameterProviders()).thenReturn(Set.of(parameterProvider)); + when(flowManager.getAllFlowRegistryClients()).thenReturn(Set.of(flowRegistryClient)); + when(flowManager.getAllConnectors()).thenReturn(List.of(connector)); + when(flowManager.getRootGroup()).thenReturn(rootGroup); + when(rootGroup.findAllProcessors()).thenReturn(List.of(processor)); + } + + private List expectedComponentNodes() { + return List.of(controllerService, reportingTask, flowAnalysisRule, parameterProvider, processor, flowRegistryClient); + } + } +}