Hello Guys, When we use the autoscaler function of flink-k8s-operator, a task flink built-in test task is started on the cluster. The configuration is as follows. When using the autoscaler, it is found that only the sink end will scale-down (the parallelism of the task is 4, and the sink end scale-down becomes 1 after two times), but the source end remains unchanged. It makes us feel very distressed, and I would like to ask whether the scaling of the source task is not supported now or there is a problem with my use here.

apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flink-k8s-operator-autoas-test
  namespace: dc-flink-test
spec:
  image: flink:1.17
  flinkVersion: v1_17
  flinkConfiguration:
    taskmanager.numberOfTaskSlots: "2"
	kubernetes.operator.job.autoscaler.enabled: "true"
    kubernetes.operator.job.autoscaler.stabilization.interval: 1m
    kubernetes.operator.job.autoscaler.metrics.window: 5m
    kubernetes.operator.job.autoscaler.scale-down.max-factor: "0.5"
    kubernetes.operator.job.autoscaler.restart.time: 2m
    kubernetes.operator.job.autoscaler.catch-up.duration: 5m
	kubernetes.operator.job.autoscaler.metrics.busy-time.aggregator: "AVG"
    pipeline.max-parallelism: "10"
  serviceAccount: flink
  jobManager:
    resource:
      memory: "2048m"
      cpu: 1
  taskManager:
    resource:
      memory: "2048m"
      cpu: 1
  job:
    jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar
    parallelism: 4
    upgradeMode: stateless

here is my task screenshot :

Im facing same problem, but mostly it goes down to the fact that busy-metric is not provided by source task (N/A -> busy). I think its same here. https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/custom-resource/autoscaler/ • Source scaling only supports modern sources which ◦ use the new Source API that exposes the busy time metric (must have, most common connectors already do) ◦ expose the standardized connector metrics for accessing backlog information (good to have, extra capacity will be added for catching up with backlog) Unsure what custom source is in your case, I bet it does not provide necessary metrics. In operator pod there should be error logs complaining about lack of busy metrics - you can check I guess.

I also noticed this. I checked the pod logs and didn't see any related error logs. I used the official test case: examples/simulation/StateMachineExample.jar I think the custom source in him did not implement the requirements in autoscaler. Are you also a custom source? I plan to use the new API to implement kafkasource and observe the situation with Autoscaler in the next step. As he said: In the current state of the autoscaler works best with Kafka sources, as they expose all the standardized metrics Then verify the source status of his autoscaler function

I mean flink kubernetes operator pod just in case, not job manager. flink-kubernetes-operator-5969bf79b6-96blk.k8soperator.flink.flink-kubernetes-operator.namespace.myns.FlinkDeployment.JOSDK.reconcile.failed.TimeSeconds: count=2, min=0, max=20, mean=10.0, stddev=14.142135623730951, p50=10.0, p75=20.0, p95=20.0, p98=20.0, p99=20.0, p999=20.0 flink-kubernetes-operator-5969bf79b6-96blk.k8soperator.flink.flink-kubernetes-operator.namespace.myns.FlinkDeployment.Lifecycle.Transition.Rollback.TimeSeconds: count=0, min=0, max=0, mean=NaN, stddev=NaN, p50=NaN, p75=NaN, p95=NaN, p98=NaN, p99=NaN, p999=NaN flink-kubernetes-operator-5969bf79b6-96blk.k8soperator.flink.flink-kubernetes-operator.system.FlinkDeployment.Lifecycle.Transition.Stabilization.TimeSeconds: count=7, min=59, max=109, mean=73.14285714285714, stddev=16.597475424780434, p50=68.0, p75=75.0, p95=109.0, p98=109.0, p99=109.0, p999=109.0 =========================== Finished metrics report =========================== 2023-08-08 084308,967 o.a.f.k.o.c.FlinkDeploymentController [INFO ][myns/test-flink-deployment] Starting reconciliation 2023-08-08 084308,968 o.a.f.k.o.s.FlinkResourceContextFactory [INFO ][myns/test-flink-deployment] Getting service for test-flink-deployment 2023-08-08 084308,968 o.a.f.k.o.o.JobStatusObserver [INFO ][myns/test-flink-deployment] Observing job status 2023-08-08 084308,973 o.a.f.k.o.o.JobStatusObserver [INFO ][myns/test-flink-deployment] Job status (RUNNING) unchanged 2023-08-08 084309,073 o.a.f.k.o.a.m.ScalingMetrics [ERROR][myns/test-flink-deployment] No busyTimeMsPerSecond metric available for 0c42f48b148d41f896bb904f7cd8fb9b. No scaling will be performed for this vertex. 2023-08-08 084309,073 o.a.f.k.o.a.m.ScalingMetrics [INFO ][myns/test-flink-deployment] Using computed source data rate 0.0 for 0c42f48b148d41f896bb904f7cd8fb9b 2023-08-08 084309,073 o.a.f.k.o.a.m.ScalingMetrics [ERROR][myns/test-flink-deployment] No busyTimeMsPerSecond metric available for a6c3ae73c49b4440adf80c1815e2d73e. No scaling will be performed for this vertex. Those kind of metrics. If idle is 0%, busy N/A I think it should be something like it.

The state machine example uses a simple generator source that doesn’t expose the busy time metrics properly there it cannot be scaled efficiently

Yes, when I used kafkasource to package the Flink program to jar and ran it on k8s operator, I found that both the source and sink sides could perform scaled up and down, which was really cool