What is the purpose of the change

Fixes the silent failure of the topology spout lag query, where the Storm UI's "Topology Lag" panel stops working with this warning being logged repeatedly for every Kafka spout:

  TopologySpoutLag [WARN] Exception thrown while getting lag for spout id: <spoutId>
  TopologySpoutLag [WARN] Exception message:null
  java.lang.ClassCastException

The regression was introduced by the Kafka client bump from 3.9.0 → 4.x (#8243). Two issues, on the same code path:

1. KafkaOffsetLagUtil — NPE on partitions with no committed offset.
2. TopologySpoutLag — ClassCastException swallows the real error. When the kafka-monitor subprocess fails (the NPE above, or any other failure: broker unreachable, auth failure, bad config, etc.), it prints a plain-text error to stdout.

Together, (1) restores correct lag reporting for partitions with no committed offset, and (2) ensures any future monitor failure is reported usefully instead of as a generic ClassCastException with a null message.

How was the change tested

Integration test

• KafkaOffsetLagUtilTest#getOffsetLagsReportsCommittedOffsetForCommittedPartitionsAndMinusOneForUncommittedPartitions creates a 2-partition topic, commits an offset on partition 0 only, and calls the production KafkaOffsetLagUtil.getOffsetLags(...) end-to-end. Asserts the committed offset is reported for partition 0 and -1 for partition 1 (the regression case — pre-fix this NPE'd inside the monitor and surfaced as ClassCastException upstream in TopologySpoutLag).

Manual verification on a Storm 2.8.7 cluster

• Built storm-core and storm-kafka-monitor JARs and replaced them on the storm-ui node.
• Confirmed the previously-failing spouts (managementKafka, secondaryKafka, Instructions, …) now return lag results in the UI with no TopologySpoutLag warnings in ui.log.
• For a spout with a fresh consumer group (no committed offsets yet), confirmed lag reports correctly instead of throwing.

Issue: #8588