[FLINK-35289][core] Batch mode onTimer should support processing ts

[jeyhunkarimov]

What is the purpose of the change

Fix incorrect timestamp of stream elements collected from onTimer in batch mode.
In batch mode, onTimer calls will be processed at the end of the data. As a result, whichever timer registered by the users, they will all be triggered at the end. Therefore, it makes sense to trigger them with the processing time timestamp.

Brief change log

• Adjust the timer processing logic of BatchExecutionInternalTimeService
• Adjust tests

Verifying this change

Via tests in BatchExecutionInternalTimeServiceTest

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:

• affc700 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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

[jeyhunkarimov]

@flinkbot run azure

[reswqa]

Thanks for the PR, but I'm not sure about this. AFAIK, processing time doesn't make much sense in batch mode. In particular, we return a processing time for event time timer.

[jeyhunkarimov]

Hi @reswqa Thanks for your comment. The issue is that in batch mode all event-time triggers will be processed at the end. So, no matter what is the event-time set, they will be all triggered at the end of data. So, for example it doesnt make sense to process Long.MAX_VALUE event time - which will be triggered at the end of data with the timestamp Long.MAX_VALUE. WDYT?