[FLINK-38364][streaming-java] Implement async state version of ProcessingTimeoutTrigger

[jiexray]

Implement AsyncProcessingTimeoutTrigger, the async state version of ProcessingTimeoutTrigger

Before this pr, ProcessingTimeoutTrigger only supports sync state mode. After this pr, user can use AsyncProcessingTimeoutTrigger, and enable async state mode.

Brief change log

• Add AsyncProcessingTimeoutTrigger, support ProcessingTimeoutTrigger in async mode.

Verifying this change

• Parameterize ProcessingTimeoutTriggerTest to test both sync and async state mode.

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

[flinkbot]

CI report:

• 09a62f9 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[jiexray]

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

[Zakelly]

Thanks for the PR! I was thinking it's better we use this converter in AsyncTriggerConverter

[jiexray]

Thanks for the PR! I was thinking it's better we use this converter in AsyncTriggerConverter

@Zakelly Thank you for review. That's a good idea. I have intergrated ProcessingTimeoutTrigger to AsyncTriggerConverter.

The changes in brief:

• flink-runtime: Introduce a new interface (AsyncTriggerConvertable). A trigger implemented the interface has an API (convertToAsync) to spawn its async version.
• flink-streaming-java: ProcessingTimeoutTrigger implements AsyncTriggerConvertable.
• tests: ProcessingTimeoutTriggerTest uses AsyncTriggerConverter#convertToAsync to generate the async version of ProcessingTimeoutTrigger.

[jiexray]

@flinkbot run azure