[FLINK-39549][flink-autoscaler] Compute OBSERVED_TPR for non-source vertices behind a feature flag

Author: Dennis-Mircea

docs/layouts/shortcodes/generated/auto_scaler_configuration.html

@@ -128,6 +128,18 @@
             <td>Integer</td>
             <td>Minimum nr of observations used when estimating / switching to observed true processing rate.</td>
         </tr>
+        <tr>
+            <td><h5>job.autoscaler.observed-true-processing-rate.non-source.enabled</h5></td>
+            <td style="word-wrap: break-word;">false</td>
+            <td>Boolean</td>
+            <td>Whether to also compute the observed (backpressure-derived) true processing rate for non-source vertices. When enabled, OBSERVED_TPR is populated for any vertex whose engagement (busy + backpressured time) crosses the configured threshold, mirroring the source-side behavior gated on lag. Default off for backward compatibility.</td>
+        </tr>
+        <tr>
+            <td><h5>job.autoscaler.observed-true-processing-rate.non-source.engagement-threshold</h5></td>
+            <td style="word-wrap: break-word;">0.95</td>
+            <td>Double</td>
+            <td>Engagement threshold (between 0 and 1) for triggering observed true processing rate computation on non-source vertices. Computed as (busyTimeMsPerSecond + backPressuredTimeMsPerSecond) / 1000. The vertex is considered fully engaged (and therefore not input-starved) when this value crosses the threshold, which is the non-source analogue of the source 'has lag' condition.</td>
+        </tr>
         <tr>
             <td><h5>job.autoscaler.observed-true-processing-rate.switch-threshold</h5></td>
             <td style="word-wrap: break-word;">0.15</td>

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricCollector.java

@@ -439,6 +439,7 @@ protected Map<JobVertexID, Map<String, FlinkMetric>> queryFilteredMetricNames(
     @SneakyThrows
     private Map<JobVertexID, Map<String, FlinkMetric>> queryFilteredMetricNames(
             Context ctx, JobTopology topology, Stream<JobVertexID> vertexStream) {
+        var conf = ctx.getConfiguration();
         try (var restClient = ctx.getRestClusterClient()) {
             return vertexStream
                     .filter(v -> !topology.getFinishedVertices().contains(v))
@@ -447,7 +448,11 @@ private Map<JobVertexID, Map<String, FlinkMetric>> queryFilteredMetricNames(
                                     v -> v,
                                     v ->
                                             getFilteredVertexMetricNames(
-                                                    restClient, ctx.getJobID(), v, topology)));
+                                                    restClient,
+                                                    ctx.getJobID(),
+                                                    v,
+                                                    topology,
+                                                    conf)));
         }
     }
 
