This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 1125957250c9 [SPARK-57438][SS] Fix NullPointerException in Kafka
source metrics when latest partition offsets are unavailable
1125957250c9 is described below
commit 1125957250c928167516a429efaf7dcb231a141c
Author: Anupam Yadav <[email protected]>
AuthorDate: Wed Jun 24 07:15:07 2026 -0700
[SPARK-57438][SS] Fix NullPointerException in Kafka source metrics when
latest partition offsets are unavailable
### What changes were proposed in this pull request?
Fix a `NullPointerException` in the Kafka micro-batch source when custom
metrics are reported before the latest partition offsets are known.
In the non-real-time-mode branch of the instance `metrics()`,
`latestPartitionOffsets` (a `var` initialized to null until `latestOffset()`
runs) was wrapped as `Some(latestPartitionOffsets)`, producing `Some(null)`.
The companion `metrics(...)` then evaluated `isDefined` as true, called `.get`
(null), and invoked `.map` on it, throwing. The fix changes that single wrap to
`Option(latestPartitionOffsets)`, so a null becomes `None` and the metrics
computation is correctly skipped.
### Why are the changes needed?
`metrics()` can be called before the first batch completes (e.g. progress
reporting), while `latestPartitionOffsets` is still null:
```
java.lang.NullPointerException: Cannot invoke
"scala.collection.IterableOps.map(scala.Function1)"
because the return value of "scala.Option.get()" is null
at
org.apache.spark.sql.kafka010.KafkaMicroBatchStream$.metrics(KafkaMicroBatchStream.scala:520)
```
This was introduced by #52729 (SPARK-54027, Kafka RTM support).
### Does this PR introduce _any_ user-facing change?
No behavioral change for well-formed inputs. It prevents the streaming
query from crashing when metrics are requested before the latest partition
offsets are available; in that case an empty metrics map is returned, as
intended.
### How was this patch tested?
Added a test in `KafkaMicroBatchSourceSuite` that drives the instance
`metrics()` while `latestPartitionOffsets` is still null (the stream is built
but `latestOffset()` has not run), asserting an empty map and no NPE. Verified
the test fails with the original NPE if the call-site fix is reverted to
`Some(...)`.
### Was this patch authored or co-authored using generative AI tooling?
Yes.
Closes #56526 from yadavay-amzn/fix/SPARK-57438-kafka-metrics-npe.
Authored-by: Anupam Yadav <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../spark/sql/kafka010/KafkaMicroBatchStream.scala | 2 +-
.../sql/kafka010/KafkaMicroBatchSourceSuite.scala | 26 ++++++++++++++++++++++
2 files changed, 27 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 828891f0b498..17323165b451 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
@@ -357,7 +357,7 @@ private[kafka010] class KafkaMicroBatchStream(
}
}
} else {
- Some(latestPartitionOffsets)
+ Option(latestPartitionOffsets)
}
KafkaMicroBatchStream.metrics(latestConsumedOffset,
reCalculatedLatestPartitionOffsets)
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 e6a794c22e1f..274be623da6f 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
@@ -1883,6 +1883,32 @@ abstract class KafkaMicroBatchV2SourceSuite extends
KafkaMicroBatchSourceSuiteBa
// test null latestAvailablePartitionOffsets
assert(KafkaMicroBatchStream.metrics(Optional.ofNullable(offset),
None).isEmpty)
}
+
+ test("SPARK-57438: metrics should not NPE when latestPartitionOffsets is
null") {
+ // Construct a KafkaMicroBatchStream instance without calling
latestOffset(),
+ // so latestPartitionOffsets remains null (its default uninitialized
value).
+ // Calling metrics() on this instance exercises the real non-RTM code path.
+ val topic = newTopic()
+ val tp = new TopicPartition(topic, 0)
+
+ SparkSession.setActiveSession(spark)
+ withTempDir { dir =>
+ val provider = new KafkaSourceProvider()
+ val options = Map(
+ "kafka.bootstrap.servers" -> testUtils.brokerAddress,
+ "subscribe" -> topic
+ )
+ val dsOptions = new CaseInsensitiveStringMap(options.asJava)
+ val table = provider.getTable(dsOptions)
+ val stream =
table.newScanBuilder(dsOptions).build().toMicroBatchStream(dir.getAbsolutePath)
+ .asInstanceOf[KafkaMicroBatchStream]
+
+ // latestPartitionOffsets is still null - metrics() must not NPE
+ val offset = KafkaSourceOffset(Map(tp -> 0L))
+ val result = stream.metrics(Optional.of(offset))
+ assert(result.isEmpty)
+ }
+ }
}
class KafkaMicroBatchV1SourceWithAdminSuite extends
KafkaMicroBatchV1SourceSuite {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
