abstract CustomerWaterMarker

Author: sishu@dtstack.com

core/src/main/java/com/dtstack/flink/sql/watermarker/AbsCustomerWaterMarker.java

@@ -21,13 +21,16 @@
 
 import com.dtstack.flink.sql.metric.EventDelayGauge;
 import com.dtstack.flink.sql.metric.MetricConstant;
+import com.dtstack.flink.sql.util.MathUtil;
 import org.apache.flink.api.common.functions.IterationRuntimeContext;
 import org.apache.flink.api.common.functions.RichFunction;
 import org.apache.flink.api.common.functions.RuntimeContext;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
 import org.apache.flink.streaming.api.windowing.time.Time;
 
+import java.util.TimeZone;
+
 /**
  * Reason:
  * Date: 2018/10/18
@@ -46,6 +49,12 @@ public abstract class AbsCustomerWaterMarker<T> extends BoundedOutOfOrdernessTim
 
     protected transient EventDelayGauge eventDelayGauge;
 
+    protected int pos;
+
+    protected long lastTime = 0;
+
+    protected TimeZone timezone;
+
     public AbsCustomerWaterMarker(Time maxOutOfOrderness) {
         super(maxOutOfOrderness);
     }
@@ -90,4 +99,13 @@ public void setRuntimeContext(RuntimeContext t) {
     public void setFromSourceTag(String fromSourceTag) {
         this.fromSourceTag = fromSourceTag;
     }
+
+    protected long getExtractTimestamp(Long extractTime){
+
+        lastTime = extractTime + timezone.getOffset(extractTime);
+
+        eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - extractTime)/1000));
+
+        return lastTime;
+    }
 }

core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForLong.java

@@ -39,14 +39,6 @@ public class CustomerWaterMarkerForLong extends AbsCustomerWaterMarker<Row> {
 
     private static final Logger logger = LoggerFactory.getLogger(CustomerWaterMarkerForLong.class);
 
-    private static final long serialVersionUID = 1L;
-
-    private int pos;
-
-    private long lastTime = 0;
-
-    private TimeZone timezone;
-
     public CustomerWaterMarkerForLong(Time maxOutOfOrderness, int pos,String timezone) {
         super(maxOutOfOrderness);
         this.pos = pos;
@@ -58,12 +50,7 @@ public long extractTimestamp(Row row) {
 
         try{
             Long extractTime = MathUtil.getLongVal(row.getField(pos));
-
-            lastTime = extractTime + timezone.getOffset(extractTime);
-
-            eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - extractTime)/1000));
-
-            return lastTime;
+            return getExtractTimestamp(extractTime);
         }catch (Exception e){
             logger.error("", e);
         }

core/src/main/java/com/dtstack/flink/sql/watermarker/CustomerWaterMarkerForTimeStamp.java

@@ -40,14 +40,6 @@ public class CustomerWaterMarkerForTimeStamp extends AbsCustomerWaterMarker<Row>
 
     private static final Logger logger = LoggerFactory.getLogger(CustomerWaterMarkerForTimeStamp.class);
 
-    private static final long serialVersionUID = 1L;
-
-    private int pos;
-
-    private long lastTime = 0;
-
-    private TimeZone timezone;
-
     public CustomerWaterMarkerForTimeStamp(Time maxOutOfOrderness, int pos,String timezone) {
         super(maxOutOfOrderness);
         this.pos = pos;
@@ -58,14 +50,8 @@ public CustomerWaterMarkerForTimeStamp(Time maxOutOfOrderness, int pos,String ti
     public long extractTimestamp(Row row) {
         try {
             Timestamp time = (Timestamp) row.getField(pos);
-
             long extractTime=time.getTime();
-
-            lastTime = extractTime + timezone.getOffset(extractTime);
-
-            eventDelayGauge.setDelayTime(MathUtil.getIntegerVal((System.currentTimeMillis() - extractTime)/1000));
-
-            return lastTime;
+            return getExtractTimestamp(extractTime);
         } catch (RuntimeException e) {
             logger.error("", e);
         }