Batch stream offset fetch in KafkaStreamMetadataProvider

[shounakmk219]

Problem

RealtimeSegmentValidationManager (RVM) runs on the lead controller and calls PinotLLCRealtimeSegmentManager.ensureAllPartitionsConsuming, which fetches stream offsets inside the per-table Helix ideal-state update lock (HelixHelper.updateIdealState → IdealStateGroupCommit).

KafkaStreamMetadataProvider.computePartitionGroupMetadata fetched those offsets one partition at a time, creating a fresh KafkaConsumer per partition via createPartitionMetadataProvider (~1s each). On a table with ~1024 partitions this meant hundreds of serial consumer creations while holding the ideal-state lock, so concurrent segment commits stalled for minutes.

Observed on a production controller log: updating ideal state: ~291000ms on stalled commits versus a ~180ms median.

Fix

Resolve all partitions that need a stream fetch in a single batched call. Kafka's beginningOffsets(Collection) / endOffsets(Collection) / offsetsForTimes(Map) each resolve a whole collection of partitions in one broker round-trip and do not require the consumer to be assigned to those partitions (the existing fetchLatestStreamOffset(Set, ...) already relies on this).

• New private helper fetchOffsetsForPartitions(Collection<Integer>, OffsetCriteria, long) performs the batched, criteria-aware fetch (SMALLEST/LARGEST/PERIOD/TIMESTAMP, with the period/timestamp → end-offset fallback preserved).
• computePartitionGroupMetadata collects the partitions not already covered by a consumption status and fetches them all at once.
• fetchStreamPartitionOffset (single partition) now delegates to the same helper.
• Applied identically to pinot-kafka-3.0 and pinot-kafka-4.0. Kinesis and Pulsar have their own computePartitionGroupMetadata (shard/partition-group semantics) and are intentionally untouched.

This collapses ~1024 serial consumer creations into one round-trip, cutting the RVM lock-hold from minutes to seconds.

Note on duplication

The helper is duplicated across pinot-kafka-3.0 and pinot-kafka-4.0, matching the existing intentional split between the two Kafka client versions (the code reads per-module instance state). Kept in sync between the modules.

Follow-up (separate PR)

Move the offset fetch entirely out of the ideal-state lock in PinotLLCRealtimeSegmentManager.ensureAllPartitionsConsuming (pre-fetch before HelixHelper.updateIdealState, pass an immutable snapshot into the updater), so the lock-hold is proportional to the ideal-state mutation and the fetch is not re-run on CAS retries.

Testing

• KafkaStreamMetadataProviderTest in both modules (10 tests each, all passing): single-batched-call guarantee (ArgumentCaptor asserts all partitions in one call), SMALLEST / LARGEST / PERIOD / TIMESTAMP-with-fallback, dead-partition-omitted, and fetchStreamPartitionOffset happy + missing-offset paths.
• spotless / checkstyle / license clean on both modules.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 92.00000% with 4 lines in your changes missing coverage. Please review.
✅ Project coverage is 65.52%. Comparing base (1aa0e4d) to head (1913f42).

Files with missing lines	Patch %	Lines
...in/stream/kafka30/KafkaStreamMetadataProvider.java	92.00%	1 Missing and 3 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19116      +/-   ##
============================================
- Coverage     65.54%   65.52%   -0.03%     
  Complexity     1423     1423              
============================================
  Files          3432     3432              
  Lines        218107   218127      +20     
  Branches      34661    34671      +10     
============================================
- Hits         142960   142929      -31     
- Misses        63598    63633      +35     
- Partials      11549    11565      +16     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	65.52% <92.00%> (-0.03%)	⬇️
temurin	65.52% <92.00%> (-0.03%)	⬇️
unittests	65.52% <92.00%> (-0.03%)	⬇️
unittests1	56.87% <ø> (-0.04%)	⬇️
unittests2	37.91% <92.00%> (+0.01%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.