[FLINK-32513][core] Add predecessor caching

[jeyhunkarimov]

What is the purpose of the change

This pull request adds caching capability to Transformation

Brief change log

• Change Transformation::getTransitivePredecessors() to utilize cache
• Introduce Transformation::getTransitivePredecessorsInternal()

Verifying this change

This change added tests to org.apache.flink.api.dag.TransformationTest::testPredecessorCache()

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)
• 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)

[flinkbot]

CI report:

• 7d343da Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[jeyhunkarimov]

Hi @zhuzhurk could you please review the PR in you available time? Thanks!

[zhuzhurk]

Thanks for creating the fix. @jeyhunkarimov
I have two comments. Please take a look.

[jeyhunkarimov]

Thanks @zhuzhurk for the review. I addressed them. Could you please check again?

[zhuzhurk]

Looks good to me.
Thanks for addressing all the comments! @jeyhunkarimov
Would you squash the last two commits and rebase the changes onto the latest Flink master branch?

[jeyhunkarimov]

Looks good to me. Thanks for addressing all the comments! @jeyhunkarimov

Thanks a lot @zhuzhurk for your reviews.

Would you squash the last two commits and rebase the changes onto the latest Flink master branch?

Done.