[FLINK-38363][Runtime] WindowedStream.trigger supports trigger with async state

[jiexray]

WindowedStream allow registering AsyncTrigger for async state.

This pull request makes WindowedStream can register AsyncTrigger via trigger(). Before this pr, only internal window function (reduce, aggregate, ...) with enableAsyncState can register AsyncTrigger. After this pr, user can register AsyncTrigger in their own DataStream job via WindowedStream#trigger API.

Brief change log

• WindowedStream adds WindowedStream#trigger whose parameter type is AsyncTrigger.

Verifying this change

• Parameterize customize trigger() tests in WindowTranslationTest, where enableAsyncState is parameterized as true or false.

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

[jiexray]

@Zakelly Could you have a look at this pr, which resolves this ticket.

[flinkbot]

CI report:

• 14ff791 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[Zakelly]

Thanks for the addition! LGTM % CI. Will re-trigger it.

[Zakelly]

@flinkbot run azure

[jiexray]

@flinkbot run azure

[jiexray]

Thanks for the addition! LGTM % CI. Will re-trigger it.

@Zakelly Thank you for review. I have rebased on master and re-trigger CI.