fix: record all deserialize cases

PeterZh6 · Oct 11, 2024 · 109ee8a · 109ee8a
1 parent bed796d
commit 109ee8a
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 1 deletion.
diff --git a/...sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceExactlyMetric.java b/...sort-flink/base/src/main/java/org/apache/inlong/sort/base/metric/SourceExactlyMetric.java
@@ -372,6 +372,12 @@ public void incNumDeserializeSuccess() {
         }
     }
 
+    public void decNumDeserializeSuccess() {
+        if (numDeserializeSuccess != null) {
+            numDeserializeSuccess.dec();
+        }
+    }
+
     public void incNumDeserializeError() {
         if (numDeserializeError != null) {
             numDeserializeError.inc();

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
@@ -96,6 +96,10 @@ public void deserialize(Message<byte[]> message, Collector<RowData> collector)
             throws IOException {
         try {
             long deserializeStartTime = System.currentTimeMillis();
+            // increase the number of deserialize success first, if deserialize failed, decrease it
+            if (sourceExactlyMetric != null) {
+                sourceExactlyMetric.incNumDeserializeSuccess();
+            }
             // Get the key row data
             List<RowData> keyRowData = new ArrayList<>();
             if (keyDeserialization != null) {
@@ -119,11 +123,11 @@ public void deserialize(Message<byte[]> message, Collector<RowData> collector)
                     message, keyRowData, valueRowData, metricsCollector);
             if (sourceExactlyMetric != null) {
                 sourceExactlyMetric.recordDeserializeDelay(System.currentTimeMillis() - deserializeStartTime);
-                sourceExactlyMetric.incNumDeserializeSuccess();
             }
         } catch (Exception e) {
             if (sourceExactlyMetric != null) {
                 sourceExactlyMetric.incNumDeserializeError();
+                sourceExactlyMetric.decNumDeserializeSuccess();
             }
             throw e;
         }