[flink] 03/03: [FLINK-23960][streaming] checkScheduledTimestamps changed according to semantic of the calculation processing time delay

[pn...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git

commit 25b985de06b3c97a550b67ed791409cd6dcbb719
Author: Anton Kalashnikov <ka...@yandex.ru>
AuthorDate: Fri Sep 3 19:54:08 2021 +0200

    [FLINK-23960][streaming] checkScheduledTimestamps changed according to semantic of the calculation processing time delay
---
 .../flink/streaming/runtime/operators/StreamTaskTimerTest.java   | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
index 765075d..f62dcea 100644
--- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
+++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/operators/StreamTaskTimerTest.java
@@ -113,7 +113,14 @@ public class StreamTaskTimerTest extends TestLogger {
         long currentTimeMillis = System.currentTimeMillis();
         ArrayList<ValidatingProcessingTimeCallback> timeCallbacks = new ArrayList<>();
 
-        timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis, 0));
+        /*
+         It is not possible to test registering timer for currentTimeMillis or value slightly greater than
+         currentTimeMillis because if the during registerTimer the internal currentTime is equal
+         to this value then according to current logic the time will be increased for 1ms while
+         `currentTimeMillis - 200` is always transform to 0, so it can lead to reordering. See
+         comment in {@link ProcessingTimeServiceUtil#getRecordProcessingTimeDelay(long, long)}.
+        */
+        timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis - 1, 0));
         timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis - 200, 1));
         timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis + 100, 2));
         timeCallbacks.add(new ValidatingProcessingTimeCallback(currentTimeMillis + 200, 3));