@Test(timeout = 10000) public void testStart() throws InterruptedException { final ProcessorThatThrowsExceptionOnScheduled processor = new ProcessorThatThrowsExceptionOnScheduled(); final String uuid = UUID.randomUUID().toString(); ProcessorInitializationContext initContext = new StandardProcessorInitializationContext(uuid, null, null, null, null); processor.initialize(initContext); final ReloadComponent reloadComponent = Mockito.mock(ReloadComponent.class); final BundleCoordinate coordinate = Mockito.mock(BundleCoordinate.class); final LoggableComponent<Processor> loggableComponent = new LoggableComponent<>(processor, coordinate, null); final StandardProcessorNode procNode = new StandardProcessorNode(loggableComponent, uuid, createValidationContextFactory(), null, null, NiFiProperties.createBasicNiFiProperties(null, null), new StandardComponentVariableRegistry(VariableRegistry.EMPTY_REGISTRY), reloadComponent, new SynchronousValidationTrigger()); final ScheduledExecutorService taskScheduler = new FlowEngine(1, "TestClasspathResources", true); final StandardProcessContext processContext = new StandardProcessContext(procNode, null, null, null, () -> false); final SchedulingAgentCallback schedulingAgentCallback = new SchedulingAgentCallback() { @Override public void onTaskComplete() { } @Override public Future<?> scheduleTask(final Callable<?> task) { return taskScheduler.submit(task); } @Override public void trigger() { Assert.fail("Should not have completed"); } }; procNode.performValidation(); procNode.start(taskScheduler, 20000L, processContext, schedulingAgentCallback, true); Thread.sleep(1000L); assertEquals(1, processor.onScheduledCount); assertEquals(1, processor.onUnscheduledCount); assertEquals(1, processor.onStoppedCount); }
@Override public void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final ProcessContext processContext, final SchedulingAgentCallback schedulingAgentCallback, final boolean failIfStopping) { switch (getValidationStatus()) { case INVALID: throw new IllegalStateException("Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); case VALIDATING: throw new IllegalStateException("Processor " + this.getName() + " cannot be started because its validation is still being performed"); } final Processor processor = processorRef.get().getProcessor(); final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); ScheduledState currentState; boolean starting; synchronized (this) { currentState = this.scheduledState.get(); if (currentState == ScheduledState.STOPPED) { starting = this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING); if (starting) { desiredState = ScheduledState.RUNNING; } } else if (currentState == ScheduledState.STOPPING && !failIfStopping) { desiredState = ScheduledState.RUNNING; return; } else { starting = false; } } if (starting) { hasActiveThreads = true; initiateStart(taskScheduler, administrativeYieldMillis, processContext, schedulingAgentCallback); } else { final String procName = processorRef.get().toString(); LOG.warn("Cannot start {} because it is not currently stopped. Current state is {}", procName, currentState); procLog.warn("Cannot start {} because it is not currently stopped. Current state is {}", new Object[] {procName, currentState}); } }
StandardProcessorNode extends ProcessorNode implements Connectable { @Override public void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final ProcessContext processContext, final SchedulingAgentCallback schedulingAgentCallback, final boolean failIfStopping) { switch (getValidationStatus()) { case INVALID: throw new IllegalStateException("Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); case VALIDATING: throw new IllegalStateException("Processor " + this.getName() + " cannot be started because its validation is still being performed"); } final Processor processor = processorRef.get().getProcessor(); final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); ScheduledState currentState; boolean starting; synchronized (this) { currentState = this.scheduledState.get(); if (currentState == ScheduledState.STOPPED) { starting = this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING); if (starting) { desiredState = ScheduledState.RUNNING; } } else if (currentState == ScheduledState.STOPPING && !failIfStopping) { desiredState = ScheduledState.RUNNING; return; } else { starting = false; } } if (starting) { hasActiveThreads = true; initiateStart(taskScheduler, administrativeYieldMillis, processContext, schedulingAgentCallback); } else { final String procName = processorRef.get().toString(); LOG.warn("Cannot start {} because it is not currently stopped. Current state is {}", procName, currentState); procLog.warn("Cannot start {} because it is not currently stopped. Current state is {}", new Object[] {procName, currentState}); } } }
StandardProcessorNode extends ProcessorNode implements Connectable { @Override public void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final ProcessContext processContext, final SchedulingAgentCallback schedulingAgentCallback, final boolean failIfStopping) { switch (getValidationStatus()) { case INVALID: throw new IllegalStateException("Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); case VALIDATING: throw new IllegalStateException("Processor " + this.getName() + " cannot be started because its validation is still being performed"); } final Processor processor = processorRef.get().getProcessor(); final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); ScheduledState currentState; boolean starting; synchronized (this) { currentState = this.scheduledState.get(); if (currentState == ScheduledState.STOPPED) { starting = this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING); if (starting) { desiredState = ScheduledState.RUNNING; } } else if (currentState == ScheduledState.STOPPING && !failIfStopping) { desiredState = ScheduledState.RUNNING; return; } else { starting = false; } } if (starting) { hasActiveThreads = true; initiateStart(taskScheduler, administrativeYieldMillis, processContext, schedulingAgentCallback); } else { final String procName = processorRef.get().toString(); LOG.warn("Cannot start {} because it is not currently stopped. Current state is {}", procName, currentState); procLog.warn("Cannot start {} because it is not currently stopped. Current state is {}", new Object[] {procName, currentState}); } } StandardProcessorNode(final LoggableComponent<Processor> processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider, final NiFiProperties nifiProperties,
                                 final ComponentVariableRegistry variableRegistry, final ReloadComponent reloadComponent, final ValidationTrigger validationTrigger);  StandardProcessorNode(final LoggableComponent<Processor> processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider,
                                 final String componentType, final String componentCanonicalClass, final NiFiProperties nifiProperties,
                                 final ComponentVariableRegistry variableRegistry, final ReloadComponent reloadComponent, final ValidationTrigger validationTrigger,
                                 final boolean isExtensionMissing); }
StandardProcessorNode extends ProcessorNode implements Connectable { @Override public void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final ProcessContext processContext, final SchedulingAgentCallback schedulingAgentCallback, final boolean failIfStopping) { switch (getValidationStatus()) { case INVALID: throw new IllegalStateException("Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); case VALIDATING: throw new IllegalStateException("Processor " + this.getName() + " cannot be started because its validation is still being performed"); } final Processor processor = processorRef.get().getProcessor(); final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); ScheduledState currentState; boolean starting; synchronized (this) { currentState = this.scheduledState.get(); if (currentState == ScheduledState.STOPPED) { starting = this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING); if (starting) { desiredState = ScheduledState.RUNNING; } } else if (currentState == ScheduledState.STOPPING && !failIfStopping) { desiredState = ScheduledState.RUNNING; return; } else { starting = false; } } if (starting) { hasActiveThreads = true; initiateStart(taskScheduler, administrativeYieldMillis, processContext, schedulingAgentCallback); } else { final String procName = processorRef.get().toString(); LOG.warn("Cannot start {} because it is not currently stopped. Current state is {}", procName, currentState); procLog.warn("Cannot start {} because it is not currently stopped. Current state is {}", new Object[] {procName, currentState}); } } StandardProcessorNode(final LoggableComponent<Processor> processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider, final NiFiProperties nifiProperties,
                                 final ComponentVariableRegistry variableRegistry, final ReloadComponent reloadComponent, final ValidationTrigger validationTrigger);  StandardProcessorNode(final LoggableComponent<Processor> processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider,
                                 final String componentType, final String componentCanonicalClass, final NiFiProperties nifiProperties,
                                 final ComponentVariableRegistry variableRegistry, final ReloadComponent reloadComponent, final ValidationTrigger validationTrigger,
                                 final boolean isExtensionMissing); @Override ConfigurableComponent getComponent(); @Override TerminationAwareLogger getLogger(); @Override Object getRunnableComponent(); @Override BundleCoordinate getBundleCoordinate(); @Override String getComments(); @Override Authorizable getParentAuthorizable(); @Override Resource getResource(); @Override boolean isRestricted(); @Override Class<?> getComponentClass(); @Override boolean isDeprecated(); @Override synchronized void setComments(final String comments); @Override Position getPosition(); @Override synchronized void setPosition(final Position position); @Override Map<String, String> getStyle(); @Override synchronized void setStyle(final Map<String, String> style); @Override String getIdentifier(); @Override boolean isLossTolerant(); @Override @SuppressWarnings("deprecation") boolean isIsolated(); @Override boolean isTriggerWhenEmpty(); @Override boolean isSideEffectFree(); @Override boolean isSessionBatchingSupported(); @Override boolean isTriggerWhenAnyDestinationAvailable(); @Override boolean isExecutionNodeRestricted(); @Override synchronized void setLossTolerant(final boolean lossTolerant); @Override boolean isAutoTerminated(final Relationship relationship); @Override void setAutoTerminatedRelationships(final Set<Relationship> terminate); @Override Set<Relationship> getAutoTerminatedRelationships(); String getProcessorDescription(); @Override synchronized void setName(final String name); @Override long getSchedulingPeriod(final TimeUnit timeUnit); @Override boolean isEventDrivenSupported(); @Override synchronized void setSchedulingStrategy(final SchedulingStrategy schedulingStrategy); @Override SchedulingStrategy getSchedulingStrategy(); @Override String getSchedulingPeriod(); @Override @SuppressWarnings("deprecation") synchronized void setScheduldingPeriod(final String schedulingPeriod); @Override synchronized void setExecutionNode(final ExecutionNode executionNode); @Override ExecutionNode getExecutionNode(); @Override long getRunDuration(final TimeUnit timeUnit); @Override synchronized void setRunDuration(final long duration, final TimeUnit timeUnit); @Override long getYieldPeriod(final TimeUnit timeUnit); @Override String getYieldPeriod(); @Override synchronized void setYieldPeriod(final String yieldPeriod); @Override void yield(); @Override void yield(final long period, final TimeUnit timeUnit); @Override long getYieldExpiration(); @Override long getPenalizationPeriod(final TimeUnit timeUnit); @Override String getPenalizationPeriod(); @Override synchronized void setPenalizationPeriod(final String penalizationPeriod); @Override synchronized void setMaxConcurrentTasks(final int taskCount); @Override boolean isTriggeredSerially(); @Override int getMaxConcurrentTasks(); @Override LogLevel getBulletinLevel(); @Override synchronized void setBulletinLevel(final LogLevel level); @Override Set<Connection> getConnections(); @Override List<Connection> getIncomingConnections(); @Override Set<Connection> getConnections(final Relationship relationship); @Override void addConnection(final Connection connection); @Override boolean hasIncomingConnection(); @Override void updateConnection(final Connection connection); @Override void removeConnection(final Connection connection); @Override Relationship getRelationship(final String relationshipName); @Override Processor getProcessor(); @Override synchronized void setProcessor(final LoggableComponent<Processor> processor); @Override synchronized void reload(final Set<URL> additionalUrls); Set<Connectable> getDestinations(); Set<Connectable> getDestinations(final Relationship relationship); Set<Relationship> getUndefinedRelationships(); @Override boolean isRunning(); @Override boolean isValidationNecessary(); @Override int getActiveThreadCount(); @Override Collection<ValidationResult> getValidationErrors(); @Override Requirement getInputRequirement(); @Override boolean equals(final Object other); @Override int hashCode(); @Override Collection<Relationship> getRelationships(); @Override String toString(); @Override ProcessGroup getProcessGroup(); @Override synchronized void setProcessGroup(final ProcessGroup group); @Override void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory); @Override ConnectableType getConnectableType(); @Override void setAnnotationData(final String data); @Override void verifyCanDelete(); @Override void verifyCanDelete(final boolean ignoreConnections); @Override void verifyCanStart(); @Override void verifyCanStart(final Set<ControllerServiceNode> ignoredReferences); @Override void verifyCanStop(); @Override void verifyCanUpdate(); @Override void verifyCanEnable(); @Override void verifyCanDisable(); @Override void verifyCanClearState(); @Override void verifyModifiable(); @Override void enable(); @Override void disable(); @Override void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final ProcessContext processContext,
            final SchedulingAgentCallback schedulingAgentCallback, final boolean failIfStopping); @Override synchronized List<ActiveThreadInfo> getActiveThreads(); @Override synchronized int getTerminatedThreadCount(); @Override int terminate(); @Override boolean isTerminated(final Thread thread); @Override void verifyCanTerminate(); @Override CompletableFuture<Void> stop(final ProcessScheduler processScheduler, final ScheduledExecutorService executor, final ProcessContext processContext,
            final SchedulingAgent schedulingAgent, final LifecycleState scheduleState); @Override String getProcessGroupIdentifier(); @Override Optional<String> getVersionedComponentId(); @Override void setVersionedComponentId(final String versionedComponentId); @Override void setProperties(Map<String, String> properties, boolean allowRemovalOfRequiredProperties); }
StandardProcessorNode extends ProcessorNode implements Connectable { @Override public void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final ProcessContext processContext, final SchedulingAgentCallback schedulingAgentCallback, final boolean failIfStopping) { switch (getValidationStatus()) { case INVALID: throw new IllegalStateException("Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); case VALIDATING: throw new IllegalStateException("Processor " + this.getName() + " cannot be started because its validation is still being performed"); } final Processor processor = processorRef.get().getProcessor(); final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); ScheduledState currentState; boolean starting; synchronized (this) { currentState = this.scheduledState.get(); if (currentState == ScheduledState.STOPPED) { starting = this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING); if (starting) { desiredState = ScheduledState.RUNNING; } } else if (currentState == ScheduledState.STOPPING && !failIfStopping) { desiredState = ScheduledState.RUNNING; return; } else { starting = false; } } if (starting) { hasActiveThreads = true; initiateStart(taskScheduler, administrativeYieldMillis, processContext, schedulingAgentCallback); } else { final String procName = processorRef.get().toString(); LOG.warn("Cannot start {} because it is not currently stopped. Current state is {}", procName, currentState); procLog.warn("Cannot start {} because it is not currently stopped. Current state is {}", new Object[] {procName, currentState}); } } StandardProcessorNode(final LoggableComponent<Processor> processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider, final NiFiProperties nifiProperties,
                                 final ComponentVariableRegistry variableRegistry, final ReloadComponent reloadComponent, final ValidationTrigger validationTrigger);  StandardProcessorNode(final LoggableComponent<Processor> processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider,
                                 final String componentType, final String componentCanonicalClass, final NiFiProperties nifiProperties,
                                 final ComponentVariableRegistry variableRegistry, final ReloadComponent reloadComponent, final ValidationTrigger validationTrigger,
                                 final boolean isExtensionMissing); @Override ConfigurableComponent getComponent(); @Override TerminationAwareLogger getLogger(); @Override Object getRunnableComponent(); @Override BundleCoordinate getBundleCoordinate(); @Override String getComments(); @Override Authorizable getParentAuthorizable(); @Override Resource getResource(); @Override boolean isRestricted(); @Override Class<?> getComponentClass(); @Override boolean isDeprecated(); @Override synchronized void setComments(final String comments); @Override Position getPosition(); @Override synchronized void setPosition(final Position position); @Override Map<String, String> getStyle(); @Override synchronized void setStyle(final Map<String, String> style); @Override String getIdentifier(); @Override boolean isLossTolerant(); @Override @SuppressWarnings("deprecation") boolean isIsolated(); @Override boolean isTriggerWhenEmpty(); @Override boolean isSideEffectFree(); @Override boolean isSessionBatchingSupported(); @Override boolean isTriggerWhenAnyDestinationAvailable(); @Override boolean isExecutionNodeRestricted(); @Override synchronized void setLossTolerant(final boolean lossTolerant); @Override boolean isAutoTerminated(final Relationship relationship); @Override void setAutoTerminatedRelationships(final Set<Relationship> terminate); @Override Set<Relationship> getAutoTerminatedRelationships(); String getProcessorDescription(); @Override synchronized void setName(final String name); @Override long getSchedulingPeriod(final TimeUnit timeUnit); @Override boolean isEventDrivenSupported(); @Override synchronized void setSchedulingStrategy(final SchedulingStrategy schedulingStrategy); @Override SchedulingStrategy getSchedulingStrategy(); @Override String getSchedulingPeriod(); @Override @SuppressWarnings("deprecation") synchronized void setScheduldingPeriod(final String schedulingPeriod); @Override synchronized void setExecutionNode(final ExecutionNode executionNode); @Override ExecutionNode getExecutionNode(); @Override long getRunDuration(final TimeUnit timeUnit); @Override synchronized void setRunDuration(final long duration, final TimeUnit timeUnit); @Override long getYieldPeriod(final TimeUnit timeUnit); @Override String getYieldPeriod(); @Override synchronized void setYieldPeriod(final String yieldPeriod); @Override void yield(); @Override void yield(final long period, final TimeUnit timeUnit); @Override long getYieldExpiration(); @Override long getPenalizationPeriod(final TimeUnit timeUnit); @Override String getPenalizationPeriod(); @Override synchronized void setPenalizationPeriod(final String penalizationPeriod); @Override synchronized void setMaxConcurrentTasks(final int taskCount); @Override boolean isTriggeredSerially(); @Override int getMaxConcurrentTasks(); @Override LogLevel getBulletinLevel(); @Override synchronized void setBulletinLevel(final LogLevel level); @Override Set<Connection> getConnections(); @Override List<Connection> getIncomingConnections(); @Override Set<Connection> getConnections(final Relationship relationship); @Override void addConnection(final Connection connection); @Override boolean hasIncomingConnection(); @Override void updateConnection(final Connection connection); @Override void removeConnection(final Connection connection); @Override Relationship getRelationship(final String relationshipName); @Override Processor getProcessor(); @Override synchronized void setProcessor(final LoggableComponent<Processor> processor); @Override synchronized void reload(final Set<URL> additionalUrls); Set<Connectable> getDestinations(); Set<Connectable> getDestinations(final Relationship relationship); Set<Relationship> getUndefinedRelationships(); @Override boolean isRunning(); @Override boolean isValidationNecessary(); @Override int getActiveThreadCount(); @Override Collection<ValidationResult> getValidationErrors(); @Override Requirement getInputRequirement(); @Override boolean equals(final Object other); @Override int hashCode(); @Override Collection<Relationship> getRelationships(); @Override String toString(); @Override ProcessGroup getProcessGroup(); @Override synchronized void setProcessGroup(final ProcessGroup group); @Override void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory); @Override ConnectableType getConnectableType(); @Override void setAnnotationData(final String data); @Override void verifyCanDelete(); @Override void verifyCanDelete(final boolean ignoreConnections); @Override void verifyCanStart(); @Override void verifyCanStart(final Set<ControllerServiceNode> ignoredReferences); @Override void verifyCanStop(); @Override void verifyCanUpdate(); @Override void verifyCanEnable(); @Override void verifyCanDisable(); @Override void verifyCanClearState(); @Override void verifyModifiable(); @Override void enable(); @Override void disable(); @Override void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final ProcessContext processContext,
            final SchedulingAgentCallback schedulingAgentCallback, final boolean failIfStopping); @Override synchronized List<ActiveThreadInfo> getActiveThreads(); @Override synchronized int getTerminatedThreadCount(); @Override int terminate(); @Override boolean isTerminated(final Thread thread); @Override void verifyCanTerminate(); @Override CompletableFuture<Void> stop(final ProcessScheduler processScheduler, final ScheduledExecutorService executor, final ProcessContext processContext,
            final SchedulingAgent schedulingAgent, final LifecycleState scheduleState); @Override String getProcessGroupIdentifier(); @Override Optional<String> getVersionedComponentId(); @Override void setVersionedComponentId(final String versionedComponentId); @Override void setProperties(Map<String, String> properties, boolean allowRemovalOfRequiredProperties); static final String BULLETIN_OBSERVER_ID; static final TimeUnit DEFAULT_TIME_UNIT; static final String DEFAULT_YIELD_PERIOD; static final String DEFAULT_PENALIZATION_PERIOD; }