@Test public void testIsWorkToDoFunnels() { final Funnel funnel = Mockito.mock(Funnel.class); Mockito.when(funnel.hasIncomingConnection()).thenReturn(false); Mockito.when(funnel.getRunnableComponent()).thenReturn(funnel); Mockito.when(funnel.getConnectableType()).thenReturn(ConnectableType.FUNNEL); Mockito.when(funnel.getIdentifier()).thenReturn("funnel-1"); final ConnectableTask task = createTask(funnel); assertTrue("If there is no incoming connection, it should be yielded.", task.invoke().isYield()); final Connection selfLoopingConnection = Mockito.mock(Connection.class); when(selfLoopingConnection.getSource()).thenReturn(funnel); when(selfLoopingConnection.getDestination()).thenReturn(funnel); when(funnel.hasIncomingConnection()).thenReturn(true); when(funnel.getIncomingConnections()).thenReturn(Collections.singletonList(selfLoopingConnection)); final FlowFileQueue emptyQueue = Mockito.mock(FlowFileQueue.class); when(emptyQueue.isActiveQueueEmpty()).thenReturn(true); when(selfLoopingConnection.getFlowFileQueue()).thenReturn(emptyQueue); final Set<Connection> outgoingConnections = new HashSet<>(); outgoingConnections.add(selfLoopingConnection); when(funnel.getConnections()).thenReturn(outgoingConnections); assertTrue("If there is no incoming connection from other components, it should be yielded.", task.invoke().isYield()); final ProcessorNode inputProcessor = Mockito.mock(ProcessorNode.class); final Connection incomingFromAnotherComponent = Mockito.mock(Connection.class); when(incomingFromAnotherComponent.getSource()).thenReturn(inputProcessor); when(incomingFromAnotherComponent.getDestination()).thenReturn(funnel); when(incomingFromAnotherComponent.getFlowFileQueue()).thenReturn(emptyQueue); when(funnel.hasIncomingConnection()).thenReturn(true); when(funnel.getIncomingConnections()).thenReturn(Arrays.asList(selfLoopingConnection, incomingFromAnotherComponent)); assertTrue("Even if there is an incoming connection from another component," + " it should be yielded because there's no outgoing connections.", task.invoke().isYield()); final ProcessorNode outputProcessor = Mockito.mock(ProcessorNode.class); final Connection outgoingToAnotherComponent = Mockito.mock(Connection.class); when(outgoingToAnotherComponent.getSource()).thenReturn(funnel); when(outgoingToAnotherComponent.getDestination()).thenReturn(outputProcessor); outgoingConnections.add(outgoingToAnotherComponent); assertTrue("Even if there is an incoming connection from another component and an outgoing connection as well," + " it should be yielded because there's no incoming FlowFiles to process.", task.invoke().isYield()); final FlowFileQueue nonEmptyQueue = Mockito.mock(FlowFileQueue.class); when(nonEmptyQueue.isActiveQueueEmpty()).thenReturn(false); when(incomingFromAnotherComponent.getFlowFileQueue()).thenReturn(nonEmptyQueue); assertFalse("When a Funnel has both incoming and outgoing connections and FlowFiles to process, then it should be executed.", task.invoke().isYield()); }
public InvocationResult invoke() { if (scheduleState.isTerminated()) { return InvocationResult.DO_NOT_YIELD; } if (isYielded()) { return InvocationResult.DO_NOT_YIELD; } if (!isRunOnCluster(flowController)) { return InvocationResult.DO_NOT_YIELD; } if (!isWorkToDo()) { return InvocationResult.yield("No work to do"); } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; if (!repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships)) { return InvocationResult.yield("Backpressure Applied"); } } final long batchNanos = connectable.getRunDuration(TimeUnit.NANOSECONDS); final ProcessSessionFactory sessionFactory; final StandardProcessSession rawSession; final boolean batch; if (connectable.isSessionBatchingSupported() && batchNanos > 0L) { rawSession = new StandardProcessSession(repositoryContext, scheduleState::isTerminated); sessionFactory = new BatchingSessionFactory(rawSession); batch = true; } else { rawSession = null; sessionFactory = new StandardProcessSessionFactory(repositoryContext, scheduleState::isTerminated); batch = false; } final ActiveProcessSessionFactory activeSessionFactory = new WeakHashMapProcessSessionFactory(sessionFactory); scheduleState.incrementActiveThreadCount(activeSessionFactory); final long startNanos = System.nanoTime(); final long finishIfBackpressureEngaged = startNanos + (batchNanos / 25L); final long finishNanos = startNanos + batchNanos; int invocationCount = 0; final String originalThreadName = Thread.currentThread().getName(); try { try (final AutoCloseable ncl = NarCloseable.withComponentNarLoader(connectable.getRunnableComponent().getClass(), connectable.getIdentifier())) { boolean shouldRun = connectable.getScheduledState() == ScheduledState.RUNNING; while (shouldRun) { connectable.onTrigger(processContext, activeSessionFactory); invocationCount++; currentExceptionCount.getAndSet(0); if (!batch) { return InvocationResult.DO_NOT_YIELD; } final long nanoTime = System.nanoTime(); if (nanoTime > finishNanos) { return InvocationResult.DO_NOT_YIELD; } if (nanoTime > finishIfBackpressureEngaged && isBackPressureEngaged()) { return InvocationResult.DO_NOT_YIELD; } if (connectable.getScheduledState() != ScheduledState.RUNNING) { break; } if (!isWorkToDo()) { break; } if (isYielded()) { break; } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; shouldRun = repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships); } } } catch (final TerminatedTaskException tte) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.info("Failed to process session due to task being terminated", new Object[] {tte}); } catch (final ProcessException pe) { currentExceptionCount.incrementAndGet() ; final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to process session due to {}", new Object[] {pe}); if (exceptionToleranceCount != 0 && currentExceptionCount.get() >= exceptionToleranceCount) { currentExceptionCount.getAndSet(0); throw pe; } } catch (final Throwable t) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("{} failed to process session due to {}; Processor Administratively Yielded for {}", new Object[] {connectable.getRunnableComponent(), t, schedulingAgent.getAdministrativeYieldDuration()}, t); logger.warn("Administratively Yielding {} due to uncaught Exception: {}", connectable.getRunnableComponent(), t.toString(), t); connectable.yield(schedulingAgent.getAdministrativeYieldDuration(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS); } } finally { try { if (batch) { try { rawSession.commit(); } catch (final Exception e) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to commit session {} due to {}; rolling back", new Object[] { rawSession, e.toString() }, e); try { rawSession.rollback(true); } catch (final Exception e1) { procLog.error("Failed to roll back session {} due to {}", new Object[] { rawSession, e.toString() }, e); } } } final long processingNanos = System.nanoTime() - startNanos; try { final StandardFlowFileEvent procEvent = new StandardFlowFileEvent(connectable.getIdentifier()); procEvent.setProcessingNanos(processingNanos); procEvent.setInvocations(invocationCount); repositoryContext.getFlowFileEventRepository().updateRepository(procEvent); } catch (final IOException e) { logger.error("Unable to update FlowFileEvent Repository for {}; statistics may be inaccurate. Reason for failure: {}", connectable.getRunnableComponent(), e.toString()); logger.error("", e); } } finally { scheduleState.decrementActiveThreadCount(activeSessionFactory); Thread.currentThread().setName(originalThreadName); } } return InvocationResult.DO_NOT_YIELD; }
ConnectableTask { public InvocationResult invoke() { if (scheduleState.isTerminated()) { return InvocationResult.DO_NOT_YIELD; } if (isYielded()) { return InvocationResult.DO_NOT_YIELD; } if (!isRunOnCluster(flowController)) { return InvocationResult.DO_NOT_YIELD; } if (!isWorkToDo()) { return InvocationResult.yield("No work to do"); } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; if (!repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships)) { return InvocationResult.yield("Backpressure Applied"); } } final long batchNanos = connectable.getRunDuration(TimeUnit.NANOSECONDS); final ProcessSessionFactory sessionFactory; final StandardProcessSession rawSession; final boolean batch; if (connectable.isSessionBatchingSupported() && batchNanos > 0L) { rawSession = new StandardProcessSession(repositoryContext, scheduleState::isTerminated); sessionFactory = new BatchingSessionFactory(rawSession); batch = true; } else { rawSession = null; sessionFactory = new StandardProcessSessionFactory(repositoryContext, scheduleState::isTerminated); batch = false; } final ActiveProcessSessionFactory activeSessionFactory = new WeakHashMapProcessSessionFactory(sessionFactory); scheduleState.incrementActiveThreadCount(activeSessionFactory); final long startNanos = System.nanoTime(); final long finishIfBackpressureEngaged = startNanos + (batchNanos / 25L); final long finishNanos = startNanos + batchNanos; int invocationCount = 0; final String originalThreadName = Thread.currentThread().getName(); try { try (final AutoCloseable ncl = NarCloseable.withComponentNarLoader(connectable.getRunnableComponent().getClass(), connectable.getIdentifier())) { boolean shouldRun = connectable.getScheduledState() == ScheduledState.RUNNING; while (shouldRun) { connectable.onTrigger(processContext, activeSessionFactory); invocationCount++; currentExceptionCount.getAndSet(0); if (!batch) { return InvocationResult.DO_NOT_YIELD; } final long nanoTime = System.nanoTime(); if (nanoTime > finishNanos) { return InvocationResult.DO_NOT_YIELD; } if (nanoTime > finishIfBackpressureEngaged && isBackPressureEngaged()) { return InvocationResult.DO_NOT_YIELD; } if (connectable.getScheduledState() != ScheduledState.RUNNING) { break; } if (!isWorkToDo()) { break; } if (isYielded()) { break; } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; shouldRun = repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships); } } } catch (final TerminatedTaskException tte) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.info("Failed to process session due to task being terminated", new Object[] {tte}); } catch (final ProcessException pe) { currentExceptionCount.incrementAndGet() ; final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to process session due to {}", new Object[] {pe}); if (exceptionToleranceCount != 0 && currentExceptionCount.get() >= exceptionToleranceCount) { currentExceptionCount.getAndSet(0); throw pe; } } catch (final Throwable t) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("{} failed to process session due to {}; Processor Administratively Yielded for {}", new Object[] {connectable.getRunnableComponent(), t, schedulingAgent.getAdministrativeYieldDuration()}, t); logger.warn("Administratively Yielding {} due to uncaught Exception: {}", connectable.getRunnableComponent(), t.toString(), t); connectable.yield(schedulingAgent.getAdministrativeYieldDuration(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS); } } finally { try { if (batch) { try { rawSession.commit(); } catch (final Exception e) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to commit session {} due to {}; rolling back", new Object[] { rawSession, e.toString() }, e); try { rawSession.rollback(true); } catch (final Exception e1) { procLog.error("Failed to roll back session {} due to {}", new Object[] { rawSession, e.toString() }, e); } } } final long processingNanos = System.nanoTime() - startNanos; try { final StandardFlowFileEvent procEvent = new StandardFlowFileEvent(connectable.getIdentifier()); procEvent.setProcessingNanos(processingNanos); procEvent.setInvocations(invocationCount); repositoryContext.getFlowFileEventRepository().updateRepository(procEvent); } catch (final IOException e) { logger.error("Unable to update FlowFileEvent Repository for {}; statistics may be inaccurate. Reason for failure: {}", connectable.getRunnableComponent(), e.toString()); logger.error("", e); } } finally { scheduleState.decrementActiveThreadCount(activeSessionFactory); Thread.currentThread().setName(originalThreadName); } } return InvocationResult.DO_NOT_YIELD; } }
ConnectableTask { public InvocationResult invoke() { if (scheduleState.isTerminated()) { return InvocationResult.DO_NOT_YIELD; } if (isYielded()) { return InvocationResult.DO_NOT_YIELD; } if (!isRunOnCluster(flowController)) { return InvocationResult.DO_NOT_YIELD; } if (!isWorkToDo()) { return InvocationResult.yield("No work to do"); } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; if (!repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships)) { return InvocationResult.yield("Backpressure Applied"); } } final long batchNanos = connectable.getRunDuration(TimeUnit.NANOSECONDS); final ProcessSessionFactory sessionFactory; final StandardProcessSession rawSession; final boolean batch; if (connectable.isSessionBatchingSupported() && batchNanos > 0L) { rawSession = new StandardProcessSession(repositoryContext, scheduleState::isTerminated); sessionFactory = new BatchingSessionFactory(rawSession); batch = true; } else { rawSession = null; sessionFactory = new StandardProcessSessionFactory(repositoryContext, scheduleState::isTerminated); batch = false; } final ActiveProcessSessionFactory activeSessionFactory = new WeakHashMapProcessSessionFactory(sessionFactory); scheduleState.incrementActiveThreadCount(activeSessionFactory); final long startNanos = System.nanoTime(); final long finishIfBackpressureEngaged = startNanos + (batchNanos / 25L); final long finishNanos = startNanos + batchNanos; int invocationCount = 0; final String originalThreadName = Thread.currentThread().getName(); try { try (final AutoCloseable ncl = NarCloseable.withComponentNarLoader(connectable.getRunnableComponent().getClass(), connectable.getIdentifier())) { boolean shouldRun = connectable.getScheduledState() == ScheduledState.RUNNING; while (shouldRun) { connectable.onTrigger(processContext, activeSessionFactory); invocationCount++; currentExceptionCount.getAndSet(0); if (!batch) { return InvocationResult.DO_NOT_YIELD; } final long nanoTime = System.nanoTime(); if (nanoTime > finishNanos) { return InvocationResult.DO_NOT_YIELD; } if (nanoTime > finishIfBackpressureEngaged && isBackPressureEngaged()) { return InvocationResult.DO_NOT_YIELD; } if (connectable.getScheduledState() != ScheduledState.RUNNING) { break; } if (!isWorkToDo()) { break; } if (isYielded()) { break; } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; shouldRun = repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships); } } } catch (final TerminatedTaskException tte) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.info("Failed to process session due to task being terminated", new Object[] {tte}); } catch (final ProcessException pe) { currentExceptionCount.incrementAndGet() ; final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to process session due to {}", new Object[] {pe}); if (exceptionToleranceCount != 0 && currentExceptionCount.get() >= exceptionToleranceCount) { currentExceptionCount.getAndSet(0); throw pe; } } catch (final Throwable t) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("{} failed to process session due to {}; Processor Administratively Yielded for {}", new Object[] {connectable.getRunnableComponent(), t, schedulingAgent.getAdministrativeYieldDuration()}, t); logger.warn("Administratively Yielding {} due to uncaught Exception: {}", connectable.getRunnableComponent(), t.toString(), t); connectable.yield(schedulingAgent.getAdministrativeYieldDuration(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS); } } finally { try { if (batch) { try { rawSession.commit(); } catch (final Exception e) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to commit session {} due to {}; rolling back", new Object[] { rawSession, e.toString() }, e); try { rawSession.rollback(true); } catch (final Exception e1) { procLog.error("Failed to roll back session {} due to {}", new Object[] { rawSession, e.toString() }, e); } } } final long processingNanos = System.nanoTime() - startNanos; try { final StandardFlowFileEvent procEvent = new StandardFlowFileEvent(connectable.getIdentifier()); procEvent.setProcessingNanos(processingNanos); procEvent.setInvocations(invocationCount); repositoryContext.getFlowFileEventRepository().updateRepository(procEvent); } catch (final IOException e) { logger.error("Unable to update FlowFileEvent Repository for {}; statistics may be inaccurate. Reason for failure: {}", connectable.getRunnableComponent(), e.toString()); logger.error("", e); } } finally { scheduleState.decrementActiveThreadCount(activeSessionFactory); Thread.currentThread().setName(originalThreadName); } } return InvocationResult.DO_NOT_YIELD; } ConnectableTask(final SchedulingAgent schedulingAgent, final Connectable connectable,
            final FlowController flowController, final RepositoryContextFactory contextFactory, final LifecycleState scheduleState,
            final StringEncryptor encryptor); }
