[SPARK-49442][SS] Cache Kafka partition metadata to reduce DescribeTopics RPCs per micro-batch

[vinodkc]

What changes were proposed in this pull request?

Adds a new source option partition.metadata.cache.ttl.ms to KafkaOffsetReaderAdmin. When set to a positive value, the driver caches the set of assigned TopicPartitions and reuses it for all offset fetch calls within the TTL window, skipping the DescribeTopics RPC to the broker. The cache is invalidated on admin reset (i.e., on retry). Default is -1 (caching disabled, preserving existing behavior).

Why are the changes needed?

Every micro-batch unconditionally called consumerStrategy.assignedTopicPartitions(admin), which issues a DescribeTopics RPC to the Kafka broker even though partition assignments are stable in most production deployments. For deployments with large partition counts and short trigger intervals, this generates unnecessary metadata traffic.

Does this PR introduce any user-facing change?

Yes. Adds a new source option partition.metadata.cache.ttl.ms (long, default -1). Valid values are -1 (caching disabled) or any positive integer representing milliseconds. Documentation is updated accordingly. Existing behavior is unchanged when the option is not set.

Note: when caching is enabled, newly added partitions will not be discovered until the cache expires.

How was this patch tested?

Added tests in KafkaOffsetReaderSuite KafkaSourceProviderSuite, KafkaOffsetReaderSuite and KafkaSourceProviderSuite.

Was this patch authored or co-authored using generative AI tooling?

Yes, 'Generated-by: Claude (Sonnet 4.6)