Surround NPE in PscSourceReaderMetrics with try/catch

pinterest · Oct 30, 2024 · 5b19053 · 5b19053
1 parent 8c9ebbe
commit 5b19053
Showing 1 changed file with 12 additions and 8 deletions.
diff --git a/...rc/main/java/com/pinterest/flink/connector/psc/source/metrics/PscSourceReaderMetrics.java b/...rc/main/java/com/pinterest/flink/connector/psc/source/metrics/PscSourceReaderMetrics.java
@@ -240,14 +240,18 @@ public void removeRecordsLagMetric(TopicUriPartition tp) {
      */
     public void updateNumBytesInCounter() {
         if (this.bytesConsumedTotalMetric != null && this.bytesConsumedTotalMetric.metricValue() != null) {
-            long bytesConsumedUntilNow =
-                    ((Number) this.bytesConsumedTotalMetric.metricValue()).longValue();
-            long bytesConsumedSinceLastUpdate = bytesConsumedUntilNow - latestBytesConsumedTotal;
-            this.sourceReaderMetricGroup
-                    .getIOMetricGroup()
-                    .getNumBytesInCounter()
-                    .inc(bytesConsumedSinceLastUpdate);
-            latestBytesConsumedTotal = bytesConsumedUntilNow;
+            try {
+                long bytesConsumedUntilNow =
+                        ((Number) this.bytesConsumedTotalMetric.metricValue()).longValue();
+                long bytesConsumedSinceLastUpdate = bytesConsumedUntilNow - latestBytesConsumedTotal;
+                this.sourceReaderMetricGroup
+                        .getIOMetricGroup()
+                        .getNumBytesInCounter()
+                        .inc(bytesConsumedSinceLastUpdate);
+                latestBytesConsumedTotal = bytesConsumedUntilNow;
+            } catch (Exception e) {
+                LOG.warn("Failed to update numBytesInCounter", e);
+            }
         }
     }