shounakmk219 opened a new pull request, #19116: URL: https://github.com/apache/pinot/pull/19116
## 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
