XiaoYou201 commented on code in PR #11130:
URL: https://github.com/apache/inlong/pull/11130#discussion_r1763622954
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/RowDataDebeziumDeserializeSchema.java:
##########
@@ -101,6 +102,9 @@ public interface ValueValidator extends Serializable {
/** Changelog Mode to use for encoding changes in Flink internal data
structure. */
private final DebeziumChangelogMode changelogMode;
private final MetricOption metricOption;
+
+ // Getter to make sourceExactlyMetric accessible to DebeziumSourceFunction
+ @Getter
Review Comment:
better to write a method~
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/DebeziumSourceFunction.java:
##########
@@ -317,6 +341,10 @@ public void snapshotState(FunctionSnapshotContext
functionSnapshotContext) throw
((RowDataDebeziumDeserializeSchema) deserializer)
.updateCurrentCheckpointId(functionSnapshotContext.getCheckpointId());
}
+ // record the start time of each checkpoint
+ long checkpointId = functionSnapshotContext.getCheckpointId();
+ checkpointStartTimeMap.put(checkpointId, System.currentTimeMillis());
+ sourceExactlyMetric.incNumSnapshotCreate();
Review Comment:
checkpointStartTimeMap.put(functionSnapshotContext.getCheckpointId(),
System.currentTimeMillis());
##########
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/DebeziumSourceFunction.java:
##########
@@ -496,6 +524,12 @@ public void notifyCheckpointComplete(long checkpointId) {
schema.flushAudit();
schema.updateLastCheckpointId(checkpointId);
}
+ // get the start time of the currently completed checkpoint
+ Long snapShotStartTimeById =
checkpointStartTimeMap.remove(checkpointId);
Review Comment:
why the remove method execute before the put method？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscr...@inlong.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
