refactor: used method instead of @Setter and removed redundant variables

apache · Sep 23, 2024 · ab4420f · ab4420f
1 parent e13b725
commit ab4420f
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 4 deletions.
diff --git a/...ors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/DebeziumSourceFunction.java b/...ors/postgres-cdc/src/main/java/org/apache/inlong/sort/postgre/DebeziumSourceFunction.java
@@ -353,9 +353,8 @@ private void doSnapshotState(FunctionSnapshotContext functionSnapshotContext) th
                     .updateCurrentCheckpointId(functionSnapshotContext.getCheckpointId());
         }
         // record the start time of each checkpoint
-        Long checkpointId = functionSnapshotContext.getCheckpointId();
         if (checkpointStartTimeMap != null) {
-            checkpointStartTimeMap.put(checkpointId, System.currentTimeMillis());
+            checkpointStartTimeMap.put(functionSnapshotContext.getCheckpointId(), System.currentTimeMillis());
         } else {
             LOG.error("checkpointStartTimeMap is null, can't record the start time of checkpoint");
         }

diff --git a/...es-cdc/src/main/java/org/apache/inlong/sort/postgre/RowDataDebeziumDeserializeSchema.java b/...es-cdc/src/main/java/org/apache/inlong/sort/postgre/RowDataDebeziumDeserializeSchema.java
@@ -34,7 +34,6 @@
 import io.debezium.time.NanoTime;
 import io.debezium.time.NanoTimestamp;
 import io.debezium.time.Timestamp;
-import lombok.Setter;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.table.data.DecimalData;
 import org.apache.flink.table.data.GenericRowData;
@@ -102,7 +101,6 @@ public interface ValueValidator extends Serializable {
     private final DebeziumChangelogMode changelogMode;
 
     /** Self-defined Flink metrics, which will be set by DebeziumSourceFunction with setter */
-    @Setter
     private SourceExactlyMetric sourceExactlyMetric;
 
     /** Returns a builder to build {@link RowDataDebeziumDeserializeSchema}. */
@@ -712,4 +710,9 @@ public void updateLastCheckpointId(long checkpointId) {
             sourceExactlyMetric.updateLastCheckpointId(checkpointId);
         }
     }
+
+    /** setter to enable DebeziumSourceFunction to set the metric */
+    public void setSourceExactlyMetric(SourceExactlyMetric sourceExactlyMetric) {
+        this.sourceExactlyMetric = sourceExactlyMetric;
+    }
 }