This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 7753a7d30ced [SPARK-55271][SS] Fix NullPointerException in Kafka
micro-batch streaming metrics reporting
7753a7d30ced is described below
commit 7753a7d30cedf98b21dbf1f56bb5c3228c3f6098
Author: zahed1994 <[email protected]>
AuthorDate: Thu Aug 27 07:45:48 2026 -0700
[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 <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
(cherry picked from commit 3a9f11ff9fae4fb3870a9cd73b1992c3f86a6eac)
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala | 2 +-
.../org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala | 2 ++
2 files changed, 3 insertions(+), 1 deletion(-)
diff --git
a/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
b/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
index 17323165b451..c7e27eb2cd0e 100644
---
a/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
+++
b/connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
@@ -514,7 +514,7 @@ object KafkaMicroBatchStream extends Logging {
latestAvailablePartitionOffsets: Option[PartitionOffsetMap]):
ju.Map[String, String] = {
val offset = Option(latestConsumedOffset.orElse(null))
- if (offset.nonEmpty && latestAvailablePartitionOffsets.isDefined) {
+ if (offset.nonEmpty && latestAvailablePartitionOffsets.exists(_ != null)) {
val consumedPartitionOffsets =
offset.map(KafkaSourceOffset(_)).get.partitionToOffsets
val offsetsBehindLatest = latestAvailablePartitionOffsets.get
.map(partitionOffset => partitionOffset._2 -
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
index 22604328765e..cde607f23453 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
@@ -1852,6 +1852,8 @@ abstract class KafkaMicroBatchV2SourceSuite extends
KafkaMicroBatchSourceSuiteBa
// test null latestAvailablePartitionOffsets
assert(KafkaMicroBatchStream.metrics(Optional.ofNullable(offset),
None).isEmpty)
+ assert(KafkaMicroBatchStream.metrics(
+ Optional.ofNullable(offset),
Some(null).asInstanceOf[Option[PartitionOffsetMap]]).isEmpty)
}
test("SPARK-57438: metrics should not NPE when latestPartitionOffsets is
null") {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
