[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>
(cherry picked from commit 3a9f11f)
Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

Author: zahed1994

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 -

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") {