fix the watermark

Author: zhisheng17

flink-learning-common/src/main/java/com/zhisheng/common/watermarks/MetricWatermark.java

@@ -16,10 +16,9 @@ public class MetricWatermark implements AssignerWithPeriodicWatermarks<MetricEve
 
     @Override
     public long extractTimestamp(MetricEvent metricEvent, long previousElementTimestamp) {
-        if (metricEvent.getTimestamp() > currentTimestamp) {
-            this.currentTimestamp = metricEvent.getTimestamp();
-        }
-        return currentTimestamp;
+        long timestamp = metricEvent.getTimestamp();
+        currentTimestamp = Math.max(timestamp, currentTimestamp);
+        return timestamp;
     }
 
     @Nullable

flink-learning-examples/src/main/java/com/zhisheng/examples/streaming/watermark/WordPeriodicWatermark.java

@@ -20,17 +20,13 @@ public class WordPeriodicWatermark implements AssignerWithPeriodicWatermarks<Wor
 
     @Override
     public long extractTimestamp(Word word, long previousElementTimestamp) {
-        this.currentTimestamp = word.getTimestamp();
+        long timestamp = word.getTimestamp();
+        currentTimestamp = Math.max(timestamp, currentTimestamp);
         log.info("event timestamp = {}, {}, CurrentWatermark = {}, {}", word.getTimestamp(),
                 DateUtil.format(word.getTimestamp(), YYYY_MM_DD_HH_MM_SS),
                 getCurrentWatermark().getTimestamp(),
                 DateUtil.format(getCurrentWatermark().getTimestamp(), YYYY_MM_DD_HH_MM_SS));
         return word.getTimestamp();
-
-//        if (word.getTimestamp() > currentTimestamp) {
-//            this.currentTimestamp = word.getTimestamp();
-//        }
-//        return currentTimestamp;
     }
 
     @Nullable

flink-learning-monitor/flink-learning-monitor-alert/src/main/java/com/zhisheng/alert/schema/OutageMetricWaterMark.java

@@ -22,9 +22,8 @@ public Watermark getCurrentWatermark() {
 
     @Override
     public long extractTimestamp(OutageMetricEvent outageMetricEvent, long l) {
-        if (outageMetricEvent.getTimestamp() > currentTimestamp) {
-            this.currentTimestamp = outageMetricEvent.getTimestamp();
-        }
-        return currentTimestamp;
+        long timestamp = outageMetricEvent.getTimestamp();
+        currentTimestamp = Math.max(timestamp, currentTimestamp);
+        return timestamp;
     }
 }

flink-learning-window/src/main/java/com/zhisheng/window/CustomTriggerMain.java

@@ -36,15 +36,14 @@ public static void main(String[] args) throws Exception {
                     @Nullable
                     @Override
                     public Watermark getCurrentWatermark() {
-                        return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - 1);
+                        return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - maxTimeLag);
                     }
 
                     @Override
                     public long extractTimestamp(WordEvent element, long previousElementTimestamp) {
-                        if (element.getTimestamp() > currentTimestamp) {
-                            this.currentTimestamp = element.getTimestamp();
-                        }
-                        return currentTimestamp;
+                        long timestamp = element.getTimestamp();
+                        currentTimestamp = Math.max(timestamp, currentTimestamp);
+                        return timestamp;
                     }
                 });
 

flink-learning-window/src/main/java/com/zhisheng/window/Main2.java

@@ -37,15 +37,14 @@ public static void main(String[] args) throws Exception {
                     @Nullable
                     @Override
                     public Watermark getCurrentWatermark() {
-                        return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - 1);
+                        return new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - maxTimeLag);
                     }
 
                     @Override
                     public long extractTimestamp(WordEvent element, long previousElementTimestamp) {
-                        if (element.getTimestamp() > currentTimestamp) {
-                            this.currentTimestamp = element.getTimestamp();
-                        }
-                        return currentTimestamp;
+                        long timestamp = element.getTimestamp();
+                        currentTimestamp = Math.max(timestamp, currentTimestamp);
+                        return timestamp;
                     }
                 });