add flink metrics for token tableInsertCount,tableUpdateCount,tableDeleteCount

Author: baisui1981

tis-incr/tis-flink-cdc-kingbase-plugin/src/main/java/com/qlangtech/plugins/incr/flink/cdc/kingbase/FlinkCDCKingBaseSourceFunction.java

@@ -52,6 +52,7 @@
  * @create: 2025-01-14 19:11
  **/
 public class FlinkCDCKingBaseSourceFunction extends FlinkCDCPGLikeSourceFunction {
+
     public FlinkCDCKingBaseSourceFunction(FlinkCDCKingBaseSourceFactory sourceFactory) {
         super(sourceFactory);
     }
@@ -110,6 +111,7 @@ public boolean visit(DBConfig config, String jdbcUrl, String ip, String dbName)
 //             .deserializer(new PostgreSQLDeserializationSchema(tabs, flinkColCreator, contextParamValsGetterMapper, sourceFactory.getRepIdentity()))
 //                .build();
 
+
         PostgreSQLDeserializationSchema deserializer = new PostgreSQLDeserializationSchema(tabs, flinkColCreator, contextParamValsGetterMapper, sourceFactory.getRepIdentity());
 
         PostgresOffsetFactory offsetFactory = new PostgresOffsetFactory();

tis-incr/tis-flink-cdc-kingbase-plugin/src/main/java/com/qlangtech/plugins/incr/flink/cdc/pglike/PostgreSQLDeserializationSchema.java

@@ -27,6 +27,8 @@
 import com.qlangtech.tis.realtime.transfer.DTO;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.Struct;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.util.List;
 import java.util.Map;
@@ -38,6 +40,7 @@
  **/
 public class PostgreSQLDeserializationSchema extends TISDeserializationSchema {
     private final ReplicaIdentity replicaIdentity;
+    private static final Logger logger = LoggerFactory.getLogger(PostgreSQLDeserializationSchema.class);
 
     /**
      * @param tabs
@@ -51,6 +54,7 @@ public PostgreSQLDeserializationSchema(List<ISelectedTab> tabs, IFlinkColCreator
     ) {
         super(new PGDTOColValProcess(tabs, flinkColCreator), new DefaultTableNameConvert(), contextParamValsGetterMapper);
         this.replicaIdentity = replicaIdentity;
+        logger.info("create deserializationSchema with replicaIdentity:{}", replicaIdentity);
     }
 
     @Override

tis-incr/tis-flink-extends/src/main/java/com/qlangtech/plugins/incr/flink/metrics/TISPBReporter.java

@@ -110,7 +110,7 @@ public void notifyOfAddedMetric(Metric metric, String metricName, MetricGroup gr
         super.notifyOfAddedMetric(metric, metricName, group);
         String name = group.getMetricIdentifier(metricName, this);
         if (scope.length > 1 && TaskExecutor.TASK_MANAGER_NAME.equals(scope[1])) {
-            if (IIncreaseCounter.TABLE_CONSUME_COUNT.equals(metricName)) {
+            if (IIncreaseCounter.COLLECTABLE_TABLE_COUNT_METRIC.contains(metricName)) {
                 //  System.out.println(metricName);
                 metricIdentifierMapper.put(name, Pair.of(metricName, group));
             }
@@ -168,7 +168,7 @@ public void report() {
             // System.out.println(metricID + ": " + counter.getCount());
             metricGroup = metricIdentifierMapper.get(metricID);
             if (metricGroup != null) {
-               // System.out.println(metricGroup);
+                // System.out.println(metricGroup);
 
                 metrics.add(new UseableMetricForTIS(counter, /**metricName*/metricGroup.getKey(), metricGroup.getRight()));
             }

tis-incr/tis-flink-extends/src/main/java/com/qlangtech/tis/realtime/DTOSourceTagProcessFunction.java

@@ -76,6 +76,20 @@ protected void increaseNumRecordsMetric(DTO in) {
             // 当记录为更新时候会有before，after两条，所以需要将before那条记录过滤掉
             return;
         }
+        switch (in.getEventType()) {
+            case ADD: {
+                this.tisInsertNumRecordsIn.inc();
+                break;
+            }
+            case UPDATE_AFTER: {
+                this.tisUpdateNumRecordsIn.inc();
+                break;
+            }
+            case DELETE: {
+                this.tisDeleteNumRecordsIn.inc();
+                break;
+            }
+        }
         super.increaseNumRecordsMetric(in);
     }
 

tis-incr/tis-flink-extends/src/main/java/com/qlangtech/tis/realtime/SourceProcessFunction.java

@@ -56,6 +56,9 @@ public abstract class SourceProcessFunction<RECORD_TYPE> extends BroadcastProces
      * 包括所有的数据流量（增删改）
      */
     private transient Counter tisNumRecordsIn;
+    protected transient Counter tisInsertNumRecordsIn;
+    protected transient Counter tisUpdateNumRecordsIn;
+    protected transient Counter tisDeleteNumRecordsIn;
     private final AtomicBoolean drain = new AtomicBoolean(false);
     // private String taskId;
 
@@ -84,6 +87,18 @@ public void open(Configuration parameters) throws Exception {
                 .getMetricGroup()
                 .counter(IIncreaseCounter.TABLE_CONSUME_COUNT);
 
+        this.tisInsertNumRecordsIn = getRuntimeContext()
+                .getMetricGroup()
+                .counter(IIncreaseCounter.TABLE_INSERT_COUNT);
+
+        this.tisUpdateNumRecordsIn = getRuntimeContext()
+                .getMetricGroup()
+                .counter(IIncreaseCounter.TABLE_UPDATE_COUNT);
+
+        this.tisDeleteNumRecordsIn = getRuntimeContext()
+                .getMetricGroup()
+                .counter(IIncreaseCounter.TABLE_DELETE_COUNT);
+
         // 获取广播状态
         //   ValueStateDescriptor<Double> descriptor = new ValueStateDescriptor<>("rate", Double.class);
         // broadcastState = getRuntimeContext().getBroadcastState(descriptor);