set withIdleness as parameter

Author: congdai-airwallex

.github/workflows/build.yaml

@@ -30,7 +30,7 @@ jobs:
           GPG_PASSWORD: ${{ secrets.GPG_KEY_PASSWORD }}
           GPG_SECRET: ${{ secrets.GPG_SECRET_KEY }}
       - name: Push Maven Artifacts to SonaType (only on Master)
-        if: ${{ github.ref == 'refs/heads/master' }}
+#        if: ${{ github.ref == 'refs/heads/master' }}
         uses: gradle/gradle-build-action@0d13054264b0bb894ded474f08ebb30921341cee
         with:
           arguments: publishToSonatype closeAndReleaseSonatypeStagingRepository --info

core/src/main/java/com/airwallex/airskiff/flink/AbstractFlinkCompiler.java

@@ -48,11 +48,14 @@ public abstract class AbstractFlinkCompiler implements Compiler<DataStream<?>> {
   protected final StreamExecutionEnvironment env;
   protected final StreamTableEnvironment tableEnv;
   protected final Duration allowedLatency;
+  protected final Duration withIdleness;
 
-  public AbstractFlinkCompiler(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, Duration allowedLatency) {
+  public AbstractFlinkCompiler(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, Duration allowedLatency,
+                               Duration withIdleness) {
     this.env = env;
     this.tableEnv = tableEnv;
     this.allowedLatency = allowedLatency;
+    this.withIdleness = withIdleness;
     Utils.registerFunctions(this.tableEnv);
   }
 
@@ -95,7 +98,7 @@ protected <K, T, U> KeyedStream<Tuple2<Long, Pair<K, Pair<T, U>>>, K> compileLef
 
     final LeftJoinPairMonoid<T, U> m = new LeftJoinPairMonoid<>();
     DataStream<Tuple2<Long, Pair<K, Pair<T, U>>>> ss = ks2.map(x -> new Tuple2<>(x.f0, new Pair<>(x.f1.l, new Pair<T, U>(null, x.f1.r))), outputType).union(ks1.map(x -> new Tuple2<>(x.f0, new Pair<>(x.f1.l, new Pair<T, U>(x.f1.r, null))), outputType));
-    return ss.assignTimestampsAndWatermarks(Utils.watermark(isBatch(), Duration.ZERO)).keyBy(t -> t.f1.l, kType)
+    return ss.assignTimestampsAndWatermarks(Utils.watermark(isBatch(), Duration.ZERO, Duration.ofMillis(300))).keyBy(t -> t.f1.l, kType)
       // a window to make sure if we have multiple events happening
       // at the same time, U is always put before T in batch mode
       .window(TumblingEventTimeWindows.of(Time.days(1))).allowedLateness(Time.days(1)).process(new ProcessWindowFunction<Tuple2<Long, Pair<K, Pair<T, U>>>, Tuple2<Long, Pair<K, Pair<T, U>>>, K, TimeWindow>() {
@@ -179,7 +182,8 @@ protected <K, T> KeyedStream<Tuple2<Long, Pair<K, T>>, K> compileOrderedSum(Orde
   }
 
   protected <T> DataStream<Tuple2<Long, T>> compileConcat(ConcatStream<T> stream) {
-    return compile(stream.a).union(compile(stream.b)).assignTimestampsAndWatermarks(Utils.watermark(isBatch(), this.allowedLatency));
+    return compile(stream.a).union(compile(stream.b))
+      .assignTimestampsAndWatermarks(Utils.watermark(isBatch(), this.allowedLatency, this.withIdleness));
   }
 
   protected <T> DataStream<Tuple2<Long, T>> compileFilter(FilterStream<T> stream) {
@@ -192,7 +196,7 @@ protected <T, U> DataStream<Tuple2<Long, U>> compileSql(final SqlStream<T, U> st
     final var typeInfo = stream.typeInfo;
     return tableEnv.toAppendStream(table, Row.class).map(r -> new Tuple2<>((Long) r.getField(0), mapper.map(r)), typeInfo)
       // watermark and timestamp is lost after table to data stream conversion?
-      .assignTimestampsAndWatermarks(Utils.watermark(isBatch(), this.allowedLatency));
+      .assignTimestampsAndWatermarks(Utils.watermark(isBatch(), this.allowedLatency, this.withIdleness));
   }
 
   protected <K, T> KeyedStream<Tuple2<Long, Pair<K, T>>, K> compileKS(KStream<K, T> ks) {

core/src/main/java/com/airwallex/airskiff/flink/FlinkLocalTextConfig.java

@@ -31,6 +31,6 @@ public DataStream<Tuple2<Long, String>> source(
       stream = env.socketTextStream("localhost", 10000);
     }
     return stream.map(t -> new Tuple2<>(System.currentTimeMillis(), t), tuple2TypeInfo(String.class))
-      .assignTimestampsAndWatermarks(Utils.watermark(isBatch, Duration.ZERO));
+      .assignTimestampsAndWatermarks(Utils.watermark(isBatch, Duration.ZERO, Duration.ofMillis(300)));
   }
 }

core/src/main/java/com/airwallex/airskiff/flink/Utils.java

@@ -28,12 +28,12 @@
 
 public class Utils {
 
-  public static <T> WatermarkStrategy<Tuple2<Long, T>> watermark(boolean isBatch, Duration allowedLatency) {
+  public static <T> WatermarkStrategy<Tuple2<Long, T>> watermark(boolean isBatch, Duration allowedLatency, Duration withIdleness) {
     if (isBatch) {
       return WatermarkStrategy.<Tuple2<Long, T>>forBoundedOutOfOrderness(Duration.ofDays(365 * 10000))
         .withTimestampAssigner((t, l) -> t.f0);
     } else {
-      return new RealtimeWatermarkStrategy<Tuple2<Long, T>>(Duration.ofDays(5), allowedLatency).withTimestampAssigner((t, l) -> t.f0).withIdleness(Duration.ofMillis(300));
+      return new RealtimeWatermarkStrategy<Tuple2<Long, T>>(Duration.ofDays(5), allowedLatency).withTimestampAssigner((t, l) -> t.f0).withIdleness(withIdleness);
     }
   }
 

core/src/test/java/com/airwallex/airskiff/testhelpers/TestFlinkConfig.java

@@ -25,6 +25,6 @@ public Class<T> clz() {
 
   @Override
   public DataStream<Tuple2<Long, T>> source(StreamExecutionEnvironment env, boolean isBatch) {
-    return env.fromCollection(_data).assignTimestampsAndWatermarks(Utils.watermark(isBatch, Duration.ZERO));
+    return env.fromCollection(_data).assignTimestampsAndWatermarks(Utils.watermark(isBatch, Duration.ZERO, Duration.ofMillis(300)));
   }
 }

gradle.properties

@@ -1,6 +1,6 @@
 group=com.airwallex.airskiff
 name=core
-version=2.0.60
+version=2.0.61
 org.gradle.daemon=true
 org.gradle.parallel=true
 file.encoding=utf-8