@@ -456,7 +461,8 @@ Map<String, FlinkMetric> getFilteredVertexMetricNames(
             RestClusterClient<?> restClient,
             JobID jobID,
             JobVertexID jobVertexID,
-            JobTopology topology) {
+            JobTopology topology,
+            Configuration conf) {
 
         var allMetricNames = queryAggregatedMetricNames(restClient, jobID, jobVertexID);
         var filteredMetrics = new HashMap<String, FlinkMetric>();
@@ -484,6 +490,17 @@ Map<String, FlinkMetric> getFilteredVertexMetricNames(
                     .findAny(allMetricNames)
                     .ifPresent(
                             m -> filteredMetrics.put(m, FlinkMetric.SOURCE_TASK_NUM_RECORDS_OUT));
+        } else if (conf != null
+                && conf.get(AutoScalerOptions.OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENABLED)) {
+            // Non-source observed TPR computation requires backpressure and a per-vertex input
+            // rate. These are best-effort: if either is unavailable for the deployed Flink
+            // version, we silently skip observed-TPR for this vertex (no exception).
+            FlinkMetric.BACKPRESSURE_TIME_PER_SEC
+                    .findAny(allMetricNames)
+                    .ifPresent(m -> filteredMetrics.put(m, FlinkMetric.BACKPRESSURE_TIME_PER_SEC));
+            FlinkMetric.NUM_RECORDS_IN_PER_SEC
+                    .findAny(allMetricNames)
+                    .ifPresent(m -> filteredMetrics.put(m, FlinkMetric.NUM_RECORDS_IN_PER_SEC));
         }
 
         for (FlinkMetric flinkMetric : requiredMetrics) {

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/config/AutoScalerOptions.java

@@ -243,6 +243,35 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
                     .withDescription(
                             "Minimum nr of observations used when estimating / switching to observed true processing rate.");
 
+    public static final ConfigOption<Boolean> OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENABLED =
+            autoScalerConfig("observed-true-processing-rate.non-source.enabled")
+                    .booleanType()
+                    .defaultValue(false)
+                    .withFallbackKeys(
+                            oldOperatorConfigKey(
+                                    "observed-true-processing-rate.non-source.enabled"))
+                    .withDescription(
+                            "Whether to also compute the observed (backpressure-derived) true processing rate "
+                                    + "for non-source vertices. When enabled, OBSERVED_TPR is populated for any vertex "
+                                    + "whose engagement (busy + backpressured time) crosses the configured threshold, "
+                                    + "mirroring the source-side behavior gated on lag. Default off for backward compatibility.");
+
+    public static final ConfigOption<Double>
+            OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENGAGEMENT_THRESHOLD =
+                    autoScalerConfig(
+                                    "observed-true-processing-rate.non-source.engagement-threshold")
+                            .doubleType()
+                            .defaultValue(0.95)
+                            .withFallbackKeys(
+                                    oldOperatorConfigKey(
+                                            "observed-true-processing-rate.non-source.engagement-threshold"))
+                            .withDescription(
+                                    "Engagement threshold (between 0 and 1) for triggering observed true processing rate "
+                                            + "computation on non-source vertices. Computed as (busyTimeMsPerSecond + "
+                                            + "backPressuredTimeMsPerSecond) / 1000. The vertex is considered fully engaged "
+                                            + "(and therefore not input-starved) when this value crosses the threshold, "
+                                            + "which is the non-source analogue of the source 'has lag' condition.");
+
     public static final ConfigOption<Boolean> SCALING_EFFECTIVENESS_DETECTION_ENABLED =
             autoScalerConfig("scaling.effectiveness.detection.enabled")
                     .booleanType()

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/FlinkMetric.java

@@ -32,6 +32,7 @@
  */
 public enum FlinkMetric {
     BUSY_TIME_PER_SEC(s -> s.equals("busyTimeMsPerSecond")),
+    NUM_RECORDS_IN_PER_SEC(s -> s.equals("numRecordsInPerSecond")),
     SOURCE_TASK_NUM_RECORDS_IN_PER_SEC(
             s -> s.startsWith("Source__") && s.endsWith(".numRecordsInPerSecond")),
     SOURCE_TASK_NUM_RECORDS_OUT(s -> s.startsWith("Source__") && s.endsWith(".numRecordsOut")),

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetric.java

@@ -33,7 +33,13 @@ public enum ScalingMetric {
     /** Processing rate at full capacity (records/sec). */
     TRUE_PROCESSING_RATE(true),
 
-    /** Observed true processing rate for sources. */
+    /**
+     * Observed (backpressure-derived) true processing rate. Always populated for source vertices
+     * when the source is "catching up" (lag-based gate). Optionally populated for non-source
+     * vertices when {@code job.autoscaler.observed-true-processing-rate.non-source.enabled} is set
+     * — gated on the vertex being fully engaged (busy + backpressured time crosses the configured
+     * engagement threshold), which is the per-vertex analogue of the source "has lag" condition.
+     */
     OBSERVED_TPR(true),
 
     /** Target processing rate of operators as derived from source inputs (records/sec). */

flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java

@@ -98,6 +98,12 @@ public static void computeDataRateMetrics(
                                 .orElseGet(observedTprAvg);
                 scalingMetrics.put(ScalingMetric.OBSERVED_TPR, observedTprOpt);
             }
+        } else if (conf.get(AutoScalerOptions.OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENABLED)) {
+            var nonSourceObservedTpr =
+                    getNonSourceObservedTpr(flinkMetrics, conf).orElseGet(observedTprAvg);
+            if (!Double.isNaN(nonSourceObservedTpr)) {
+                scalingMetrics.put(ScalingMetric.OBSERVED_TPR, nonSourceObservedTpr);
+            }
         }
     }
 
@@ -131,6 +137,58 @@ private static Optional<Double> getObservedTpr(
         return Double.isNaN(observedTpr) ? Optional.empty() : Optional.of(observedTpr);
     }
 
+    /**
+     * Compute the observed (backpressure-derived) true processing rate for a non-source vertex.
+     *
+     * <p>Triggered only when the vertex is "fully engaged" — i.e. spending nearly all its time
+     * either busy or backpressured. This is the per-vertex analogue of the source-side "has lag"
+     * gate in {@link #getObservedTpr}: a fully engaged vertex is not input-starved by upstream, so
+     * its observed input rate represents its true capacity (modulo backpressure).
+     *
+     * <p>Returns {@link Optional#empty()} when the required metrics are unavailable, the vertex is
+     * not engaged enough, or the formula degenerates (e.g. {@code backpressure >= 1000ms/s}). In
+     * those cases the caller falls back to the historical observed TPR average (if any).
+     */
+    private static Optional<Double> getNonSourceObservedTpr(
+            Map<FlinkMetric, AggregatedMetric> flinkMetrics, Configuration conf) {
+
+        var bpMetric = flinkMetrics.get(FlinkMetric.BACKPRESSURE_TIME_PER_SEC);
+        var rateMetric = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+        var busyMetric = flinkMetrics.get(FlinkMetric.BUSY_TIME_PER_SEC);
+        if (bpMetric == null || rateMetric == null || busyMetric == null) {
+            return Optional.empty();
+        }
+
+        double busyMsPerSec = busyMetric.getAvg();
+        double bpMsPerSec = bpMetric.getAvg();
+        if (!Double.isFinite(busyMsPerSec) || !Double.isFinite(bpMsPerSec)) {
+            return Optional.empty();
+        }
+        // Engagement = fraction of time the vertex is either doing work or blocked downstream.
+        // High engagement means the vertex is not waiting for input -> rate reflects capacity.
+        double engagement = (Math.max(0, busyMsPerSec) + Math.max(0, bpMsPerSec)) / 1000.;
+        double engagementThreshold =
+                conf.get(
+                        AutoScalerOptions
+                                .OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENGAGEMENT_THRESHOLD);
+        if (engagement < engagementThreshold) {
+            return Optional.empty();
+        }
+
+        double numRecordsInPerSecond = rateMetric.getSum();
+        if (Double.isNaN(numRecordsInPerSecond)) {
+            return Optional.empty();
+        }
+        // Mirror source-side semantics: zero input rate while engaged signals no work to do,
+        // allow scale down.
+        if (numRecordsInPerSecond == 0) {
+            return Optional.of(Double.POSITIVE_INFINITY);
+        }
+
+        double observedTpr = computeObservedTprWithBackpressure(numRecordsInPerSecond, bpMsPerSec);
+        return Double.isNaN(observedTpr) ? Optional.empty() : Optional.of(observedTpr);
+    }
+
     public static double computeObservedTprWithBackpressure(
             double numRecordsInPerSecond, double backpressureMsPerSeconds) {
         if (backpressureMsPerSeconds >= 1000) {

flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricCollectorTest.java

@@ -392,7 +392,11 @@ public void testQueryNamesOnTopologyChange() {
                 new RestApiMetricsCollector<JobID, JobAutoScalerContext<JobID>>() {
                     @Override
                     protected Map<String, FlinkMetric> getFilteredVertexMetricNames(
-                            RestClusterClient<?> rc, JobID id, JobVertexID jvi, JobTopology t) {
+                            RestClusterClient<?> rc,
+                            JobID id,
+                            JobVertexID jvi,
+                            JobTopology t,
+                            Configuration conf) {
                         metricNameQueryCounter.compute(jvi, (j, c) -> c + 1);
                         return Map.of();
                     }
@@ -517,7 +521,8 @@ private void testRequiredMetrics(
             metricList.addAll(requiredMetrics);
             metricList.remove(m);
             try {
-                testCollector.getFilteredVertexMetricNames(null, new JobID(), vertex, topology);
+                testCollector.getFilteredVertexMetricNames(
+                        null, new JobID(), vertex, topology, new Configuration());
                 fail(m);
             } catch (Exception e) {
                 assertTrue(e.getMessage().startsWith("Could not find required metric "));
@@ -551,6 +556,6 @@ public void testThrowsMetricNotFoundException() {
                 MetricNotFoundException.class,
                 () ->
                         metricCollector.getFilteredVertexMetricNames(
-                                null, new JobID(), source, topology));
+                                null, new JobID(), source, topology, new Configuration()));
     }
 }

flink-autoscaler/src/test/java/org/apache/flink/autoscaler/metrics/ScalingMetricsTest.java

@@ -291,6 +291,107 @@ private static AggregatedMetric aggSum(double sum) {
         return new AggregatedMetric("", Double.NaN, Double.NaN, Double.NaN, sum, Double.NaN);
     }
 
+    private static AggregatedMetric aggAvg(double avg) {
+        return new AggregatedMetric("", Double.NaN, Double.NaN, avg, Double.NaN, Double.NaN);
+    }
+
+    private static double computeNonSourceObservedTpr(
+            double busyAvg,
+            double bpAvg,
+            double inputRatePerSecSum,
+            Configuration conf,
+            double prevTpr) {
+        var source = new JobVertexID();
+        var sink = new JobVertexID();
+        var topology =
+                new JobTopology(
+                        new VertexInfo(
+                                source, Collections.emptyMap(), 1, 1, new IOMetrics(0, 0, 0)),
+                        new VertexInfo(
+                                sink, Map.of(source, REBALANCE), 1, 1, new IOMetrics(0, 0, 0)));
+
+        Map<ScalingMetric, Double> scalingMetrics = new HashMap<>();
+        var flinkMetrics = new HashMap<FlinkMetric, AggregatedMetric>();
+        flinkMetrics.put(FlinkMetric.BUSY_TIME_PER_SEC, aggAvg(busyAvg));
+        if (!Double.isNaN(bpAvg)) {
+            flinkMetrics.put(FlinkMetric.BACKPRESSURE_TIME_PER_SEC, aggAvg(bpAvg));
+        }
+        if (!Double.isNaN(inputRatePerSecSum)) {
+            flinkMetrics.put(FlinkMetric.NUM_RECORDS_IN_PER_SEC, aggSum(inputRatePerSecSum));
+        }
+        ScalingMetrics.computeDataRateMetrics(
+                sink, flinkMetrics, scalingMetrics, topology, conf, () -> prevTpr);
+        return scalingMetrics.getOrDefault(ScalingMetric.OBSERVED_TPR, Double.NaN);
+    }
+
+    @Test
+    public void testNonSourceObservedTprDisabledByDefault() {
+        // Even with all required metrics present and the vertex fully engaged,
+        // OBSERVED_TPR must NOT be populated for non-sources unless the flag is on.
+        var conf = new Configuration();
+        // Flag intentionally NOT set -> default false.
+        double tpr = computeNonSourceObservedTpr(900., 100., 1000., conf, Double.NaN);
+        assertTrue(Double.isNaN(tpr));
+    }
+
+    @Test
+    public void testNonSourceObservedTprEnabled() {
+        var conf = new Configuration();
+        conf.set(AutoScalerOptions.OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENABLED, true);
+
+        // Fully engaged (busy 900 + bp 100 = 1000ms/s) and 200ms/s backpressure.
+        // Wait — set bp=200 so it's clearly above zero and the formula is exercised.
+        // engagement = (busy + bp)/1000 = (700+200)/1000 = 0.9 -> below default 0.95 -> NaN
+        assertTrue(Double.isNaN(computeNonSourceObservedTpr(700., 200., 1000., conf, Double.NaN)));
+
+        // engagement = (800+200)/1000 = 1.0 >= 0.95 -> compute
+        // observedTpr = rate / (1 - bp/1000) = 1000 / 0.8 = 1250
+        assertEquals(1000. / 0.8, computeNonSourceObservedTpr(800., 200., 1000., conf, Double.NaN));
+
+        // Zero input rate while engaged -> POSITIVE_INFINITY (mirrors source behavior).
+        assertEquals(
+                Double.POSITIVE_INFINITY,
+                computeNonSourceObservedTpr(950., 50., 0., conf, Double.NaN));
+
+        // Backpressure >= 1000ms/s -> formula degenerates -> fallback supplier (null here)
+        // is returned only if non-NaN, otherwise OBSERVED_TPR is not set.
+        assertTrue(Double.isNaN(computeNonSourceObservedTpr(0., 1000., 1000., conf, Double.NaN)));
+
+        // Below engagement threshold -> fallback to historical observedTprAvg (PREV_TPR).
+        assertEquals(PREV_TPR, computeNonSourceObservedTpr(100., 100., 1000., conf, PREV_TPR));
+    }
+
+    @Test
+    public void testNonSourceObservedTprCustomEngagementThreshold() {
+        var conf = new Configuration();
+        conf.set(AutoScalerOptions.OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENABLED, true);
+        conf.set(
+                AutoScalerOptions.OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENGAGEMENT_THRESHOLD,
+                0.5);
+
+        // engagement = 0.6 >= 0.5 -> compute
+        assertEquals(
+                500. / (1 - 0.1), computeNonSourceObservedTpr(500., 100., 500., conf, Double.NaN));
+
+        // engagement = 0.4 < 0.5 -> NaN
+        assertTrue(Double.isNaN(computeNonSourceObservedTpr(300., 100., 500., conf, Double.NaN)));
+    }
+
+    @Test
+    public void testNonSourceObservedTprMissingMetricsAreSilentlySkipped() {
+        var conf = new Configuration();
+        conf.set(AutoScalerOptions.OBSERVED_TRUE_PROCESSING_RATE_NON_SOURCE_ENABLED, true);
+
+        // No backpressure metric available (e.g. older Flink / metric not requested) -> NaN
+        assertTrue(
+                Double.isNaN(
+                        computeNonSourceObservedTpr(900., Double.NaN, 1000., conf, Double.NaN)));
+        // No input rate metric available -> NaN
+        assertTrue(
+                Double.isNaN(
+                        computeNonSourceObservedTpr(900., 100., Double.NaN, conf, Double.NaN)));
+    }
+
     private static AggregatedMetric aggMax(double max) {
         return new AggregatedMetric("", Double.NaN, max, Double.NaN, Double.NaN, Double.NaN);
     }