diff --git a/src/main/java/org/apache/nifi/components/Backlog.java b/src/main/java/org/apache/nifi/components/Backlog.java
new file mode 100644
index 0000000..29402b7
--- /dev/null
+++ b/src/main/java/org/apache/nifi/components/Backlog.java
@@ -0,0 +1,437 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.components;
+
+import java.time.Instant;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.OptionalLong;
+
+/**
+ * <p>
+ *     An immutable description of how much work remains for a Connector or Processor
+ *     to consume from its source system. A {@code Backlog} may report any combination of
+ *     FlowFile count, byte count, and record count, and may also indicate the most recent
+ *     time at which the component observed itself as fully caught up with the source.
+ * </p>
+ *
+ * <p>
+ *     All four dimensions are optional and independent. A component reports only those
+ *     dimensions that it can determine. For example, a Connector pulling from an object
+ *     store may know both the number of objects remaining and the total bytes those
+ *     objects represent, while a Connector pulling from a streaming system may know only
+ *     the number of records remaining.
+ * </p>
+ *
+ * <p>
+ *     The numeric dimensions are interpreted in light of the {@link Precision} attribute,
+ *     which applies to the entire {@code Backlog}. {@link Precision#EXACT} (the default)
+ *     means the values are exact counts; {@link Precision#AT_LEAST} means the values are
+ *     lower bounds and the real counts may be higher.
+ * </p>
+ *
+ * <p>
+ *     The {@link #getLastCaughtUp() lastCaughtUp} timestamp is unaffected by precision and
+ *     is always treated as exact. It represents the most recent moment the component
+ *     determined that there was zero data remaining to pull from the source. It is computed
+ *     by the component using whatever knowledge it has — typically during its normal polling
+ *     of the source — and is not merely the time a backlog query last returned zero.
+ * </p>
+ *
+ * <p>
+ *     Backlogs are composed via {@link #plus(Backlog)}; see that method for the per-field
+ *     combination rules.
+ * </p>
+ *
+ * <p>
+ *     This API is experimental and may change without notice between releases.
+ * </p>
+ */
+public final class Backlog {
+
+    private final OptionalLong flowFileCount;
+    private final OptionalLong byteCount;
+    private final OptionalLong recordCount;
+    private final Optional<Instant> lastCaughtUp;
+    private final Precision precision;
+
+    private Backlog(final Builder builder) {
+        this.flowFileCount = builder.flowFileCount;
+        this.byteCount = builder.byteCount;
+        this.recordCount = builder.recordCount;
+        this.lastCaughtUp = builder.lastCaughtUp;
+        this.precision = builder.precision;
+    }
+
+    /**
+     * @return the number of FlowFiles remaining on the source, if known
+     */
+    public OptionalLong getFlowFileCount() {
+        return flowFileCount;
+    }
+
+    /**
+     * @return the total number of bytes remaining on the source, if known
+     */
+    public OptionalLong getByteCount() {
+        return byteCount;
+    }
+
+    /**
+     * @return the number of records remaining on the source, if known
+     */
+    public OptionalLong getRecordCount() {
+        return recordCount;
+    }
+
+    /**
+     * @return the most recent moment the component observed itself as fully caught up with the source, if known
+     */
+    public Optional<Instant> getLastCaughtUp() {
+        return lastCaughtUp;
+    }
+
+    /**
+     * @return the {@link Precision} of the numeric dimensions of this Backlog. Never null.
+     */
+    public Precision getPrecision() {
+        return precision;
+    }
+
+    /**
+     * <p>
+     *     Combines this Backlog with another to produce a new {@code Backlog} describing
+     *     the union of the two. Useful for composing a flow-wide backlog from multiple sources
+     *     such as multiple Processor reports and queue snapshots.
+     * </p>
+     *
+     * <p>
+     *     Combination rules:
+     * </p>
+     * <ul>
+     *     <li>
+     *         Numeric dimensions ({@code flowFileCount}, {@code byteCount}, {@code recordCount}):
+     *         if both sides have a value, the result is their sum. If only one side has a value,
+     *         it is carried through but the combined {@link Precision} is downgraded to
+     *         {@link Precision#AT_LEAST} (see the precision rule below) when the other side reports
+     *         at least one numeric dimension, because the side that did not report the dimension is
+     *         unknown rather than zero. If the other side reports no numeric dimensions at all, the
+     *         known value is carried through without adding numeric uncertainty. If neither side has
+     *         a value, the field stays empty.
+     *     </li>
+     *     <li>
+     *         {@link #getLastCaughtUp() lastCaughtUp}: <b>not summable.</b> If both sides have a value,
+     *         the result is the earlier of the two timestamps — the more conservative claim about how
+     *         recently the system was fully caught up. If only one side has it, that one is carried
+     *         through.
+     *     </li>
+     *     <li>
+     *         {@link #getPrecision() precision}: the result is {@link Precision#EXACT} only when both
+     *         sides are {@code EXACT} <i>and</i> either both sides report the same set of populated
+     *         numeric dimensions or one side reports no numeric dimensions at all. Otherwise the
+     *         result is {@link Precision#AT_LEAST}. Any uncertainty in either operand — including a
+     *         missing dimension on one side that the other side reported — taints the result,
+     *         because "unknown" must not be treated as zero.
+     *     </li>
+     * </ul>
+     *
+     * @param other the other Backlog to combine with this one
+     * @return a new Backlog representing the combination
+     * @throws ArithmeticException if a numeric sum would overflow {@code long}
+     */
+    public Backlog plus(final Backlog other) {
+        Objects.requireNonNull(other, "other Backlog must not be null");
+
+        final OptionalLong sumFlowFiles = sumOptional(flowFileCount, other.flowFileCount);
+        final OptionalLong sumBytes = sumOptional(byteCount, other.byteCount);
+        final OptionalLong sumRecords = sumOptional(recordCount, other.recordCount);
+        final Optional<Instant> earliestCaughtUp = earlierOf(lastCaughtUp, other.lastCaughtUp);
+
+        final boolean reportsNumericDimensions = flowFileCount.isPresent() || byteCount.isPresent() || recordCount.isPresent();
+        final boolean otherReportsNumericDimensions = other.flowFileCount.isPresent() || other.byteCount.isPresent() || other.recordCount.isPresent();
+
+        // A dimension that one side reports and the other does not is "unknown" on the omitting side,
+        // not zero. Carrying the known value forward and still reporting EXACT would let the result
+        // claim completeness it cannot back up, so any such asymmetry forces AT_LEAST. However, a
+        // Backlog with no numeric dimensions contributes only non-numeric information, such as a
+        // lastCaughtUp timestamp, and therefore does not make numeric counts less exact.
+        final boolean dimensionsAsymmetric = reportsNumericDimensions && otherReportsNumericDimensions
+                && (flowFileCount.isPresent() != other.flowFileCount.isPresent()
+                || byteCount.isPresent() != other.byteCount.isPresent()
+                || recordCount.isPresent() != other.recordCount.isPresent());
+        final boolean precisionExact = precision == Precision.EXACT || !reportsNumericDimensions;
+        final boolean otherPrecisionExact = other.precision == Precision.EXACT || !otherReportsNumericDimensions;
+        final Precision combinedPrecision = (precisionExact && otherPrecisionExact && !dimensionsAsymmetric) ? Precision.EXACT : Precision.AT_LEAST;
+
+        final Builder builder = new Builder().precision(combinedPrecision);
+        if (sumFlowFiles.isPresent()) {
+            builder.flowFiles(sumFlowFiles.getAsLong());
+        }
+
+        if (sumBytes.isPresent()) {
+            builder.bytes(sumBytes.getAsLong());
+        }
+
+        if (sumRecords.isPresent()) {
+            builder.records(sumRecords.getAsLong());
+        }
+
+        earliestCaughtUp.ifPresent(builder::lastCaughtUp);
+        return builder.build();
+    }
+
+    private static OptionalLong sumOptional(final OptionalLong left, final OptionalLong right) {
+        if (left.isPresent() && right.isPresent()) {
+            return OptionalLong.of(Math.addExact(left.getAsLong(), right.getAsLong()));
+        }
+
+        if (left.isPresent()) {
+            return left;
+        }
+
+        if (right.isPresent()) {
+            return right;
+        }
+
+        return OptionalLong.empty();
+    }
+
+    private static Optional<Instant> earlierOf(final Optional<Instant> left, final Optional<Instant> right) {
+        if (left.isPresent() && right.isPresent()) {
+            return left.get().isBefore(right.get()) ? left : right;
+        }
+
+        if (left.isPresent()) {
+            return left;
+        }
+
+        return right;
+    }
+
+    /**
+     * Creates a Backlog whose only populated dimension is the FlowFile count, with {@link Precision#EXACT}.
+     *
+     * @param count the number of FlowFiles remaining on the source; must be non-negative
+     * @return a new Backlog
+     */
+    public static Backlog flowFiles(final long count) {
+        return new Builder().flowFiles(count).build();
+    }
+
+    /**
+     * Creates a Backlog whose only populated dimension is the byte count, with {@link Precision#EXACT}.
+     *
+     * @param bytes the number of bytes remaining on the source; must be non-negative
+     * @return a new Backlog
+     */
+    public static Backlog bytes(final long bytes) {
+        return new Builder().bytes(bytes).build();
+    }
+
+    /**
+     * Creates a Backlog whose only populated dimension is the record count, with {@link Precision#EXACT}.
+     *
+     * @param count the number of records remaining on the source; must be non-negative
+     * @return a new Backlog
+     */
+    public static Backlog records(final long count) {
+        return new Builder().records(count).build();
+    }
+
+    /**
+     * Creates a Backlog populated with the provided {@code lastCaughtUp} timestamp. Useful for
+     * combining with a count-only Backlog via {@link #plus(Backlog)}.
+     *
+     * @param instant the moment at which the component was last observed as fully caught up; may be
+     *        null, in which case the returned Backlog has no populated dimensions
+     * @return a new Backlog
+     */
+    public static Backlog lastCaughtUp(final Instant instant) {
+        return new Builder().lastCaughtUp(instant).build();
+    }
+
+    /**
+     * <p>
+     *     Returns the canonical "fully caught up" Backlog: zero FlowFiles, zero bytes, zero records,
+     *     and a {@code lastCaughtUp} timestamp of {@link Instant#now()}.
+     * </p>
+     *
+     * <p>
+     *     This state is the unambiguous "I am fully synchronized with the source as of now" signal.
+     * </p>
+     *
+     * @return a new "caught up" Backlog
+     */
+    public static Backlog caughtUp() {
+        return new Builder()
+                .flowFiles(0L)
+                .bytes(0L)
+                .records(0L)
+                .lastCaughtUp(Instant.now())
+                .build();
+    }
+
+    /**
+     * @return a new {@link Builder} for constructing a Backlog with multiple dimensions and/or {@link Precision#AT_LEAST}
+     */
+    public static Builder builder() {
+        return new Builder();
+    }
+
+    @Override
+    public boolean equals(final Object object) {
+        if (this == object) {
+            return true;
+        }
+
+        if (!(object instanceof Backlog)) {
+            return false;
+        }
+
+        final Backlog other = (Backlog) object;
+        return Objects.equals(flowFileCount, other.flowFileCount)
+                && Objects.equals(byteCount, other.byteCount)
+                && Objects.equals(recordCount, other.recordCount)
+                && Objects.equals(lastCaughtUp, other.lastCaughtUp)
+                && precision == other.precision;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(flowFileCount, byteCount, recordCount, lastCaughtUp, precision);
+    }
+
+    @Override
+    public String toString() {
+        return "Backlog["
+                + "flowFileCount=" + flowFileCount
+                + ", byteCount=" + byteCount
+                + ", recordCount=" + recordCount
+                + ", lastCaughtUp=" + lastCaughtUp
+                + ", precision=" + precision
+                + "]";
+    }
+
+    /**
+     * Builder for constructing a {@link Backlog} that populates more than one dimension or uses a
+     * non-default {@link Precision}. For single-dimension Backlogs, prefer the static factory methods
+     * such as {@link Backlog#flowFiles(long)}.
+     */
+    public static final class Builder {
+
+        private OptionalLong flowFileCount = OptionalLong.empty();
+        private OptionalLong byteCount = OptionalLong.empty();
+        private OptionalLong recordCount = OptionalLong.empty();
+        private Optional<Instant> lastCaughtUp = Optional.empty();
+        private Precision precision = Precision.EXACT;
+
+        /**
+         * Sets the FlowFile count. Overwrites any previously set value.
+         *
+         * @param count the number of FlowFiles remaining on the source; must be non-negative
+         * @return this Builder
+         * @throws IllegalArgumentException if {@code count} is negative
+         */
+        public Builder flowFiles(final long count) {
+            requireNonNegative("flowFiles", count);
+            this.flowFileCount = OptionalLong.of(count);
+            return this;
+        }
+
+        /**
+         * Sets the byte count. Overwrites any previously set value.
+         *
+         * @param bytes the number of bytes remaining on the source; must be non-negative
+         * @return this Builder
+         * @throws IllegalArgumentException if {@code bytes} is negative
+         */
+        public Builder bytes(final long bytes) {
+            requireNonNegative("bytes", bytes);
+            this.byteCount = OptionalLong.of(bytes);
+            return this;
+        }
+
+        /**
+         * Sets the record count. Overwrites any previously set value.
+         *
+         * @param count the number of records remaining on the source; must be non-negative
+         * @return this Builder
+         * @throws IllegalArgumentException if {@code count} is negative
+         */
+        public Builder records(final long count) {
+            requireNonNegative("records", count);
+            this.recordCount = OptionalLong.of(count);
+            return this;
+        }
+
+        /**
+         * Sets the timestamp at which the component was last observed as fully caught up. Overwrites
+         * any previously set value.
+         *
+         * @param instant the moment at which the component was last fully caught up; may be null to clear
+         * @return this Builder
+         */
+        public Builder lastCaughtUp(final Instant instant) {
+            this.lastCaughtUp = Optional.ofNullable(instant);
+            return this;
+        }
+
+        /**
+         * Sets the {@link Precision} for this Backlog. Defaults to {@link Precision#EXACT} if not set.
+         *
+         * @param precision the precision; must not be null
+         * @return this Builder
+         */
+        public Builder precision(final Precision precision) {
+            this.precision = Objects.requireNonNull(precision, "precision must not be null");
+            return this;
+        }
+
+        /**
+         * @return a new immutable {@link Backlog} reflecting the values set on this Builder
+         */
+        public Backlog build() {
+            return new Backlog(this);
+        }
+
+        private static void requireNonNegative(final String fieldName, final long value) {
+            if (value < 0L) {
+                throw new IllegalArgumentException(fieldName + " must be non-negative but was " + value);
+            }
+        }
+    }
+
+    /**
+     * <p>
+     *     The precision of the numeric dimensions of a {@link Backlog}. Applies to the entire Backlog
+     *     rather than to individual dimensions.
+     * </p>
+     */
+    public enum Precision {
+
+        /**
+         * Numeric dimensions reflect exact counts.
+         */
+        EXACT,
+
+        /**
+         * Numeric dimensions are lower bounds; the real counts may be higher.
+         */
+        AT_LEAST
+    }
+}
diff --git a/src/main/java/org/apache/nifi/components/BacklogReportingException.java b/src/main/java/org/apache/nifi/components/BacklogReportingException.java
new file mode 100644
index 0000000..942c62c
--- /dev/null
+++ b/src/main/java/org/apache/nifi/components/BacklogReportingException.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.components;
+
+/**
+ * <p>
+ *     Thrown by a Connector or Processor that supports backlog reporting but cannot
+ *     determine its current backlog due to a transient or permanent failure. Typical causes
+ *     include I/O errors reaching the source system, authorization failures (the component is
+ *     allowed to consume data but is not authorized to call the offset or listing API), and
+ *     state errors that prevent computation.
+ * </p>
+ *
+ * <p>
+ *     This exception is distinct from a Connector or Processor not supporting backlog reporting
+ *     at all. Components that do not support backlog reporting simply do not implement
+ *     {@link org.apache.nifi.processor.BacklogReportingProcessor} or
+ *     {@link org.apache.nifi.components.connector.BacklogReportingConnector}. Components that do
+ *     support backlog reporting may return {@link java.util.Optional#empty()} when they cannot
+ *     determine a value right now, or throw this exception when an attempt to determine the
+ *     backlog fails. Callers should surface the cause rather than silently treat the failure as
+ *     "not supported".
+ * </p>
+ */
+public class BacklogReportingException extends Exception {
+
+    public BacklogReportingException(final String message) {
+        super(message);
+    }
+
+    public BacklogReportingException(final String message, final Throwable cause) {
+        super(message, cause);
+    }
+
+    public BacklogReportingException(final Throwable cause) {
+        super(cause);
+    }
+}
diff --git a/src/main/java/org/apache/nifi/components/connector/BacklogReportingConnector.java b/src/main/java/org/apache/nifi/components/connector/BacklogReportingConnector.java
new file mode 100644
index 0000000..b71f8b6
--- /dev/null
+++ b/src/main/java/org/apache/nifi/components/connector/BacklogReportingConnector.java
@@ -0,0 +1,102 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.components.connector;
+
+import org.apache.nifi.components.Backlog;
+import org.apache.nifi.components.BacklogReportingException;
+import org.apache.nifi.components.connector.components.FlowContext;
+import org.apache.nifi.components.connector.components.ProcessorFacade;
+
+import java.util.Optional;
+
+/**
+ * <p>
+ *     Optional capability interface implemented by {@link Connector}s that can report a
+ *     {@link Backlog} describing how much data remains on the source system(s) the Connector
+ *     pulls from but has not yet brought into NiFi. Implementing this interface is the opt-in
+ *     signal: Connectors that do not implement it are simply not asked for backlog, and the
+ *     framework's REST endpoints return HTTP {@code 409 Conflict} rather than treating "not
+ *     supported" the same as "supported but currently unknown".
+ * </p>
+ *
+ * <p>
+ *     This mirrors the {@code BacklogReportingProcessor} capability interface on the Processor
+ *     side: the {@code Connector} base interface stays focused on lifecycle and configuration,
+ *     and backlog reporting is layered on as a separate opt-in.
+ * </p>
+ *
+ * <p>
+ *     Implementations that do report backlog typically derive the value from a single source —
+ *     most often a single Processor's {@code BacklogReportingProcessor} report exposed via
+ *     {@link ProcessorFacade#getBacklog()}. Some Connectors may compose a flow-wide picture from
+ *     multiple sources: for example, summing the {@link Backlog} from multiple Processors, or, for
+ *     a List/Fetch pattern, also inspecting queues and FlowFile attributes to account for data that
+ *     has been "Listed" but not yet "Fetched".
+ * </p>
+ *
+ * <h2>Return value semantics</h2>
+ * <ul>
+ *     <li>
+ *         {@link Optional#empty()} means the Connector cannot report a {@link Backlog} right now,
+ *         even though it implements this interface. This is distinct from a hard failure.
+ *     </li>
+ *     <li>
+ *         <code>Optional.of(Backlog.caughtUp())</code> means the Connector is fully synchronized
+ *         with its source as of now.
+ *     </li>
+ *     <li>
+ *         <code>Optional.of(...)</code> with non-zero numeric dimensions means the Connector knows
+ *         that this much data remains on the source. See {@link Backlog.Precision} for whether the
+ *         numbers are exact or lower bounds.
+ *     </li>
+ * </ul>
+ *
+ * <p>
+ *     Hard failures — source unreachable, authorization denied, state errors that prevent
+ *     computation — are reported by throwing {@link BacklogReportingException}, not by returning
+ *     {@link Optional#empty()}. This keeps failures distinguishable from "no value available."
+ * </p>
+ *
+ * <p>
+ *     <b>Implementation Note:</b> This API is currently experimental, as it is under very active
+ *     development. As such, it is subject to change without notice between releases.
+ * </p>
+ */
+public interface BacklogReportingConnector {
+
+    /**
+     * <p>
+     *     Reports how much data remains on the source system(s) that this Connector pulls from but
+     *     has not yet brought into NiFi.
+     * </p>
+     *
+     * <p>
+     *     This method is invoked against the <b>active</b> {@link FlowContext}, because backlog is
+     *     a runtime property of the running flow. The working flow context is not used for backlog
+     *     reporting. Expected call frequency is low — typically on the order of once per minute,
+     *     sometimes once every few minutes. Implementations may cache results internally if computing
+     *     backlog puts non-trivial load on the source, but caching is not required.
+     * </p>
+     *
+     * @param activeFlowContext the active flow context against which to compute backlog
+     * @return the Connector's reported {@link Backlog}, or {@link Optional#empty()} if the Connector
+     *         cannot determine a value right now (for example, it has never polled the source)
+     * @throws BacklogReportingException if the Connector attempted to determine its backlog and failed
+     */
+    Optional<Backlog> getBacklog(final FlowContext activeFlowContext) throws BacklogReportingException;
+}
diff --git a/src/main/java/org/apache/nifi/components/connector/Connector.java b/src/main/java/org/apache/nifi/components/connector/Connector.java
index 2fcafb0..36d2998 100644
--- a/src/main/java/org/apache/nifi/components/connector/Connector.java
+++ b/src/main/java/org/apache/nifi/components/connector/Connector.java
@@ -32,7 +32,7 @@
  *     A Connector is a component that encapsulates and manages a NiFi flow, in such a way that the flow
  *     can be treated as a single component. The Connector is responsible for managing the lifecycle of the flow,
  *     including starting and stopping the flow, as well as validating that the flow is correctly configured.
