@Test(timeout = 10000) public void testStart() throws InterruptedException { System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, TestStandardProcessorNode.class.getResource("/conf/nifi.properties").getFile()); final ProcessorThatThrowsExceptionOnScheduled processor = new ProcessorThatThrowsExceptionOnScheduled(); final String uuid = UUID.randomUUID().toString(); ProcessorInitializationContext initContext = new StandardProcessorInitializationContext(uuid, null, null, null, null); processor.initialize(initContext); final StandardProcessorNode procNode = new StandardProcessorNode(processor, uuid, createValidationContextFactory(), null, null, NiFiProperties.createBasicNiFiProperties(null, null), VariableRegistry.EMPTY_REGISTRY, Mockito.mock(ComponentLog.class)); final ScheduledExecutorService taskScheduler = new FlowEngine(2, "TestClasspathResources", true); final StandardProcessContext processContext = new StandardProcessContext(procNode, null, null, null, null); final SchedulingAgentCallback schedulingAgentCallback = new SchedulingAgentCallback() { @Override public void postMonitor() { } @Override public Future<?> invokeMonitoringTask(final Callable<?> task) { return taskScheduler.submit(task); } @Override public void trigger() { Assert.fail("Should not have completed"); } }; procNode.start(taskScheduler, 20000L, processContext, schedulingAgentCallback); Thread.sleep(1000L); assertEquals(1, processor.onScheduledCount); assertEquals(1, processor.onUnscheduledCount); assertEquals(1, processor.onStoppedCount); }
@Override public <T extends ProcessContext & ControllerServiceLookup> void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final T processContext, final SchedulingAgentCallback schedulingAgentCallback) { if (!this.isValid()) { throw new IllegalStateException( "Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); } final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); if (this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING)) { final Runnable startProcRunnable = new Runnable() { @Override public void run() { try { invokeTaskAsCancelableFuture(schedulingAgentCallback, new Callable<Void>() { @Override public Void call() throws Exception { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.invokeMethodsWithAnnotation(OnScheduled.class, processor, processContext); return null; } } }); if (scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING)) { schedulingAgentCallback.trigger(); } else { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); } scheduledState.set(ScheduledState.STOPPED); } } catch (final Exception e) { final Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e; procLog.error("{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run for {} seconds", new Object[] {StandardProcessorNode.this.getProcessor(), cause, administrativeYieldMillis / 1000L}, cause); LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext); if (scheduledState.get() != ScheduledState.STOPPING) { taskScheduler.schedule(this, administrativeYieldMillis, TimeUnit.MILLISECONDS); } else { scheduledState.set(ScheduledState.STOPPED); } } } }; taskScheduler.execute(startProcRunnable); } else { final String procName = this.processor.getClass().getSimpleName(); LOG.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); procLog.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); } }
StandardProcessorNode extends ProcessorNode implements Connectable { @Override public <T extends ProcessContext & ControllerServiceLookup> void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final T processContext, final SchedulingAgentCallback schedulingAgentCallback) { if (!this.isValid()) { throw new IllegalStateException( "Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); } final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); if (this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING)) { final Runnable startProcRunnable = new Runnable() { @Override public void run() { try { invokeTaskAsCancelableFuture(schedulingAgentCallback, new Callable<Void>() { @Override public Void call() throws Exception { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.invokeMethodsWithAnnotation(OnScheduled.class, processor, processContext); return null; } } }); if (scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING)) { schedulingAgentCallback.trigger(); } else { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); } scheduledState.set(ScheduledState.STOPPED); } } catch (final Exception e) { final Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e; procLog.error("{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run for {} seconds", new Object[] {StandardProcessorNode.this.getProcessor(), cause, administrativeYieldMillis / 1000L}, cause); LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext); if (scheduledState.get() != ScheduledState.STOPPING) { taskScheduler.schedule(this, administrativeYieldMillis, TimeUnit.MILLISECONDS); } else { scheduledState.set(ScheduledState.STOPPED); } } } }; taskScheduler.execute(startProcRunnable); } else { final String procName = this.processor.getClass().getSimpleName(); LOG.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); procLog.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); } } }
StandardProcessorNode extends ProcessorNode implements Connectable { @Override public <T extends ProcessContext & ControllerServiceLookup> void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final T processContext, final SchedulingAgentCallback schedulingAgentCallback) { if (!this.isValid()) { throw new IllegalStateException( "Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); } final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); if (this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING)) { final Runnable startProcRunnable = new Runnable() { @Override public void run() { try { invokeTaskAsCancelableFuture(schedulingAgentCallback, new Callable<Void>() { @Override public Void call() throws Exception { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.invokeMethodsWithAnnotation(OnScheduled.class, processor, processContext); return null; } } }); if (scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING)) { schedulingAgentCallback.trigger(); } else { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); } scheduledState.set(ScheduledState.STOPPED); } } catch (final Exception e) { final Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e; procLog.error("{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run for {} seconds", new Object[] {StandardProcessorNode.this.getProcessor(), cause, administrativeYieldMillis / 1000L}, cause); LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext); if (scheduledState.get() != ScheduledState.STOPPING) { taskScheduler.schedule(this, administrativeYieldMillis, TimeUnit.MILLISECONDS); } else { scheduledState.set(ScheduledState.STOPPED); } } } }; taskScheduler.execute(startProcRunnable); } else { final String procName = this.processor.getClass().getSimpleName(); LOG.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); procLog.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); } } StandardProcessorNode(final Processor processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider, final NiFiProperties nifiProperties,
                                 final VariableRegistry variableRegistry, final ComponentLog logger);  StandardProcessorNode(final Processor processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider,
                                 final String componentType, final String componentCanonicalClass, final NiFiProperties nifiProperties,
                                 final VariableRegistry variableRegistry, final ComponentLog logger); }
StandardProcessorNode extends ProcessorNode implements Connectable { @Override public <T extends ProcessContext & ControllerServiceLookup> void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final T processContext, final SchedulingAgentCallback schedulingAgentCallback) { if (!this.isValid()) { throw new IllegalStateException( "Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); } final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); if (this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING)) { final Runnable startProcRunnable = new Runnable() { @Override public void run() { try { invokeTaskAsCancelableFuture(schedulingAgentCallback, new Callable<Void>() { @Override public Void call() throws Exception { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.invokeMethodsWithAnnotation(OnScheduled.class, processor, processContext); return null; } } }); if (scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING)) { schedulingAgentCallback.trigger(); } else { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); } scheduledState.set(ScheduledState.STOPPED); } } catch (final Exception e) { final Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e; procLog.error("{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run for {} seconds", new Object[] {StandardProcessorNode.this.getProcessor(), cause, administrativeYieldMillis / 1000L}, cause); LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext); if (scheduledState.get() != ScheduledState.STOPPING) { taskScheduler.schedule(this, administrativeYieldMillis, TimeUnit.MILLISECONDS); } else { scheduledState.set(ScheduledState.STOPPED); } } } }; taskScheduler.execute(startProcRunnable); } else { final String procName = this.processor.getClass().getSimpleName(); LOG.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); procLog.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); } } StandardProcessorNode(final Processor processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider, final NiFiProperties nifiProperties,
                                 final VariableRegistry variableRegistry, final ComponentLog logger);  StandardProcessorNode(final Processor processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider,
                                 final String componentType, final String componentCanonicalClass, final NiFiProperties nifiProperties,
                                 final VariableRegistry variableRegistry, final ComponentLog logger); @Override String getComments(); @Override Authorizable getParentAuthorizable(); @Override Resource getResource(); @Override boolean isRestricted(); @Override void setComments(final String comments); @Override Position getPosition(); @Override void setPosition(final Position position); @Override Map<String, String> getStyle(); @Override void setStyle(final Map<String, String> style); @Override String getIdentifier(); @Override boolean isLossTolerant(); @Override boolean isIsolated(); @Override boolean isTriggerWhenEmpty(); @Override boolean isSideEffectFree(); @Override boolean isHighThroughputSupported(); @Override boolean isTriggerWhenAnyDestinationAvailable(); @Override 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 void setName(final String name); @Override long getSchedulingPeriod(final TimeUnit timeUnit); @Override boolean isEventDrivenSupported(); @Override void setSchedulingStrategy(final SchedulingStrategy schedulingStrategy); @Override SchedulingStrategy getSchedulingStrategy(); @Override String getSchedulingPeriod(); @Override void setScheduldingPeriod(final String schedulingPeriod); @Override void setExecutionNode(final ExecutionNode executionNode); @Override ExecutionNode getExecutionNode(); @Override long getRunDuration(final TimeUnit timeUnit); @Override void setRunDuration(final long duration, final TimeUnit timeUnit); @Override long getYieldPeriod(final TimeUnit timeUnit); @Override String getYieldPeriod(); @Override 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 void setPenalizationPeriod(final String penalizationPeriod); @Override void setMaxConcurrentTasks(final int taskCount); @Override boolean isTriggeredSerially(); @Override int getMaxConcurrentTasks(); @Override LogLevel getBulletinLevel(); @Override 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(); Set<Connectable> getDestinations(); Set<Connectable> getDestinations(final Relationship relationship); Set<Relationship> getUndefinedRelationships(); @Override boolean isRunning(); @Override int getActiveThreadCount(); @Override boolean isValid(); @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 void setProcessGroup(final ProcessGroup group); @Override void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory); @Override ConnectableType getConnectableType(); @Override void setAnnotationData(final String data); @Override Collection<ValidationResult> validate(final ValidationContext validationContext); @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 start(final ScheduledExecutorService taskScheduler,
            final long administrativeYieldMillis, final T processContext, final SchedulingAgentCallback schedulingAgentCallback); @Override void stop(final ScheduledExecutorService scheduler,
        final T processContext, final SchedulingAgent schedulingAgent, final ScheduleState scheduleState); @Override String getProcessGroupIdentifier(); }
StandardProcessorNode extends ProcessorNode implements Connectable { @Override public <T extends ProcessContext & ControllerServiceLookup> void start(final ScheduledExecutorService taskScheduler, final long administrativeYieldMillis, final T processContext, final SchedulingAgentCallback schedulingAgentCallback) { if (!this.isValid()) { throw new IllegalStateException( "Processor " + this.getName() + " is not in a valid state due to " + this.getValidationErrors()); } final ComponentLog procLog = new SimpleProcessLogger(StandardProcessorNode.this.getIdentifier(), processor); if (this.scheduledState.compareAndSet(ScheduledState.STOPPED, ScheduledState.STARTING)) { final Runnable startProcRunnable = new Runnable() { @Override public void run() { try { invokeTaskAsCancelableFuture(schedulingAgentCallback, new Callable<Void>() { @Override public Void call() throws Exception { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.invokeMethodsWithAnnotation(OnScheduled.class, processor, processContext); return null; } } }); if (scheduledState.compareAndSet(ScheduledState.STARTING, ScheduledState.RUNNING)) { schedulingAgentCallback.trigger(); } else { try (final NarCloseable nc = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) { ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); } scheduledState.set(ScheduledState.STOPPED); } } catch (final Exception e) { final Throwable cause = e instanceof InvocationTargetException ? e.getCause() : e; procLog.error("{} failed to invoke @OnScheduled method due to {}; processor will not be scheduled to run for {} seconds", new Object[] {StandardProcessorNode.this.getProcessor(), cause, administrativeYieldMillis / 1000L}, cause); LOG.error("Failed to invoke @OnScheduled method due to {}", cause.toString(), cause); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnUnscheduled.class, processor, processContext); ReflectionUtils.quietlyInvokeMethodsWithAnnotation(OnStopped.class, processor, processContext); if (scheduledState.get() != ScheduledState.STOPPING) { taskScheduler.schedule(this, administrativeYieldMillis, TimeUnit.MILLISECONDS); } else { scheduledState.set(ScheduledState.STOPPED); } } } }; taskScheduler.execute(startProcRunnable); } else { final String procName = this.processor.getClass().getSimpleName(); LOG.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); procLog.warn("Can not start '" + procName + "' since it's already in the process of being started or it is DISABLED - " + scheduledState.get()); } } StandardProcessorNode(final Processor processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider, final NiFiProperties nifiProperties,
                                 final VariableRegistry variableRegistry, final ComponentLog logger);  StandardProcessorNode(final Processor processor, final String uuid,
                                 final ValidationContextFactory validationContextFactory, final ProcessScheduler scheduler,
                                 final ControllerServiceProvider controllerServiceProvider,
                                 final String componentType, final String componentCanonicalClass, final NiFiProperties nifiProperties,
                                 final VariableRegistry variableRegistry, final ComponentLog logger); @Override String getComments(); @Override Authorizable getParentAuthorizable(); @Override Resource getResource(); @Override boolean isRestricted(); @Override void setComments(final String comments); @Override Position getPosition(); @Override void setPosition(final Position position); @Override Map<String, String> getStyle(); @Override void setStyle(final Map<String, String> style); @Override String getIdentifier(); @Override boolean isLossTolerant(); @Override boolean isIsolated(); @Override boolean isTriggerWhenEmpty(); @Override boolean isSideEffectFree(); @Override boolean isHighThroughputSupported(); @Override boolean isTriggerWhenAnyDestinationAvailable(); @Override 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 void setName(final String name); @Override long getSchedulingPeriod(final TimeUnit timeUnit); @Override boolean isEventDrivenSupported(); @Override void setSchedulingStrategy(final SchedulingStrategy schedulingStrategy); @Override SchedulingStrategy getSchedulingStrategy(); @Override String getSchedulingPeriod(); @Override void setScheduldingPeriod(final String schedulingPeriod); @Override void setExecutionNode(final ExecutionNode executionNode); @Override ExecutionNode getExecutionNode(); @Override long getRunDuration(final TimeUnit timeUnit); @Override void setRunDuration(final long duration, final TimeUnit timeUnit); @Override long getYieldPeriod(final TimeUnit timeUnit); @Override String getYieldPeriod(); @Override 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 void setPenalizationPeriod(final String penalizationPeriod); @Override void setMaxConcurrentTasks(final int taskCount); @Override boolean isTriggeredSerially(); @Override int getMaxConcurrentTasks(); @Override LogLevel getBulletinLevel(); @Override 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(); Set<Connectable> getDestinations(); Set<Connectable> getDestinations(final Relationship relationship); Set<Relationship> getUndefinedRelationships(); @Override boolean isRunning(); @Override int getActiveThreadCount(); @Override boolean isValid(); @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 void setProcessGroup(final ProcessGroup group); @Override void onTrigger(final ProcessContext context, final ProcessSessionFactory sessionFactory); @Override ConnectableType getConnectableType(); @Override void setAnnotationData(final String data); @Override Collection<ValidationResult> validate(final ValidationContext validationContext); @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 start(final ScheduledExecutorService taskScheduler,
            final long administrativeYieldMillis, final T processContext, final SchedulingAgentCallback schedulingAgentCallback); @Override void stop(final ScheduledExecutorService scheduler,
        final T processContext, final SchedulingAgent schedulingAgent, final ScheduleState scheduleState); @Override String getProcessGroupIdentifier(); static final String BULLETIN_OBSERVER_ID; static final TimeUnit DEFAULT_TIME_UNIT; static final String DEFAULT_YIELD_PERIOD; static final String DEFAULT_PENALIZATION_PERIOD; }