[FLINK-27925] [kubernetes]Performance optimization when watch tm pod and list pod.

[Mrart]

What is the purpose of the change

In the case of large-scale start and stop jobs, constantly reading data from etcd can cause a bottleneck in etcd performance. We can increase resourceversion=0 in watch to reduce data read from etcd.

Brief change log

• • Add withResourceVersion("0") in Fabric8FlinkKubeClient#watchPodsAndDoCallback

Verifying this change

Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing

(Please pick either of the following options)

This change is a trivial rework / code cleanup without any test coverage.

(or)

This change is already covered by existing tests, such as (please describe tests).

(or)

This change added tests and can be verified as follows:

(example:)

• Added integration tests for end-to-end deployment with large payloads (100MB)
• Extended integration test for recovery after master (JobManager) failure
• Added test that validates that TaskInfo is transferred only once across recoveries
• Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API, i.e., is any changed class annotated with @Public(Evolving): ( no)
• The serializers: （no / don't)
• The runtime per-record code paths (performance sensitive): (no)
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no )
• The S3 file system connector: (no)

Documentation

• Does this pull request introduce a new feature? (no)
• If yes, how is the feature documented? (not documented)

[flinkbot]

CI report:

• 94e155b Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[wangyang0918]

We need to add a test to guard this behavior.

[Mrart]

@wangyang0918 Added a unit test to test that we can use withResourceversion("0 ") to listen on pods events when pod resourceversion=5668.5668 is a random resource version.

[Mrart]

@flinkbot run azure

[huwh]

Thanks @Mrart prepare for this pr, I have left some comments, please take a look.

[reswqa]

Thanks @Mrart for this optimization and thanks @huwh for the review!

Apart from a suggestion for refactoring, my opinion is basically in line with Weihua. Please take a look~

[reswqa]

It seems that some conflicts need to be resolved.

[Mrart]

@flinkbot run azure

[reswqa]

Thanks for the update, I have no other comments. Let's wait and see if @huwh has any other concerns.

[huwh]

Thanks @Mrart and @reswqa for the update. The PR looks much better now. LGTM

[reswqa]

@Mrart It seems that the e2e test case is failed, can you rebase on latest master branch to see if it is fixed.

[Mrart]

@Mrart It seems that the e2e test case is failed, can you rebase on latest master branch to see if it is fixed.

Fixed