Skip to content

Commit 3a9f11f

Browse files
zahed1994viirya
authored andcommitted
[SPARK-55271][SS] Fix NullPointerException in Kafka micro-batch streaming metrics reporting
### What changes were proposed in this pull request? This PR fixes a `NullPointerException` in `KafkaMicroBatchStream.metrics()` during Kafka micro-batch streaming progress reporting (`finishTrigger`). Specifically, `KafkaMicroBatchStream.metrics()` checked `latestAvailablePartitionOffsets.isDefined` before extracting partition offsets. If `latestAvailablePartitionOffsets` was `Some(null)`, `isDefined` returned `true`, causing `latestAvailablePartitionOffsets.get` to return `null` and throwing a `NullPointerException` when `.map()` was invoked on it. This PR updates the condition to `latestAvailablePartitionOffsets.exists(_ != null)` to safely ensure partition offsets are non-null before invoking `.map()`. ### Why are the changes needed? When uninitialized partition offsets or race conditions occur during progress reporting (`finishTrigger`), `latestAvailablePartitionOffsets` can be `Some(null)`. Without this check, calling `.map()` on `null` throws a `NullPointerException`, crashing the entire streaming query job in production after batch execution. ### Does this PR introduce _any_ user-facing change? No API changes. Fixes a `NullPointerException` in progress reporting metrics. ### How was this patch tested? Added unit test assertion in `KafkaMicroBatchSourceSuite`. ### Was this patch authored or co-authored using generative AI tooling? No. Closes #58133 from zahed1994/SPARK-55271-kafka-streaming-metrics-npe. Authored-by: zahed1994 <zahedshareef@gmail.com> Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>
1 parent c3d9631 commit 3a9f11f

2 files changed

Lines changed: 3 additions & 1 deletion

File tree

connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -514,7 +514,7 @@ object KafkaMicroBatchStream extends Logging {
514514
latestAvailablePartitionOffsets: Option[PartitionOffsetMap]): ju.Map[String, String] = {
515515
val offset = Option(latestConsumedOffset.orElse(null))
516516

517-
if (offset.nonEmpty && latestAvailablePartitionOffsets.isDefined) {
517+
if (offset.nonEmpty && latestAvailablePartitionOffsets.exists(_ != null)) {
518518
val consumedPartitionOffsets = offset.map(KafkaSourceOffset(_)).get.partitionToOffsets
519519
val offsetsBehindLatest = latestAvailablePartitionOffsets.get
520520
.map(partitionOffset => partitionOffset._2 -

connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1930,6 +1930,8 @@ abstract class KafkaMicroBatchV2SourceSuite extends KafkaMicroBatchSourceSuiteBa
19301930

19311931
// test null latestAvailablePartitionOffsets
19321932
assert(KafkaMicroBatchStream.metrics(Optional.ofNullable(offset), None).isEmpty)
1933+
assert(KafkaMicroBatchStream.metrics(
1934+
Optional.ofNullable(offset), Some(null).asInstanceOf[Option[PartitionOffsetMap]]).isEmpty)
19331935
}
19341936

19351937
test("SPARK-57438: metrics should not NPE when latestPartitionOffsets is null") {

0 commit comments

Comments
 (0)