@Test(timeout = 10000) public void testProcessorTimeOutNoResponseToInterrupt() throws InterruptedException { final FailOnScheduledProcessor proc = new FailOnScheduledProcessor(); proc.setDesiredFailureCount(0); proc.setOnScheduledSleepDuration(20, TimeUnit.MINUTES, false, 1); proc.initialize(new StandardProcessorInitializationContext(UUID.randomUUID().toString(), null, null, null, nifiProperties)); final ReloadComponent reloadComponent = Mockito.mock(ReloadComponent.class); final LoggableComponent<Processor> loggableComponent = new LoggableComponent<>(proc, systemBundle.getBundleDetails().getCoordinate(), null); final ProcessorNode procNode = new StandardProcessorNode(loggableComponent, UUID.randomUUID().toString(), new StandardValidationContextFactory(controller, variableRegistry), scheduler, controller, nifiProperties, new StandardComponentVariableRegistry(VariableRegistry.EMPTY_REGISTRY), reloadComponent, new SynchronousValidationTrigger()); rootGroup.addProcessor(procNode); procNode.performValidation(); scheduler.startProcessor(procNode, true); Thread.sleep(100L); assertEquals(1, proc.getOnScheduledInvocationCount()); Thread.sleep(100L); assertEquals(1, proc.getOnScheduledInvocationCount()); proc.setAllowSleepInterrupt(true); }
@Override public synchronized CompletableFuture<Void> startProcessor(final ProcessorNode procNode, final boolean failIfStopping) { final LifecycleState lifecycleState = getLifecycleState(requireNonNull(procNode), true); final StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider, this.encryptor, getStateManager(procNode.getIdentifier()), lifecycleState::isTerminated); final CompletableFuture<Void> future = new CompletableFuture<>(); final SchedulingAgentCallback callback = new SchedulingAgentCallback() { @Override public void trigger() { lifecycleState.clearTerminationFlag(); getSchedulingAgent(procNode).schedule(procNode, lifecycleState); future.complete(null); } @Override public Future<?> scheduleTask(final Callable<?> task) { lifecycleState.incrementActiveThreadCount(null); return componentLifeCycleThreadPool.submit(task); } @Override public void onTaskComplete() { lifecycleState.decrementActiveThreadCount(null); } }; LOG.info("Starting {}", procNode); procNode.start(this.componentMonitoringThreadPool, this.administrativeYieldMillis, processContext, callback, failIfStopping); return future; }
StandardProcessScheduler implements ProcessScheduler { @Override public synchronized CompletableFuture<Void> startProcessor(final ProcessorNode procNode, final boolean failIfStopping) { final LifecycleState lifecycleState = getLifecycleState(requireNonNull(procNode), true); final StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider, this.encryptor, getStateManager(procNode.getIdentifier()), lifecycleState::isTerminated); final CompletableFuture<Void> future = new CompletableFuture<>(); final SchedulingAgentCallback callback = new SchedulingAgentCallback() { @Override public void trigger() { lifecycleState.clearTerminationFlag(); getSchedulingAgent(procNode).schedule(procNode, lifecycleState); future.complete(null); } @Override public Future<?> scheduleTask(final Callable<?> task) { lifecycleState.incrementActiveThreadCount(null); return componentLifeCycleThreadPool.submit(task); } @Override public void onTaskComplete() { lifecycleState.decrementActiveThreadCount(null); } }; LOG.info("Starting {}", procNode); procNode.start(this.componentMonitoringThreadPool, this.administrativeYieldMillis, processContext, callback, failIfStopping); return future; } }
StandardProcessScheduler implements ProcessScheduler { @Override public synchronized CompletableFuture<Void> startProcessor(final ProcessorNode procNode, final boolean failIfStopping) { final LifecycleState lifecycleState = getLifecycleState(requireNonNull(procNode), true); final StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider, this.encryptor, getStateManager(procNode.getIdentifier()), lifecycleState::isTerminated); final CompletableFuture<Void> future = new CompletableFuture<>(); final SchedulingAgentCallback callback = new SchedulingAgentCallback() { @Override public void trigger() { lifecycleState.clearTerminationFlag(); getSchedulingAgent(procNode).schedule(procNode, lifecycleState); future.complete(null); } @Override public Future<?> scheduleTask(final Callable<?> task) { lifecycleState.incrementActiveThreadCount(null); return componentLifeCycleThreadPool.submit(task); } @Override public void onTaskComplete() { lifecycleState.decrementActiveThreadCount(null); } }; LOG.info("Starting {}", procNode); procNode.start(this.componentMonitoringThreadPool, this.administrativeYieldMillis, processContext, callback, failIfStopping); return future; } StandardProcessScheduler(final FlowEngine componentLifecycleThreadPool, final FlowController flowController, final StringEncryptor encryptor,
        final StateManagerProvider stateManagerProvider, final NiFiProperties nifiProperties); }
