Add NPE check for updateNumBytesInCounter

pinterest · Oct 30, 2024 · 8c9ebbe · 8c9ebbe
1 parent 8be3077
commit 8c9ebbe
Showing 1 changed file with 1 addition and 1 deletion.
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
@@ -239,7 +239,7 @@ public void removeRecordsLagMetric(TopicUriPartition tp) {
      * bytes-consumed-total and count it towards the counter.
      */
     public void updateNumBytesInCounter() {
-        if (this.bytesConsumedTotalMetric != null) {
+        if (this.bytesConsumedTotalMetric != null && this.bytesConsumedTotalMetric.metricValue() != null) {
             long bytesConsumedUntilNow =
                     ((Number) this.bytesConsumedTotalMetric.metricValue()).longValue();
             long bytesConsumedSinceLastUpdate = bytesConsumedUntilNow - latestBytesConsumedTotal;