[FLINK-34655] Copy IOMetricsInfo to flink-autoscaler-standalone module

[1996fanrui]

flink-ubernetes-operator is committed to supporting the latest 4 flink minor versions, and autoscaler is a part of flink-ubernetes-operator. Currently, the latest 4 flink minor versions are 1.15, 1.16, 1.17 and 1.18.

But autoscaler doesn't work for flink 1.15.

Root cause:

• FLINK-28310 added some properties in IOMetricsInfo in flink-1.16
• IOMetricsInfo is a part of JobDetailsInfo
• JobDetailsInfo is necessary for autoscaler [1]
• flink's RestClient doesn't allow miss any property during deserializing the json

That means that the RestClient after 1.15 cannot fetch JobDetailsInfo for 1.15 jobs.

How to fix it properly?

• [FLINK-34655] Copy IOMetricsInfo to flink-autoscaler-standalone module
• Removing them after 1.15 are not supported

[1]

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

Line 109 in ede1a61

getJobDetailsInfo(ctx, conf.get(AutoScalerOptions.FLINK_CLIENT_TIMEOUT));

[2] https://cwiki.apache.org/confluence/display/FLINK/FLIP-401%3A+REST+API+JSON+response+deserialization+unknown+field+tolerance

[gyfora]

Instead of a new rest client you could also do what is done in the Kubernetes-operator module, see:
https://github.com/apache/flink-kubernetes-operator/blob/main/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/IOMetricsInfo.java

By shadowing the class under the same package this will hide the incompatible implementation coming from flink

[gyfora]

But still it will not make it work fully for 1.15 as the aggregated metrics need to be back ported on the Flink side (similar to other features required for the autoscaler)

[1996fanrui]

Thanks @gyfora for the valuable feedback!

Instead of a new rest client you could also do what is done in the Kubernetes-operator module, see: https://github.com/apache/flink-kubernetes-operator/blob/main/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/IOMetricsInfo.java

By shadowing the class under the same package this will hide the incompatible implementation coming from flink

If so, autoscaler works with 1.15 when users uses the flink-kubernetes-operator, right?

I'm using the autoscaler Standalone, and I found 1.15 job doesn't work. In the short term, could I move the IOMetricsInfo to autoscaler module? IIUC, it will let both of flink-kubernetes-operator and autoscaler Standalone work.

In the long term, should we maintain the IOMetricsInfo class in the flink-kubernetes-operator repo or we disable DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES in the flink repo?

Looking forward to your opinion, and I'm happy to fix them.

But still it will not make it work fully for 1.15 as the aggregated metrics need to be back ported on the Flink side (similar to other features required for the autoscaler)

Thanks for the reminder, we have a lot of jobs run with 1.15 version, so we are backing port them. Of course, we (our internal platform) only analyse them and don't scale them when the version before 1.18.

[gyfora]

Thanks @gyfora for the valuable feedback!

Instead of a new rest client you could also do what is done in the Kubernetes-operator module, see: https://github.com/apache/flink-kubernetes-operator/blob/main/flink-kubernetes-operator/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/IOMetricsInfo.java
By shadowing the class under the same package this will hide the incompatible implementation coming from flink

If so, autoscaler works with 1.15 when users uses the flink-kubernetes-operator, right?

I'm using the autoscaler Standalone, and I found 1.15 job doesn't work. In the short term, could I move the IOMetricsInfo to autoscaler module? IIUC, it will let both of flink-kubernetes-operator and autoscaler Standalone work.

In the long term, should we maintain the IOMetricsInfo class in the flink-kubernetes-operator repo or we disable DeserializationFeature.FAIL_ON_NULL_FOR_PRIMITIVES in the flink repo?

Looking forward to your opinion, and I'm happy to fix them.

But still it will not make it work fully for 1.15 as the aggregated metrics need to be back ported on the Flink side (similar to other features required for the autoscaler)

Thanks for the reminder, we have a lot of jobs run with 1.15 version, so we are backing port them. Of course, we (our internal platform) only analyse them and don't scale them when the version before 1.18.

Not sure if we need to have the IOMetricsInfo override in both packages to be safe actually. I think it has to be copied.

I can personally confirm that given the aggregated metrics back ported to 1.15 (easy to do) + the IOMetricsInfo, the autoscaler works with 1.15

[gyfora]

looks good, if you tested it and works 🚢

[1996fanrui]

Thanks @gyfora and @mxm for the review!

looks good, if you tested it and works 🚢

I have tested locally, the main branch doesn't work for 1.15 job. And it works with current PR. Merging~