ConnectableTask { public InvocationResult invoke() { if (scheduleState.isTerminated()) { return InvocationResult.DO_NOT_YIELD; } if (isYielded()) { return InvocationResult.DO_NOT_YIELD; } if (!isRunOnCluster(flowController)) { return InvocationResult.DO_NOT_YIELD; } if (!isWorkToDo()) { return InvocationResult.yield("No work to do"); } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; if (!repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships)) { return InvocationResult.yield("Backpressure Applied"); } } final long batchNanos = connectable.getRunDuration(TimeUnit.NANOSECONDS); final ProcessSessionFactory sessionFactory; final StandardProcessSession rawSession; final boolean batch; if (connectable.isSessionBatchingSupported() && batchNanos > 0L) { rawSession = new StandardProcessSession(repositoryContext, scheduleState::isTerminated); sessionFactory = new BatchingSessionFactory(rawSession); batch = true; } else { rawSession = null; sessionFactory = new StandardProcessSessionFactory(repositoryContext, scheduleState::isTerminated); batch = false; } final ActiveProcessSessionFactory activeSessionFactory = new WeakHashMapProcessSessionFactory(sessionFactory); scheduleState.incrementActiveThreadCount(activeSessionFactory); final long startNanos = System.nanoTime(); final long finishIfBackpressureEngaged = startNanos + (batchNanos / 25L); final long finishNanos = startNanos + batchNanos; int invocationCount = 0; final String originalThreadName = Thread.currentThread().getName(); try { try (final AutoCloseable ncl = NarCloseable.withComponentNarLoader(connectable.getRunnableComponent().getClass(), connectable.getIdentifier())) { boolean shouldRun = connectable.getScheduledState() == ScheduledState.RUNNING; while (shouldRun) { connectable.onTrigger(processContext, activeSessionFactory); invocationCount++; currentExceptionCount.getAndSet(0); if (!batch) { return InvocationResult.DO_NOT_YIELD; } final long nanoTime = System.nanoTime(); if (nanoTime > finishNanos) { return InvocationResult.DO_NOT_YIELD; } if (nanoTime > finishIfBackpressureEngaged && isBackPressureEngaged()) { return InvocationResult.DO_NOT_YIELD; } if (connectable.getScheduledState() != ScheduledState.RUNNING) { break; } if (!isWorkToDo()) { break; } if (isYielded()) { break; } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; shouldRun = repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships); } } } catch (final TerminatedTaskException tte) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.info("Failed to process session due to task being terminated", new Object[] {tte}); } catch (final ProcessException pe) { currentExceptionCount.incrementAndGet() ; final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to process session due to {}", new Object[] {pe}); if (exceptionToleranceCount != 0 && currentExceptionCount.get() >= exceptionToleranceCount) { currentExceptionCount.getAndSet(0); throw pe; } } catch (final Throwable t) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("{} failed to process session due to {}; Processor Administratively Yielded for {}", new Object[] {connectable.getRunnableComponent(), t, schedulingAgent.getAdministrativeYieldDuration()}, t); logger.warn("Administratively Yielding {} due to uncaught Exception: {}", connectable.getRunnableComponent(), t.toString(), t); connectable.yield(schedulingAgent.getAdministrativeYieldDuration(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS); } } finally { try { if (batch) { try { rawSession.commit(); } catch (final Exception e) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to commit session {} due to {}; rolling back", new Object[] { rawSession, e.toString() }, e); try { rawSession.rollback(true); } catch (final Exception e1) { procLog.error("Failed to roll back session {} due to {}", new Object[] { rawSession, e.toString() }, e); } } } final long processingNanos = System.nanoTime() - startNanos; try { final StandardFlowFileEvent procEvent = new StandardFlowFileEvent(connectable.getIdentifier()); procEvent.setProcessingNanos(processingNanos); procEvent.setInvocations(invocationCount); repositoryContext.getFlowFileEventRepository().updateRepository(procEvent); } catch (final IOException e) { logger.error("Unable to update FlowFileEvent Repository for {}; statistics may be inaccurate. Reason for failure: {}", connectable.getRunnableComponent(), e.toString()); logger.error("", e); } } finally { scheduleState.decrementActiveThreadCount(activeSessionFactory); Thread.currentThread().setName(originalThreadName); } } return InvocationResult.DO_NOT_YIELD; } ConnectableTask(final SchedulingAgent schedulingAgent, final Connectable connectable,
            final FlowController flowController, final RepositoryContextFactory contextFactory, final LifecycleState scheduleState,
            final StringEncryptor encryptor); void setExceptionToleranceCount(int exceptionToleranceCount); int getExceptionToleranceCount(); Connectable getConnectable(); InvocationResult invoke(); }
