fix

vernedeng · Nov 10, 2023 · a473c5a · a473c5a
1 parent 7154bbc
commit a473c5a
Showing 1 changed file with 1 addition and 1 deletion.
diff --git a/...r/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarTableDeserializationSchema.java b/...r/src/main/java/org/apache/inlong/sort/pulsar/table/PulsarTableDeserializationSchema.java
@@ -118,7 +118,7 @@ public void deserialize(Message<byte[]> message, Collector<RowData> collector)
 
         // reset time stamp if the deserialize schema has not inner format
         if (!innerFormat) {
-            metricsCollector.resetTimestamp(message.getEventTime());
+            metricsCollector.resetTimestamp(System.currentTimeMillis());
         }
 
         valueDeserialization.deserialize(message.getData(), metricsCollector);