StandardProcessScheduler implements ProcessScheduler { @Override public synchronized CompletableFuture<Void> startProcessor(final ProcessorNode procNode, final boolean failIfStopping) { final LifecycleState lifecycleState = getLifecycleState(requireNonNull(procNode), true); final StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider, this.encryptor, getStateManager(procNode.getIdentifier()), lifecycleState::isTerminated); final CompletableFuture<Void> future = new CompletableFuture<>(); final SchedulingAgentCallback callback = new SchedulingAgentCallback() { @Override public void trigger() { lifecycleState.clearTerminationFlag(); getSchedulingAgent(procNode).schedule(procNode, lifecycleState); future.complete(null); } @Override public Future<?> scheduleTask(final Callable<?> task) { lifecycleState.incrementActiveThreadCount(null); return componentLifeCycleThreadPool.submit(task); } @Override public void onTaskComplete() { lifecycleState.decrementActiveThreadCount(null); } }; LOG.info("Starting {}", procNode); procNode.start(this.componentMonitoringThreadPool, this.administrativeYieldMillis, processContext, callback, failIfStopping); return future; } StandardProcessScheduler(final FlowEngine componentLifecycleThreadPool, final FlowController flowController, final StringEncryptor encryptor,
        final StateManagerProvider stateManagerProvider, final NiFiProperties nifiProperties); void scheduleFrameworkTask(final Runnable command, final String taskName, final long initialDelay, final long delay, final TimeUnit timeUnit); Future<?> submitFrameworkTask(final Runnable task); @Override void setMaxThreadCount(final SchedulingStrategy schedulingStrategy, final int maxThreadCount); void setSchedulingAgent(final SchedulingStrategy strategy, final SchedulingAgent agent); SchedulingAgent getSchedulingAgent(final SchedulingStrategy strategy); @Override void shutdown(); @Override void schedule(final ReportingTaskNode taskNode); @Override void unschedule(final ReportingTaskNode taskNode); @Override synchronized CompletableFuture<Void> startProcessor(final ProcessorNode procNode, final boolean failIfStopping); @Override synchronized CompletableFuture<Void> stopProcessor(final ProcessorNode procNode); @Override synchronized void terminateProcessor(final ProcessorNode procNode); @Override void onProcessorRemoved(final ProcessorNode procNode); @Override void onPortRemoved(final Port port); @Override void onFunnelRemoved(Funnel funnel); @Override void onReportingTaskRemoved(final ReportingTaskNode reportingTask); @Override void yield(final ProcessorNode procNode); @Override void registerEvent(final Connectable worker); @Override int getActiveThreadCount(final Object scheduled); @Override void startPort(final Port port); @Override void startFunnel(final Funnel funnel); @Override void stopPort(final Port port); @Override void stopFunnel(final Funnel funnel); @Override synchronized void enableFunnel(final Funnel funnel); @Override synchronized void disableFunnel(final Funnel funnel); @Override synchronized void disablePort(final Port port); @Override synchronized void enablePort(final Port port); @Override synchronized void enableProcessor(final ProcessorNode procNode); @Override synchronized void disableProcessor(final ProcessorNode procNode); synchronized void enableReportingTask(final ReportingTaskNode taskNode); synchronized void disableReportingTask(final ReportingTaskNode taskNode); @Override boolean isScheduled(final Object scheduled); @Override CompletableFuture<Void> enableControllerService(final ControllerServiceNode service); @Override CompletableFuture<Void> disableControllerService(final ControllerServiceNode service); @Override CompletableFuture<Void> disableControllerServices(final List<ControllerServiceNode> services); }
