[FLINK-34371][runtime] Support EndOfStreamTrigger and isOutputOnlyAfterEndOfStream operator attribute to optimize task deployment

[yunfengzhou-hub]

What is the purpose of the change

This pull request adds support for EndOfStreamTrigger and isOutputOnlyAfterEndOfStream operator attribute to optimize task deployment.

Brief change log

• Introduces OperatorAttributes with isOutputOnlyAfterEndOfStream attribute
• Schedules operators with isOutputOnlyAfterEndOfStream=true to run in blocking mode
• Make WindowOperator and StreamSortOperator return isOutputOnlyAfterEndOfStream=true in certain cases

Verifying this change

• This change adds unit tests for OperatorAttributes and the corresponding blocking mode changes.
• The change on WindowOperator and StreamSortOperator is covered by existing tests.

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): yes
• The serializers: no
• The runtime per-record code paths (performance sensitive): yes
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? JavaDocs

[flinkbot]

CI report:

• 6567058eee9d67fb1c9ea3e1d160204a568f745a Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[yunfengzhou-hub]

Thanks for the comments @Sxnan @mohitjain2504 . I have updated the PR according to the comments.

[Sxnan]

@yunfengzhou-hub Thanks for the update. LGTM!

[mohitjain2504]

What is the reason for closing? Did I miss something?

[xintongsong]

@mohitjain2504 It's merged.