[
https://issues.apache.org/jira/browse/NIP-46?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18126478#comment-18126478
]
Mike Thomsen commented on NIP-46:
---------------------------------
[~markap14] have you had a chance to look at the PR based on your suggestion
yet?
> Increase Error Observability in Stateless NiFi
> ----------------------------------------------
>
> Key: NIP-46
> URL: https://issues.apache.org/jira/browse/NIP-46
> Project: NiFi Improvement Proposal
> Issue Type: Improvement
> Reporter: Mike Thomsen
> Assignee: Mike Thomsen
> Priority: Major
>
> h2. Increase Error Observability in Stateless NiFi
> h2. Motivation
> Creating a simple observability framework for Stateless NiFi would allow
> applications that start stateless NiFi to hook into the flow components and
> more effectively track how, where and why a flow failed. One particularly
> strong use case this could open for stateless NiFi is to be used as a CI/CD
> test harness for NiFi flows wherein they would be run in an integration test
> and the results cleanly reported to the CI/CD pipeline.
> h2. Scope
> This proposal aims to actively avoid any changes to nifi-api and focus
> entirely on non-breaking changes to the stateless NiFi framework. It would
> add a callback framework focused on functional interfaces and event handlers
> that would default to no-op implementations in the absence of a developer
> defining callbacks for their particular use case.
> h2. Description
> The following will be added to provide a standard callback API that Stateless
> NiFi can use to link the calling Java code and stateless NiFi. It will be
> wired into the classes in org.apache.nifi.stateless as needed to ensure that
> component errors can be tracked effectively by the calling application.
> There are two independent callbacks, each with a no-op default and each
> purely observational: the
> dataflow fails exactly as it would have with no listener registered, and a
> listener that throws has its
> exception logged and discarded. {{DataflowTriggerContext}} exposes both via
> defaulted methods
> ({{getComponentFailureListener()}}, {{getFailurePortListener()}}), each
> returning its {{NOP}} instance.
> h3. Component failures
> Notified whenever a component throws from {{onTrigger}}.
> {code:language=|borderStyle=solid|theme=RDark|linenumbers=true|collapse=true}
> public record ComponentFailure(String componentId, String componentName,
> String componentType, Throwable cause) {
> }
> @FunctionalInterface
> public interface ComponentFailureListener {
> ComponentFailureListener NOP = failure -> {
> };
> /**
> * Invoked after a component has thrown and before the exception is
> propagated.
> */
> void onComponentFailure(ComponentFailure failure);
> }
> public class ComponentFailureNotifier {
> private final ComponentFailureListener listener;
> // By identity: two distinct exceptions with equal messages are still two
> separate failures.
> private final Set<Throwable> alreadyReported =
> Collections.synchronizedSet(Collections.newSetFromMap(new
> IdentityHashMap<>()));
> public ComponentFailureNotifier(final ComponentFailureListener listener) {
> this.listener = listener == null ? ComponentFailureListener.NOP :
> listener;
> }
> public void notifyFailure(final Connectable connectable, final Throwable
> cause) {
> // FailurePortEncounteredException, DataflowAbortedException and
> TerminatedTaskException unwind
> // through whichever component was committing a session; they are the
> dataflow's outcome, not a
> // component failure, so a cause chain containing one is skipped.
> if (isDataflowOutcome(cause) || !alreadyReported.add(cause)) {
> return;
> }
> try {
> listener.onComponentFailure(new
> ComponentFailure(connectable.getIdentifier(), connectable.getName(),
> connectable.getComponentType(), cause));
> } catch (final Throwable listenerFailure) {
> logger.error("Failed to notify {} that {} threw an exception",
> listener, connectable, listenerFailure);
> }
> }
> }
> {code}
> The notifier is invoked from the main scheduling loop, from inline
> synchronous-commit unwinding (where
> only the innermost point knows which component threw, hence the identity
> de-duplication), and from
> asynchronous commit success-callbacks that run after the dataflow has
> otherwise completed.
> h3. Failure-port routing
> Notified when a FlowFile reaches a port marked as a failure port. This is the
> observational counterpart
> to the {{FailurePortEncounteredException}} that aborts the dataflow, adding
> the identity of the routing
> component when it is known. The routing component is carried by value, not
> reference, so the event can
> cross the dataflow's ClassLoader; it is absent when no component in the flow
> routed the FlowFile, e.g.
> an input port wired straight to a failure port.
> {code:language=|borderStyle=solid|theme=RDark|linenumbers=true|collapse=false}
> public record FailurePortEvent(String portName, String sourceComponentId,
> String sourceComponentName,
> String sourceComponentType) {
> }
> @FunctionalInterface
> public interface FailurePortListener {
> FailurePortListener NOP = event -> {
> };
> /**
> * Invoked when a FlowFile has reached a failure port, before the
> dataflow's failure is propagated.
> */
> void onFailurePort(FailurePortEvent event);
> }
> {code}
> {{ExecutionProgress.reportFailurePort(Connectable sourceComponent, String
> portName)}} fires the listener,
> at most once per port name, from the synchronous and asynchronous commit
> paths in
> {{StatelessProcessSession}} and from the output check performed when the
> {{TriggerResult}} is built.
> h2. Compatibility
> This change will aim to make no changes to nifi-api and no breaking changes
> to the NiFi stateless implementations.
> h2. Verification
> Integration tests will be added to verify the new functionality and existing
> integration tests will be retained to ensure that existing implementations
> remain fully support.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
