[
https://issues.apache.org/jira/browse/NIP-31?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Mark Payne resolved NIP-31.
---------------------------
Assignee: Mark Payne
Resolution: Done
Vote passed; vote thread:
https://lists.apache.org/thread/b88m355dbt246df94gq0ryx03pvklmhb
> Processor / Connector Backlog Reporting
> ---------------------------------------
>
> Key: NIP-31
> URL: https://issues.apache.org/jira/browse/NIP-31
> Project: NiFi Improvement Proposal
> Issue Type: New Feature
> Reporter: Mark Payne
> Assignee: Mark Payne
> Priority: Major
>
> h2. Motivation
> NiFi shows operators how much data is queued _inside_ a flow, but has no way
> to
> report how much data is still waiting on the _source_ system that a flow pulls
> from and has not yet ingested. An operator can see that a ConsumeKafka
> processor
> has 0 FlowFiles queued, but cannot tell from NiFi whether the topic it
> consumes
> has 10 unconsumed messages or 10 million.
> This "source backlog" is the signal operators need to answer "is this flow
> caught
> up?" and "is it keeping pace with its source?" Today they must use the
> external
> system's own tooling to find out. This proposal adds a first-class way for a
> source Processor (or a Connector built around one) to report its source
> backlog,
> and surfaces that information through the REST API and the UI.
> h2. Proposed Solution
> Introduce an opt-in capability that a source component implements to report
> its
> backlog:
> * A new {{Backlog}} value type describing what remains on the source: number
> of
> FlowFiles, bytes, and records; the time the component was last fully caught
> up;
> and a precision indicating whether those numbers are exact or a lower bound.
> * A {{BacklogReportingProcessor}} interface that a Processor implements to
> report
> a {{{}Backlog{}}}. A Processor opts in by implementing the interface;
> Processors that
> do not implement it are never asked.
> * A {{BacklogReportingConnector}} interface that does the same for
> Connectors. A
> Connector typically derives its {{Backlog}} by delegating to a source
> Processor
> in its managed flow, but may also inspect queued data.
> The framework exposes the reported {{Backlog}} through new REST endpoints and
> a
> UI dialog. In a cluster, each node reports its own view and the responses are
> aggregated into a single cluster-wide {{{}Backlog{}}}.
> Computing a backlog frequently requires the component to make a request to the
> external source system, so this is treated as a {*}WRITE{*}-authorized
> operation even
> though it is conceptually a read.
> A reference set of source components will implement the capability as part of
> this
> work: ConsumeKafka, ListS3, and ConsumeKinesis, plus the KafkaToS3 Connector
> (which delegates to its ConsumeKafka).
> h2. NiFi API Changes
> New, additive types (no existing API is removed or changed in a breaking way):
> * {{org.apache.nifi.components.Backlog}} — the value type, with a
> {{Precision}}
> enum ({{{}EXACT{}}}, {{{}AT_LEAST{}}}) and factory methods for the common
> "caught up" cases.
> * {{org.apache.nifi.components.BacklogReportingException}} — thrown by a
> component
> that attempted to determine its backlog and failed (for example, the source is
> unreachable).
> * {{org.apache.nifi.processor.BacklogReportingProcessor}} — opt-in capability
> interface; {{{}getBacklog(ProcessContext){}}}.
> * {{org.apache.nifi.components.connector.BacklogReportingConnector}} — opt-in
> capability interface; {{{}getBacklog(FlowContext){}}}.
> * {{org.apache.nifi.components.connector.components.QueueSnapshot}} — an
> atomic,
> point-in-time view of a connection's queue, used by Connectors that compute
> backlog from queued data.
> Additive methods are also added to the framework-supplied facade interfaces
> {{ProcessorFacade}} ({{{}reportsBacklog(){}}}, {{{}getBacklog(){}}}) and
> {{ConnectionFacade}} ({{{}getQueueSnapshot(){}}}). These facades are
> implemented only
> by the framework, not by extensions, so adding methods does not break existing
> extensions.
> h2. REST API Changes
> * GET /processors/ \{id}/backlog — returns the Processor's {{{}Backlog{}}}.
> * A new {{{}BacklogEntity{}}}/{{{}BacklogDTO{}}} carrying both raw numeric
> values and
> server-formatted, human-readable strings.
> Both require *WRITE* authorization. A component that does not implement the
> capability returns {{{}409 Conflict{}}}; a {{BacklogReportingException}} is
> mapped to
> an error response that is distinguishable from "not supported."
> h2. Framework Changes
> * {{FlowFileQueue}} gains an atomic queue-snapshot operation so that a
> connection's size and its FlowFile list are captured together at a single
> point
> in time (required for Connectors that compute backlog from queued FlowFiles).
> * Cluster aggregation merges per-node {{Backlog}} responses: numeric
> dimensions
> are summed, the "last caught up" time is the earliest across nodes, and
> precision
> is downgraded to {{AT_LEAST}} if any node is inexact.
> * Human-readable formatting (counts, data sizes, and local-time "last caught
> up")
> is computed server-side so that the UI and any other client render
> consistently.
> h2. UI Changes
> * A context-menu entry on eligible Processors and Connectors fetches the
> current
> {{{}Backlog{}}}.
> * The {{Backlog}} is shown in a dialog laid out as a vertical two-column
> table
> (measurement / value), with pretty-printed numbers and the raw value on hover,
> local-time "last caught up" with timezone, and "No value set" for unreported
> dimensions.
> h2. Scope / Not in Scope
> In scope: the API capability, REST endpoints, cluster aggregation, UI
> presentation, and reference implementations on ConsumeKafka, ListS3,
> ConsumeKinesis, and the KafkaToS3 Connector.
> Not in scope: implementing the capability on every source Processor.
> Components
> opt in over time; a component that does not implement the interface behaves
> exactly as it does today.
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