ConnectableTask { public InvocationResult invoke() { if (scheduleState.isTerminated()) { return InvocationResult.DO_NOT_YIELD; } if (isYielded()) { return InvocationResult.DO_NOT_YIELD; } if (!isRunOnCluster(flowController)) { return InvocationResult.DO_NOT_YIELD; } if (!isWorkToDo()) { return InvocationResult.yield("No work to do"); } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; if (!repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships)) { return InvocationResult.yield("Backpressure Applied"); } } final long batchNanos = connectable.getRunDuration(TimeUnit.NANOSECONDS); final ProcessSessionFactory sessionFactory; final StandardProcessSession rawSession; final boolean batch; if (connectable.isSessionBatchingSupported() && batchNanos > 0L) { rawSession = new StandardProcessSession(repositoryContext, scheduleState::isTerminated); sessionFactory = new BatchingSessionFactory(rawSession); batch = true; } else { rawSession = null; sessionFactory = new StandardProcessSessionFactory(repositoryContext, scheduleState::isTerminated); batch = false; } final ActiveProcessSessionFactory activeSessionFactory = new WeakHashMapProcessSessionFactory(sessionFactory); scheduleState.incrementActiveThreadCount(activeSessionFactory); final long startNanos = System.nanoTime(); final long finishIfBackpressureEngaged = startNanos + (batchNanos / 25L); final long finishNanos = startNanos + batchNanos; int invocationCount = 0; final String originalThreadName = Thread.currentThread().getName(); try { try (final AutoCloseable ncl = NarCloseable.withComponentNarLoader(connectable.getRunnableComponent().getClass(), connectable.getIdentifier())) { boolean shouldRun = connectable.getScheduledState() == ScheduledState.RUNNING; while (shouldRun) { connectable.onTrigger(processContext, activeSessionFactory); invocationCount++; currentExceptionCount.getAndSet(0); if (!batch) { return InvocationResult.DO_NOT_YIELD; } final long nanoTime = System.nanoTime(); if (nanoTime > finishNanos) { return InvocationResult.DO_NOT_YIELD; } if (nanoTime > finishIfBackpressureEngaged && isBackPressureEngaged()) { return InvocationResult.DO_NOT_YIELD; } if (connectable.getScheduledState() != ScheduledState.RUNNING) { break; } if (!isWorkToDo()) { break; } if (isYielded()) { break; } if (numRelationships > 0) { final int requiredNumberOfAvailableRelationships = connectable.isTriggerWhenAnyDestinationAvailable() ? 1 : numRelationships; shouldRun = repositoryContext.isRelationshipAvailabilitySatisfied(requiredNumberOfAvailableRelationships); } } } catch (final TerminatedTaskException tte) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.info("Failed to process session due to task being terminated", new Object[] {tte}); } catch (final ProcessException pe) { currentExceptionCount.incrementAndGet() ; final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to process session due to {}", new Object[] {pe}); if (exceptionToleranceCount != 0 && currentExceptionCount.get() >= exceptionToleranceCount) { currentExceptionCount.getAndSet(0); throw pe; } } catch (final Throwable t) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("{} failed to process session due to {}; Processor Administratively Yielded for {}", new Object[] {connectable.getRunnableComponent(), t, schedulingAgent.getAdministrativeYieldDuration()}, t); logger.warn("Administratively Yielding {} due to uncaught Exception: {}", connectable.getRunnableComponent(), t.toString(), t); connectable.yield(schedulingAgent.getAdministrativeYieldDuration(TimeUnit.NANOSECONDS), TimeUnit.NANOSECONDS); } } finally { try { if (batch) { try { rawSession.commit(); } catch (final Exception e) { final ComponentLog procLog = new SimpleProcessLogger(connectable.getIdentifier(), connectable.getRunnableComponent()); procLog.error("Failed to commit session {} due to {}; rolling back", new Object[] { rawSession, e.toString() }, e); try { rawSession.rollback(true); } catch (final Exception e1) { procLog.error("Failed to roll back session {} due to {}", new Object[] { rawSession, e.toString() }, e); } } } final long processingNanos = System.nanoTime() - startNanos; try { final StandardFlowFileEvent procEvent = new StandardFlowFileEvent(connectable.getIdentifier()); procEvent.setProcessingNanos(processingNanos); procEvent.setInvocations(invocationCount); repositoryContext.getFlowFileEventRepository().updateRepository(procEvent); } catch (final IOException e) { logger.error("Unable to update FlowFileEvent Repository for {}; statistics may be inaccurate. Reason for failure: {}", connectable.getRunnableComponent(), e.toString()); logger.error("", e); } } finally { scheduleState.decrementActiveThreadCount(activeSessionFactory); Thread.currentThread().setName(originalThreadName); } } return InvocationResult.DO_NOT_YIELD; } ConnectableTask(final SchedulingAgent schedulingAgent, final Connectable connectable,
            final FlowController flowController, final RepositoryContextFactory contextFactory, final LifecycleState scheduleState,
            final StringEncryptor encryptor); void setExceptionToleranceCount(int exceptionToleranceCount); int getExceptionToleranceCount(); Connectable getConnectable(); InvocationResult invoke();  }