- *     The Connector exposes a single holistic configuration that is encapsulates the configuration of the
+ *     The Connector exposes a single holistic configuration that encapsulates the configuration of the
  *     sources, sinks, transformations, routing logic, and any other components that make up the flow.
  * </p>
  *
@@ -117,7 +117,7 @@ public interface Connector {
 
     /**
      * Validates that the Connector is valid according to its current configuration. Validity of a Connector may be
-     * defined simply as the all components being valid, or it may encompass more complex validation logic, such
+     * defined simply as all components being valid, or it may encompass more complex validation logic, such
      * as ensuring that a Source Processor is able to connect to a remote system, or that a Sink Processor
      * is able to write to a remote system.
      *
@@ -131,7 +131,7 @@ public interface Connector {
 
     /**
      * Validates the configuration for a specific configuration step. This allows the Connector to indicate any
-     * issues with syntactic configuration issues but is not as comprehensive as the overall validation provided
+     * syntax issues in the configuration but is not as comprehensive as the overall validation provided
      * by {@link #validate(FlowContext, ConnectorValidationContext)} due to the fact that it does not have access
      * to the full configuration of the Connector. This provides immediate feedback to users
      * as they are configuring each step.
@@ -146,8 +146,8 @@ public interface Connector {
 
     /**
      * Verifies the configuration for a specific configuration step. This allows the Connector to perform
-     * more comprehensive verification of the configuration for a step than does validation, such as attempting to connect to
-     * remote systems, sample data and ensure that it can be parsed correctly, etc.
+     * more comprehensive verification of the configuration for a step than does validation, such as connecting to
+     * remote systems or sampling data and ensuring that it can be parsed correctly.
      *
      * @param stepName the name of the configuration step being verified
      * @param propertyValueOverrides any overrides to the currently configured property values that should be used for verification
@@ -181,6 +181,7 @@ public interface Connector {
      *
      * @param stepName the name of the step
      * @param workingFlowContext the working flow context that is being used for the update
+     * @throws FlowUpdateException if there is an error updating the flow for the configured step
      */
     void onConfigurationStepConfigured(String stepName, FlowContext workingFlowContext) throws FlowUpdateException;
 
@@ -190,6 +191,7 @@ public interface Connector {
      *
      * @param workingFlowContext the working flow context that has been created for the update
      * @param activeFlowContext the active flow context that is currently in use
+     * @throws FlowUpdateException if the Connector cannot prepare the flow for the update
      */
     void prepareForUpdate(FlowContext workingFlowContext, FlowContext activeFlowContext) throws FlowUpdateException;
 
@@ -210,6 +212,7 @@ public interface Connector {
      *
      * @param workingFlowContext the working flow context that represents the updated configuration
      * @param activeFlowContext the flow context that represents the active flow
+     * @throws FlowUpdateException if the updated configuration cannot be applied to the active flow
      */
     void applyUpdate(FlowContext workingFlowContext, FlowContext activeFlowContext) throws FlowUpdateException;
 
diff --git a/src/main/java/org/apache/nifi/components/connector/components/ConnectionFacade.java b/src/main/java/org/apache/nifi/components/connector/components/ConnectionFacade.java
index ff17d2b..79ea9b9 100644
--- a/src/main/java/org/apache/nifi/components/connector/components/ConnectionFacade.java
+++ b/src/main/java/org/apache/nifi/components/connector/components/ConnectionFacade.java
@@ -25,6 +25,12 @@
 import java.io.IOException;
 import java.util.function.Predicate;
 
+/**
+ * <p>
+ *     Facade exposing per-Connection operations to a Connector implementation. The framework constructs
+ *     and supplies these facades; Connector extensions do not implement this interface themselves.
+ * </p>
+ */
 public interface ConnectionFacade {
 
     VersionedConnection getDefinition();
@@ -49,4 +55,12 @@ public interface ConnectionFacade {
      */
     DropFlowFileSummary dropFlowFiles(Predicate<FlowFile> predicate) throws IOException;
 
+    /**
+     * Returns a read-only point-in-time snapshot of this connection's queue. See
+     * {@link QueueSnapshot} for the snapshot semantics.
+     *
+     * @return a non-null snapshot of the connection's queue
+     */
+    QueueSnapshot getQueueSnapshot();
+
 }
diff --git a/src/main/java/org/apache/nifi/components/connector/components/ProcessorFacade.java b/src/main/java/org/apache/nifi/components/connector/components/ProcessorFacade.java
index bb57e3f..23f4f22 100644
--- a/src/main/java/org/apache/nifi/components/connector/components/ProcessorFacade.java
+++ b/src/main/java/org/apache/nifi/components/connector/components/ProcessorFacade.java
@@ -17,16 +17,26 @@
 
 package org.apache.nifi.components.connector.components;
 
+import org.apache.nifi.components.Backlog;
+import org.apache.nifi.components.BacklogReportingException;
 import org.apache.nifi.components.ConfigVerificationResult;
 import org.apache.nifi.components.ValidationResult;
 import org.apache.nifi.components.connector.InvocationFailedException;
 import org.apache.nifi.flow.VersionedExternalFlow;
 import org.apache.nifi.flow.VersionedParameterContext;
 import org.apache.nifi.flow.VersionedProcessor;
+import org.apache.nifi.processor.BacklogReportingProcessor;
 
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 
+/**
+ * <p>
+ *     Facade exposing per-Processor operations to a Connector implementation. The framework constructs
+ *     and supplies these facades; Connector extensions do not implement this interface themselves.
+ * </p>
+ */
 public interface ProcessorFacade {
 
     VersionedProcessor getDefinition();
@@ -75,4 +85,32 @@ public interface ProcessorFacade {
      * @throws InvocationFailedException if unable to invoke the method
      */
     <T> T invokeConnectorMethod(String methodName, Map<String, Object> arguments, Class<T> returnType) throws InvocationFailedException;
+
+    /**
+     * Indicates whether the underlying Processor implements {@link BacklogReportingProcessor}. This
+     * is a cheap capability check that does not call into the Processor.
+     *
+     * @return {@code true} if the underlying Processor implements {@link BacklogReportingProcessor};
+     *         {@code false} otherwise
+     */
+    boolean reportsBacklog();
+
+    /**
+     * <p>
+     *     Returns the underlying Processor's reported {@link Backlog}. This is the bridge a
+     *     Connector uses to ask a Processor in its flow how much data remains on the source.
+     * </p>
+     *
+     * <p>
+     *     The return value and exception semantics mirror those of
+     *     {@link BacklogReportingProcessor#getBacklog(org.apache.nifi.processor.ProcessContext)}. If
+     *     the underlying Processor does not implement {@link BacklogReportingProcessor}, this method
+     *     returns {@link Optional#empty()}.
+     * </p>
+     *
+     * @return the Processor's reported {@link Backlog}, or {@link Optional#empty()} if the Processor does not
+     *         implement {@link BacklogReportingProcessor} or has nothing to report
+     * @throws BacklogReportingException if the Processor attempted to determine its backlog and failed
+     */
+    Optional<Backlog> getBacklog() throws BacklogReportingException;
 }
diff --git a/src/main/java/org/apache/nifi/components/connector/components/QueueSnapshot.java b/src/main/java/org/apache/nifi/components/connector/components/QueueSnapshot.java
new file mode 100644
index 0000000..da520da
--- /dev/null
+++ b/src/main/java/org/apache/nifi/components/connector/components/QueueSnapshot.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.components.connector.components;
+
+import org.apache.nifi.controller.queue.QueueSize;
+import org.apache.nifi.flowfile.FlowFile;
+
+import java.util.List;
+
+/**
+ * <p>
+ *     A point-in-time view of a connection's queue, returned by
+ *     {@link ConnectionFacade#getQueueSnapshot()}. Bundles the connection's total {@link QueueSize}
+ *     together with the {@link FlowFile}s currently held in this node's active in-memory queue,
+ *     meaning the FlowFiles resident in memory on this node and available in poll order.
+ * </p>
+ *
+ * <p>
+ *     The active list and the {@link QueueSize} are captured atomically, so they describe the same
+ *     point in time. The active list is not always the entire queue. FlowFiles that have been
+ *     swapped out are excluded from {@link #getActiveFlowFiles()}. On load-balanced connections,
+ *     FlowFiles in flight between nodes are excluded as well. Those FlowFiles are still counted in
+ *     {@link #getQueueSize()}. Use {@link #isActiveListExhaustive()} to determine whether the
+ *     active list contains every FlowFile counted in the queue size.
+ * </p>
+ *
+ * <p>
+ *     {@link #getActiveFlowFiles()} returns the full active list without a caller-supplied limit, but
+ *     the list itself is bounded by the framework's maximum active in-memory queue size. The
+ *     snapshot does not clone the {@link FlowFile} instances; it references the existing active
+ *     FlowFiles.
+ * </p>
+ */
+public interface QueueSnapshot {
+
+    /**
+     * @return the total {@link QueueSize} of the connection at the time the snapshot was taken
+     */
+    QueueSize getQueueSize();
+
+    /**
+     * @return the active in-memory FlowFiles in poll order, never null; excludes FlowFiles that
+     *         have been swapped out and, for load-balanced connections, FlowFiles in flight between
+     *         nodes
+     */
+    List<FlowFile> getActiveFlowFiles();
+
+    /**
+     * @return {@code true} if {@link #getActiveFlowFiles()} contains every FlowFile counted in
+     *         {@link #getQueueSize()}; otherwise {@code false}
+     */
+    boolean isActiveListExhaustive();
+}
diff --git a/src/main/java/org/apache/nifi/processor/BacklogReportingProcessor.java b/src/main/java/org/apache/nifi/processor/BacklogReportingProcessor.java
new file mode 100644
index 0000000..24b3df0
--- /dev/null
+++ b/src/main/java/org/apache/nifi/processor/BacklogReportingProcessor.java
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.nifi.processor;
+
+import org.apache.nifi.components.Backlog;
+import org.apache.nifi.components.BacklogReportingException;
+
+import java.util.Optional;
+
+/**
+ * <p>
+ *     Optional capability interface implemented by Processors that can report a {@link Backlog}
+ *     describing how much data remains on the source system that the Processor has not yet pulled
+ *     into NiFi. Implementing this interface is the opt-in signal: Processors that do not implement
+ *     it are simply not asked for backlog.
+ * </p>
+ *
+ * <p>
+ *     The reported {@link Backlog} reflects only the data the Processor knows about from the source.
+ *     A Connector that composes a flow-wide backlog generally also accounts for FlowFiles already
+ *     enqueued in NiFi between processors; that composition is the Connector's responsibility, not
+ *     the Processor's.
+ * </p>
+ *
+ * <p>
+ *     The expected call frequency for {@link #getBacklog(ProcessContext)} is low — on the order of
+ *     once per minute, sometimes once every few minutes. Implementations may cache results internally
+ *     if computing backlog puts non-trivial load on the source, but caching is not required.
+ * </p>
+ *
+ * <h2>Return value semantics</h2>
+ * <ul>
+ *     <li>
+ *         {@link Optional#empty()} means the Processor cannot report a {@code Backlog} right now,
+ *         even though it implements this interface. This is distinct from a hard failure.
+ *     </li>
+ *     <li>
+ *         <code>Optional.of(Backlog.caughtUp())</code> means the Processor is fully
+ *         synchronized with its source as of now.
+ *     </li>
+ *     <li>
+ *         <code>Optional.of(...)</code> with non-zero numeric dimensions means the Processor knows that
+ *         this much data remains on the source. See {@link Backlog.Precision} for whether the
+ *         numbers are exact or lower bounds.
+ *     </li>
+ * </ul>
+ *
+ * <p>
+ *     Hard failures — source unreachable, authorization denied, state errors that prevent
+ *     computation — are reported by throwing {@link BacklogReportingException}, not by returning
+ *     {@link Optional#empty()}. This keeps failures distinguishable from "no value available."
+ * </p>
+ */
+public interface BacklogReportingProcessor {
+
+    /**
+     * Returns the Processor's current view of how much data remains on the source.
+     *
+     * <p>
+     *     The framework always supplies a non-null {@link ProcessContext}. Implementations may use
+     *     the supplied context to read property values and to build whatever transient clients they
+     *     need to query the source. Implementations are responsible for closing any transient
+     *     clients they open during the call.
+     * </p>
+     *
+     * <p>
+     *     The framework does not serialize calls to this method. Multiple {@code getBacklog}
+     *     invocations may run concurrently against the same Processor instance, so implementations
+     *     must be safe for concurrent use. Implementations that open per-call transient clients are
+     *     naturally isolated. Implementations that cache shared clients across calls must guard
+     *     them appropriately or rely on thread-safe SDK clients.
+     * </p>
+     *
+     * @param context a non-null {@link ProcessContext} usable to look up property values and to
+     *         construct transient clients to the source
+     * @return the Processor's reported {@link Backlog}, or {@link Optional#empty()} if the Processor
+     *         cannot determine a value right now (e.g. it has never polled the source)
+     * @throws BacklogReportingException if the Processor attempted to determine its backlog and
+     *         failed due to a transient or permanent error such as I/O failure or authorization denial
+     */
+    Optional<Backlog> getBacklog(final ProcessContext context) throws BacklogReportingException;
+}
diff --git a/src/test/java/org/apache/nifi/components/TestBacklog.java b/src/test/java/org/apache/nifi/components/TestBacklog.java
new file mode 100644
index 0000000..98b514e
--- /dev/null
+++ b/src/test/java/org/apache/nifi/components/TestBacklog.java
@@ -0,0 +1,144 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.components;
+
+import org.apache.nifi.components.Backlog.Precision;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.OptionalLong;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class TestBacklog {
+
+    @Test
+    public void testPlusSumsDimensionsAndKeepsExactWhenBothSidesReportTheSameShape() {
+        final Backlog left = Backlog.builder().flowFiles(10L).bytes(100L).build();
+        final Backlog right = Backlog.builder().flowFiles(5L).bytes(50L).build();
+
+        final Backlog combined = left.plus(right);
+
+        assertEquals(OptionalLong.of(15L), combined.getFlowFileCount());
+        assertEquals(OptionalLong.of(150L), combined.getByteCount());
+        assertFalse(combined.getRecordCount().isPresent());
+        assertEquals(Precision.EXACT, combined.getPrecision());
+    }
+
+    @Test
+    public void testPlusDowngradesToAtLeastWhenOneSideOmitsADimensionTheOtherReports() {
+        // Left knows only flowFiles; right knows only bytes. The result should still carry the known
+        // numeric values, but the precision must be AT_LEAST because each side is "unknown" with
+        // respect to the dimension the other side reported. Treating "unknown" as zero would let the
+        // combined Backlog falsely claim EXACT completeness.
+        final Backlog left = Backlog.flowFiles(10L);
+        final Backlog right = Backlog.bytes(100L);
+
+        final Backlog combined = left.plus(right);
+
+        assertEquals(OptionalLong.of(10L), combined.getFlowFileCount());
+        assertEquals(OptionalLong.of(100L), combined.getByteCount());
+        assertFalse(combined.getRecordCount().isPresent());
+        assertEquals(Precision.AT_LEAST, combined.getPrecision());
+    }
+
+    @Test
+    public void testPlusDowngradesToAtLeastWhenAsymmetricEvenIfBothSidesAreExact() {
+        // Both sides are EXACT individually but report different sets of dimensions. The combined
+        // view cannot be EXACT because each side is silent about a dimension the other side knows.
+        final Backlog left = Backlog.builder().flowFiles(1L).bytes(2L).precision(Precision.EXACT).build();
+        final Backlog right = Backlog.builder().flowFiles(3L).bytes(4L).records(5L).precision(Precision.EXACT).build();
+
+        final Backlog combined = left.plus(right);
+
+        assertEquals(OptionalLong.of(4L), combined.getFlowFileCount());
+        assertEquals(OptionalLong.of(6L), combined.getByteCount());
+        assertEquals(OptionalLong.of(5L), combined.getRecordCount());
+        assertEquals(Precision.AT_LEAST, combined.getPrecision());
+    }
+
+    @Test
+    public void testPlusPropagatesAtLeastFromEitherOperand() {
+        final Backlog left = Backlog.builder().flowFiles(1L).precision(Precision.AT_LEAST).build();
+        final Backlog right = Backlog.builder().flowFiles(2L).precision(Precision.EXACT).build();
+
+        assertEquals(Precision.AT_LEAST, left.plus(right).getPrecision());
+        assertEquals(Precision.AT_LEAST, right.plus(left).getPrecision());
+    }
+
+    @Test
+    public void testPlusKeepsExactWhenCombinedWithTimestampOnlyBacklog() {
+        final Instant lastCaughtUp = Instant.parse("2025-01-01T00:00:00Z");
+        final Backlog countOnly = Backlog.flowFiles(10L);
+        final Backlog timestampOnly = Backlog.lastCaughtUp(lastCaughtUp);
+
+        final Backlog combinedWithTimestampOnRight = countOnly.plus(timestampOnly);
+        assertEquals(OptionalLong.of(10L), combinedWithTimestampOnRight.getFlowFileCount());
+        assertEquals(Precision.EXACT, combinedWithTimestampOnRight.getPrecision());
+        assertEquals(lastCaughtUp, combinedWithTimestampOnRight.getLastCaughtUp().orElseThrow());
+
+        final Backlog combinedWithTimestampOnLeft = timestampOnly.plus(countOnly);
+        assertEquals(OptionalLong.of(10L), combinedWithTimestampOnLeft.getFlowFileCount());
+        assertEquals(Precision.EXACT, combinedWithTimestampOnLeft.getPrecision());
+        assertEquals(lastCaughtUp, combinedWithTimestampOnLeft.getLastCaughtUp().orElseThrow());
+    }
+
+    @Test
+    public void testPlusUsesEarlierLastCaughtUpAndKeepsOnlySideWhenOtherMissing() {
+        final Instant earlier = Instant.parse("2025-01-01T00:00:00Z");
+        final Instant later = Instant.parse("2025-01-02T00:00:00Z");
+
+        final Backlog withEarlier = Backlog.builder().flowFiles(0L).lastCaughtUp(earlier).build();
+        final Backlog withLater = Backlog.builder().flowFiles(0L).lastCaughtUp(later).build();
+        assertEquals(earlier, withEarlier.plus(withLater).getLastCaughtUp().orElseThrow());
+
+        final Backlog withoutTimestamp = Backlog.flowFiles(0L);
+        assertEquals(later, withoutTimestamp.plus(withLater).getLastCaughtUp().orElseThrow());
+    }
+
+    @Test
+    public void testPlusOverflowOnLongSum() {
+        final Backlog left = Backlog.flowFiles(Long.MAX_VALUE);
+        final Backlog right = Backlog.flowFiles(1L);
+        assertThrows(ArithmeticException.class, () -> left.plus(right));
+    }
+
+    @Test
+    public void testCaughtUpFactoryProducesZerosAndTimestamp() {
+        final Backlog caughtUp = Backlog.caughtUp();
+        assertEquals(OptionalLong.of(0L), caughtUp.getFlowFileCount());
+        assertEquals(OptionalLong.of(0L), caughtUp.getByteCount());
+        assertEquals(OptionalLong.of(0L), caughtUp.getRecordCount());
+        assertTrue(caughtUp.getLastCaughtUp().isPresent());
+        assertEquals(Precision.EXACT, caughtUp.getPrecision());
+    }
+
+    @Test
+    public void testBuilderRejectsNegativeNumericDimensions() {
+        final IllegalArgumentException flowFilesException = assertThrows(IllegalArgumentException.class, () -> Backlog.builder().flowFiles(-1L));
+        assertEquals("flowFiles must be non-negative but was -1", flowFilesException.getMessage());
+
+        final IllegalArgumentException bytesException = assertThrows(IllegalArgumentException.class, () -> Backlog.builder().bytes(-1L));
+        assertEquals("bytes must be non-negative but was -1", bytesException.getMessage());
+
+        final IllegalArgumentException recordsException = assertThrows(IllegalArgumentException.class, () -> Backlog.builder().records(-1L));
+        assertEquals("records must be non-negative but was -1", recordsException.getMessage());
+    }
+}