You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2023/10/30 09:04:04 UTC
(flink-kubernetes-operator) branch main updated: [FLINK-33306] Always use busy tpr when observed is nan
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 582c07ae [FLINK-33306] Always use busy tpr when observed is nan
582c07ae is described below
commit 582c07aeec6b4bbcc68e0a85d31000d89dbd1958
Author: Gyula Fora <g_...@apple.com>
AuthorDate: Sat Oct 28 10:35:42 2023 +0200
[FLINK-33306] Always use busy tpr when observed is nan
---
.../flink/autoscaler/ScalingMetricEvaluator.java | 8 +++----
.../autoscaler/ScalingMetricEvaluatorTest.java | 26 ++++++++++++++++++++++
2 files changed, 30 insertions(+), 4 deletions(-)
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
index a97ac6f1..61aa6f2f 100644
--- a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/ScalingMetricEvaluator.java
@@ -179,14 +179,14 @@ public class ScalingMetricEvaluator {
double busyTimeTprAvg,
double observedTprAvg) {
- if (Double.isInfinite(busyTimeTprAvg) || Double.isNaN(busyTimeTprAvg)) {
- return OBSERVED_TPR;
- }
-
if (Double.isNaN(observedTprAvg)) {
return TRUE_PROCESSING_RATE;
}
+ if (Double.isInfinite(busyTimeTprAvg) || Double.isNaN(busyTimeTprAvg)) {
+ return OBSERVED_TPR;
+ }
+
double switchThreshold =
conf.get(AutoScalerOptions.OBSERVED_TRUE_PROCESSING_RATE_SWITCH_THRESHOLD);
// If we could measure the observed tpr we decide whether to switch to using it
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
index ce7c5cb5..839bda2e 100644
--- a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/ScalingMetricEvaluatorTest.java
@@ -433,6 +433,32 @@ public class ScalingMetricEvaluatorTest {
.evaluate(conf, new CollectedMetricHistory(topology, metricHistory))
.get(source)
.get(ScalingMetric.TRUE_PROCESSING_RATE));
+
+ metricHistory.put(
+ Instant.ofEpochMilli(100),
+ new CollectedMetrics(
+ Map.of(
+ source,
+ Map.of(
+ ScalingMetric.LAG,
+ 0.,
+ ScalingMetric.TRUE_PROCESSING_RATE,
+ Double.POSITIVE_INFINITY,
+ ScalingMetric.CURRENT_PROCESSING_RATE,
+ 100.,
+ ScalingMetric.SOURCE_DATA_RATE,
+ 50.,
+ ScalingMetric.LOAD,
+ 10.)),
+ Map.of()));
+
+ // Test that we used busy time based TPR even when infinity
+ assertEquals(
+ new EvaluatedScalingMetric(Double.POSITIVE_INFINITY, Double.POSITIVE_INFINITY),
+ evaluator
+ .evaluate(conf, new CollectedMetricHistory(topology, metricHistory))
+ .get(source)
+ .get(ScalingMetric.TRUE_PROCESSING_RATE));
}
private Tuple2<Double, Double> getThresholds(