Mike Thomsen created NIP-46:
-------------------------------
Summary: 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
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)
