NIFI-16046: Initial implementation for Processor / Connector Backlog

nifi-commons/nifi-utils/src/main/java/org/apache/nifi/util/FormatUtils.java

@@ -134,6 +134,81 @@ public static String formatDataSize(final double dataSize) {
         return format.format(dataSize) + " bytes";
     }
 
+    /**
+     * Formats the gap between two instants as a coarse-grained, human-readable relative-time
+     * string, for example {@code "5 secs ago"}, {@code "in 2 mins"}, {@code "yesterday"},
+     * {@code "tomorrow"}, {@code "3 days ago"}, {@code "2 weeks ago"}, {@code "4 months ago"},
+     * or {@code "1 year ago"}. When {@code from} is before {@code to}, the result reads as
+     * "X ago"; when {@code from} is after {@code to}, the result reads as "in X". The unit
+     * granularity is chosen so that the displayed number is small but informative — sub-minute
+     * gaps surface as seconds, sub-hour gaps as minutes, and so on up through years.
+     *
+     * <p>This method does not return a special {@code "now"} string when the gap is small;
+     * callers that want to collapse a small window around {@code to} (for example, "(now)" when
+     * a backlog is current within a few seconds) should test for that condition themselves
+     * before invoking this method.</p>
+     *
+     * @param from the instant to describe relative to {@code to}, never {@code null}
+     * @param to the reference instant (usually "now"), never {@code null}
+     * @return a relative-time string describing the gap from {@code from} to {@code to}
+     */
+    public static String formatRelativeTime(final Instant from, final Instant to) {
+        final long differenceMillis = to.toEpochMilli() - from.toEpochMilli();
+        final boolean past = differenceMillis >= 0;
+        final long absoluteDifferenceMillis = Math.abs(differenceMillis);
+
+        final long secondMillis = 1000L;
+        final long minuteMillis = 60L * secondMillis;
+        final long hourMillis = 60L * minuteMillis;
+        final long dayMillis = 24L * hourMillis;
+        final long weekMillis = 7L * dayMillis;
+        final long monthMillis = 30L * dayMillis;
+        final long yearMillis = 365L * dayMillis;
+
+        // Round the difference into each candidate unit before deciding which bucket it belongs in.
+        // Rounding first keeps the chosen bucket and the displayed number consistent, so a value near a
+        // boundary (for example 59.5 seconds) is promoted to the next unit ("1 min ago") instead of
+        // producing a rounded number that overflows its own bucket ("60 secs ago").
+        final long seconds = Math.round((double) absoluteDifferenceMillis / secondMillis);
+        if (seconds < 60L) {
+            return formatRelativeUnit(Math.max(1L, seconds), "sec", past);
+        }
+
+        final long minutes = Math.round((double) absoluteDifferenceMillis / minuteMillis);
+        if (minutes < 60L) {
+            return formatRelativeUnit(minutes, "min", past);
+        }
+
+        final long hours = Math.round((double) absoluteDifferenceMillis / hourMillis);
+        if (hours < 24L) {
+            return formatRelativeUnit(hours, "hour", past);
+        }
+
+        final long days = Math.round((double) absoluteDifferenceMillis / dayMillis);
+        if (days == 1L) {
+            return past ? "yesterday" : "tomorrow";
+        }
+        if (days < 7L) {
+            return formatRelativeUnit(days, "day", past);
+        }
+        if (absoluteDifferenceMillis < monthMillis) {
+            final long weeks = Math.round((double) absoluteDifferenceMillis / weekMillis);
+            return formatRelativeUnit(weeks, "week", past);
+        }
+        if (absoluteDifferenceMillis < yearMillis) {
+            final long months = Math.round((double) absoluteDifferenceMillis / monthMillis);
+            return formatRelativeUnit(months, "month", past);
+        }
+
+        final long years = Math.round((double) absoluteDifferenceMillis / yearMillis);
+        return formatRelativeUnit(years, "year", past);
+    }
+
+    private static String formatRelativeUnit(final long value, final String unit, final boolean past) {
+        final String suffix = (value == 1L) ? unit : unit + "s";
+        return past ? value + " " + suffix + " ago" : "in " + value + " " + suffix;
+    }
+
     /**
      * Returns a time duration in the requested {@link TimeUnit} after parsing the {@code String}
      * input. If the resulting value is a decimal (i.e.

nifi-commons/nifi-utils/src/test/java/org/apache/nifi/util/TestFormatUtils.java

@@ -177,6 +177,61 @@ public void testFormatTime(long sourceDuration, TimeUnit sourceUnit, String expe
         assertEquals(expected, FormatUtils.formatHoursMinutesSeconds(sourceDuration, sourceUnit));
     }
 
+    @ParameterizedTest
+    @MethodSource("getRelativeTimeArguments")
+    public void testFormatRelativeTime(final long differenceMillis, final String expected) {
+        final Instant reference = Instant.parse("2026-05-15T10:00:00Z");
+        final Instant from = reference.minusMillis(differenceMillis);
+        assertEquals(expected, FormatUtils.formatRelativeTime(from, reference));
+    }
+
+    private static Stream<Arguments> getRelativeTimeArguments() {
+        final long second = 1_000L;
+        final long minute = 60L * second;
+        final long hour = 60L * minute;
+        final long day = 24L * hour;
+        final long week = 7L * day;
+        final long month = 30L * day;
+        final long year = 365L * day;
+        return Stream.of(
+            Arguments.of(0L, "1 sec ago"),
+            Arguments.of(500L, "1 sec ago"),
+            Arguments.of(second, "1 sec ago"),
+            Arguments.of(45L * second, "45 secs ago"),
+            Arguments.of(minute, "1 min ago"),
+            Arguments.of(2L * minute, "2 mins ago"),
+            Arguments.of(59L * minute, "59 mins ago"),
+            Arguments.of(hour, "1 hour ago"),
+            Arguments.of(5L * hour, "5 hours ago"),
+            Arguments.of(day, "yesterday"),
+            Arguments.of(2L * day, "2 days ago"),
+            Arguments.of(6L * day, "6 days ago"),
+            Arguments.of(week, "1 week ago"),
+            Arguments.of(3L * week, "3 weeks ago"),
+            Arguments.of(month, "1 month ago"),
+            Arguments.of(6L * month, "6 months ago"),
+            Arguments.of(year, "1 year ago"),
+            Arguments.of(3L * year, "3 years ago"),
+            // Half-unit boundary values: a difference that rounds up to a full next unit must be
+            // promoted to that unit rather than rendering an overflowed count in the smaller unit.
+            Arguments.of(minute - second / 2 - 1L, "59 secs ago"),
+            Arguments.of(minute - second / 2, "1 min ago"),
+            Arguments.of(hour - minute / 2 - second, "59 mins ago"),
+            Arguments.of(hour - minute / 2, "1 hour ago"),
+            Arguments.of(day - hour / 2 - minute, "23 hours ago"),
+            Arguments.of(day - hour / 2, "yesterday"),
+            Arguments.of(week - day / 2 - hour, "6 days ago"),
+            Arguments.of(week - day / 2, "1 week ago"),
+            Arguments.of(month - week / 2, "4 weeks ago"),
+            // Negative offset (from is after to): future-tense rendering
+            Arguments.of(-1L * second, "in 1 sec"),
+            Arguments.of(-2L * minute, "in 2 mins"),
+            Arguments.of(-1L * day, "tomorrow"),
+            Arguments.of(-3L * day, "in 3 days"),
+            Arguments.of(-1L * year, "in 1 year")
+        );
+    }
+
     private static Stream<Arguments> getFormatTime() {
         return Stream.of(Arguments.of(0L, TimeUnit.DAYS, "00:00:00.000"),
             Arguments.of(1L, TimeUnit.HOURS, "01:00:00.000"),

nifi-connectors/nifi-kafka-to-s3-bundle/nifi-kafka-to-s3-connector/src/main/java/org/apache/nifi/connectors/kafkas3/KafkaToS3.java

@@ -20,10 +20,13 @@
 import org.apache.nifi.annotation.documentation.CapabilityDescription;
 import org.apache.nifi.annotation.documentation.Tags;
 import org.apache.nifi.components.AllowableValue;
+import org.apache.nifi.components.Backlog;
+import org.apache.nifi.components.BacklogReportingException;
 import org.apache.nifi.components.ConfigVerificationResult;
 import org.apache.nifi.components.ConfigVerificationResult.Outcome;
 import org.apache.nifi.components.DescribedValue;
 import org.apache.nifi.components.connector.AbstractConnector;
+import org.apache.nifi.components.connector.BacklogReportingConnector;
 import org.apache.nifi.components.connector.ConfigurationStep;
 import org.apache.nifi.components.connector.ConnectorConfigurationContext;
 import org.apache.nifi.components.connector.FlowUpdateException;
@@ -41,6 +44,7 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
@@ -51,7 +55,7 @@
 @CapabilityDescription("Provides the ability to ingest data from Apache Kafka topics, merge it together into an object of reasonable " +
                        "size, and write that data to Amazon S3.")
 @Tags({"kafka", "s3"})
-public class KafkaToS3 extends AbstractConnector {
+public class KafkaToS3 extends AbstractConnector implements BacklogReportingConnector {
 
     private static final List<ConfigurationStep> configurationSteps = List.of(
         KafkaConnectionStep.KAFKA_CONNECTION_STEP,
@@ -146,6 +150,20 @@ public List<ConfigVerificationResult> verifyConfigurationStep(final String stepN
         return Collections.emptyList();
     }
 
+    @Override
+    public Optional<Backlog> getBacklog(final FlowContext activeFlowContext) throws BacklogReportingException {
+        final Optional<ProcessorFacade> consumeKafkaProcessor = findConsumeKafkaProcessor(activeFlowContext);
+        if (consumeKafkaProcessor.isEmpty()) {
+            return Optional.empty();
+        }
+
+        return consumeKafkaProcessor.get().getBacklog();
+    }
+
+    private Optional<ProcessorFacade> findConsumeKafkaProcessor(final FlowContext flowContext) {
+        return findProcessors(flowContext.getRootGroup(), processor -> processor.getDefinition().getType().endsWith("ConsumeKafka")).stream().findFirst();
+    }
+
     private List<ConfigVerificationResult> verifyKafkaParsability(final FlowContext workingFlowContext, final VersionedExternalFlow flow) {
         // Enable Controller Services necessary for parsing records.
         // We determine which Controller Services are referenced by the flow and enable them, but we do not use
@@ -170,8 +188,7 @@ private List<ConfigVerificationResult> verifyKafkaParsability(final FlowContext
         }
 
         try {
-            final ProcessorFacade consumeKafkaFacade = findProcessors(rootGroup,
-                processor -> processor.getDefinition().getType().endsWith("ConsumeKafka")).getFirst();
+            final ProcessorFacade consumeKafkaFacade = findConsumeKafkaProcessor(workingFlowContext).orElseThrow();
 
             final List<ConfigVerificationResult> configVerificationResults = consumeKafkaFacade.verify(flow, Map.of());
             for (final ConfigVerificationResult result : configVerificationResults) {

nifi-docs/src/main/asciidoc/user-guide.adoc

@@ -306,6 +306,7 @@ NOTE: For Processors, Ports, Remote Process Groups, Connections and Labels, it i
 - *View data provenance*: This option displays the NiFi Data Provenance table, with information about data provenance events for the FlowFiles routed through that Processor (see <<data_provenance>>).
 - *Replay last event*: This option will replay the last Provenance event, effectively requeuing the last FlowFile that was processed by the Processor (see <<replay_flowfile>>).
 - *View status history*: This option opens a graphical representation of the Processor's statistical information over time.
+- *View Backlog*: For Processors that support reporting backlog against an external system (such as a queue or topic), this option displays the current backlog information. Because computing backlog may require querying the external system, this action requires *modify the component* access, not merely *view the component* access.
 - *View usage*: This option takes the user to the Processor's usage documentation.
 - *View connections->Upstream*: This option allows the user to see and "jump to" upstream connections that are coming into the Processor. This is particularly useful when processors connect into and out of other Process Groups.
 - *View connections->Downstream*: This option allows the user to see and "jump to" downstream connections that are going out of the Processor. This is particularly useful when processors connect into and out of other Process Groups.