[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

@@ -136,7 +136,7 @@
         </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 style="word-wrap: break-word;">0.9</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>

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

@@ -492,9 +492,6 @@ Map<String, FlinkMetric> getFilteredVertexMetricNames(
                             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));

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

@@ -261,7 +261,7 @@ private static ConfigOptions.OptionBuilder autoScalerConfig(String key) {
                     autoScalerConfig(
                                     "observed-true-processing-rate.non-source.engagement-threshold")
                             .doubleType()
-                            .defaultValue(0.95)
+                            .defaultValue(0.9)
                             .withFallbackKeys(
                                     oldOperatorConfigKey(
                                             "observed-true-processing-rate.non-source.engagement-threshold"))

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

@@ -32,7 +32,6 @@
  */
 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")),
@@ -41,6 +40,7 @@ public enum FlinkMetric {
     SOURCE_TASK_NUM_RECORDS_IN(s -> s.startsWith("Source__") && s.endsWith(".numRecordsIn")),
     PENDING_RECORDS(s -> s.endsWith(".pendingRecords")),
     BACKPRESSURE_TIME_PER_SEC(s -> s.equals("backPressuredTimeMsPerSecond")),
+    NUM_RECORDS_IN_PER_SEC(s -> s.equals("numRecordsInPerSecond")),
 
     HEAP_MEMORY_MAX(s -> s.equals("Status.JVM.Memory.Heap.Max")),
     HEAP_MEMORY_USED(s -> s.equals("Status.JVM.Memory.Heap.Used")),

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

@@ -34,11 +34,13 @@ public enum ScalingMetric {
     TRUE_PROCESSING_RATE(true),
 
     /**
-     * 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 true processing rate, derived from backpressure. Always populated for source
+     * vertices when the source is catching up, based on source lag. Optionally populated for
+     * non-source vertices when {@code
+     * job.autoscaler.observed-true-processing-rate.non-source.enabled} is set, in which case it is
+     * gated on the vertex being fully engaged (busy plus backpressured time crosses the configured
+     * engagement threshold). That gate is the per vertex analogue of the source "has lag"
+     * condition.
      */
     OBSERVED_TPR(true),
 

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

@@ -140,10 +140,10 @@ private static Optional<Double> getObservedTpr(
     /**
      * 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>Triggered only when the vertex is "fully engaged", meaning it spends nearly all of its
+     * time either busy or backpressured. This mirrors 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
@@ -152,20 +152,21 @@ private static Optional<Double> getObservedTpr(
     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) {
+        var backpressureMsPerSecond = flinkMetrics.get(FlinkMetric.BACKPRESSURE_TIME_PER_SEC);
+        var numRecordsInPerSecond = flinkMetrics.get(FlinkMetric.NUM_RECORDS_IN_PER_SEC);
+        var busyTimePerSecond = flinkMetrics.get(FlinkMetric.BUSY_TIME_PER_SEC);
+        if (backpressureMsPerSecond == null
+                || numRecordsInPerSecond == null
+                || busyTimePerSecond == null) {
             return Optional.empty();
         }
 
-        double busyMsPerSec = busyMetric.getAvg();
-        double bpMsPerSec = bpMetric.getAvg();
+        double busyMsPerSec = busyTimePerSecond.getAvg();
+        double bpMsPerSec = backpressureMsPerSecond.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(
@@ -175,17 +176,18 @@ private static Optional<Double> getNonSourceObservedTpr(
             return Optional.empty();
         }
 
-        double numRecordsInPerSecond = rateMetric.getSum();
-        if (Double.isNaN(numRecordsInPerSecond)) {
+        double numRecordsInPerSecondSum = numRecordsInPerSecond.getSum();
+        if (Double.isNaN(numRecordsInPerSecondSum)) {
             return Optional.empty();
         }
         // Mirror source-side semantics: zero input rate while engaged signals no work to do,
         // allow scale down.
-        if (numRecordsInPerSecond == 0) {
+        if (numRecordsInPerSecondSum == 0) {
             return Optional.of(Double.POSITIVE_INFINITY);
         }
 
-        double observedTpr = computeObservedTprWithBackpressure(numRecordsInPerSecond, bpMsPerSec);
+        double observedTpr =
+                computeObservedTprWithBackpressure(numRecordsInPerSecondSum, bpMsPerSec);
         return Double.isNaN(observedTpr) ? Optional.empty() : Optional.of(observedTpr);
     }
 

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

@@ -287,50 +287,11 @@ public void testGlobalMetrics() {
                         conf));
     }
 
-    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);
+        double tpr =
+                computeNonSourceObservedTpr(900., 100., 1000., new Configuration(), Double.NaN);
+
         assertTrue(Double.isNaN(tpr));
     }
 
@@ -339,25 +300,12 @@ 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));
     }
 
@@ -369,11 +317,8 @@ public void testNonSourceObservedTprCustomEngagementThreshold() {
                 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)));
     }
 
@@ -382,21 +327,56 @@ 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 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 AggregatedMetric aggMax(double max) {
         return new AggregatedMetric("", Double.NaN, max, Double.NaN, Double.NaN, Double.NaN);
     }
 
     private static AggregatedMetric aggAvgMax(double avg, double max) {
         return new AggregatedMetric("", Double.NaN, max, 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);
+    }
 }