Hello, I see we have 2 times the same condition check in the function getNumRecordsInPerSecond (L220 <https://github.com/apache/flink-kubernetes-operator/blob/main/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java#L220> and L224 <https://github.com/apache/flink-kubernetes-operator/blob/main/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/metrics/ScalingMetrics.java#L224>). I imagine you want to use SOURCE_TASK_NUM_RECORDS_OUT_PER_SEC when the operator is not the source. Can you confirm this and if we have a FIP ticket to fix this?
Regards, Yang LI