StandardProcessScheduler implements ProcessScheduler { @Override public synchronized CompletableFuture<Void> startProcessor(final ProcessorNode procNode, final boolean failIfStopping) { final LifecycleState lifecycleState = getLifecycleState(requireNonNull(procNode), true); final StandardProcessContext processContext = new StandardProcessContext(procNode, this.controllerServiceProvider, this.encryptor, getStateManager(procNode.getIdentifier()), lifecycleState::isTerminated); final CompletableFuture<Void> future = new CompletableFuture<>(); final SchedulingAgentCallback callback = new SchedulingAgentCallback() { @Override public void trigger() { lifecycleState.clearTerminationFlag(); getSchedulingAgent(procNode).schedule(procNode, lifecycleState); future.complete(null); } @Override public Future<?> scheduleTask(final Callable<?> task) { lifecycleState.incrementActiveThreadCount(null); return componentLifeCycleThreadPool.submit(task); } @Override public void onTaskComplete() { lifecycleState.decrementActiveThreadCount(null); } }; LOG.info("Starting {}", procNode); procNode.start(this.componentMonitoringThreadPool, this.administrativeYieldMillis, processContext, callback, failIfStopping); return future; } StandardProcessScheduler(final FlowEngine componentLifecycleThreadPool, final FlowController flowController, final StringEncryptor encryptor,
        final StateManagerProvider stateManagerProvider, final NiFiProperties nifiProperties); void scheduleFrameworkTask(final Runnable command, final String taskName, final long initialDelay, final long delay, final TimeUnit timeUnit); Future<?> submitFrameworkTask(final Runnable task); @Override void setMaxThreadCount(final SchedulingStrategy schedulingStrategy, final int maxThreadCount); void setSchedulingAgent(final SchedulingStrategy strategy, final SchedulingAgent agent); SchedulingAgent getSchedulingAgent(final SchedulingStrategy strategy); @Override void shutdown(); @Override void schedule(final ReportingTaskNode taskNode); @Override void unschedule(final ReportingTaskNode taskNode); @Override synchronized CompletableFuture<Void> startProcessor(final ProcessorNode procNode, final boolean failIfStopping); @Override synchronized CompletableFuture<Void> stopProcessor(final ProcessorNode procNode); @Override synchronized void terminateProcessor(final ProcessorNode procNode); @Override void onProcessorRemoved(final ProcessorNode procNode); @Override void onPortRemoved(final Port port); @Override void onFunnelRemoved(Funnel funnel); @Override void onReportingTaskRemoved(final ReportingTaskNode reportingTask); @Override void yield(final ProcessorNode procNode); @Override void registerEvent(final Connectable worker); @Override int getActiveThreadCount(final Object scheduled); @Override void startPort(final Port port); @Override void startFunnel(final Funnel funnel); @Override void stopPort(final Port port); @Override void stopFunnel(final Funnel funnel); @Override synchronized void enableFunnel(final Funnel funnel); @Override synchronized void disableFunnel(final Funnel funnel); @Override synchronized void disablePort(final Port port); @Override synchronized void enablePort(final Port port); @Override synchronized void enableProcessor(final ProcessorNode procNode); @Override synchronized void disableProcessor(final ProcessorNode procNode); synchronized void enableReportingTask(final ReportingTaskNode taskNode); synchronized void disableReportingTask(final ReportingTaskNode taskNode); @Override boolean isScheduled(final Object scheduled); @Override CompletableFuture<Void> enableControllerService(final ControllerServiceNode service); @Override CompletableFuture<Void> disableControllerService(final ControllerServiceNode service); @Override CompletableFuture<Void> disableControllerServices(final List<ControllerServiceNode> services);  }