xiangfu0 commented on PR #18687: URL: https://github.com/apache/pinot/pull/18687#issuecomment-4628016499
## Review notes Traced the change end-to-end (Kafka providers → `PinotLLCRealtimeSegmentManager` consumers) and cross-checked with an independent reviewer. **No CRITICAL/MAJOR issues** — the fix is correct, scoped to the real root cause, preserves steady-state behavior, and has a genuine regression test. Three MINOR follow-ups noted below. ### Why the fix is correct The empty-subset branch used to delegate to the SPI default, which enumerates new partitions with `for (i = statuses.size(); i < partitionCount; i++)` — it treats the **status-list count** as the next partition id. With a missing low id (a partition that lost all its segments), that yields e.g. `[0,1,3,4,5,6,7,7]`: partition `2` omitted, `7` duplicated. The new code keys current statuses by `getStreamPartitionGroupId()`, enumerates the **actual** Kafka ids from `fetchPartitionIds()` (sorted), and per id reuses the status `endOffset` if present else fetches `fetchStreamPartitionOffset(offsetCriteria)` → `[0..7]`. The recovered id then flows into `setupNewPartitionGroup` (`PinotLLCRealtimeSegmentManager.java`, "Set up new partitions if not exist" loop), so the consuming segment is actually recreated. One precision note on impact: the manifestation is **silent non-recovery, not an NPE**. The offset-repair path is guarded by `containsKey` before `selectStartOffset`, and the smallest-offset map is built with `forceGetOffsetFromStream=true` (empty statuses → contiguous `[0..N-1]`, never sparse-broken). So the bug omits the hole id from the new-partition setup loop, which then never runs for it. ### Parity / safety verified - **Steady state unchanged**: when every partition has a status, the new path emits the same `(id, endOffset)` pairs, just sorted by id rather than status-list order. All controller consumers key by partition id, so ordering is irrelevant; no provider connections are opened when there are no holes. - **Multi-topic**: keys on `getStreamPartitionGroupId()` (raw stream id), matching `fetchPartitionIds()` space; `PartitionGroupMetadataFetcher` re-encodes raw→Pinot ids afterward — same contract the old SPI default relied on. - **`forceGetOffsetFromStream`**: Kafka doesn't override the 5-arg overload, so the repair path still passes an empty list → all partitions fetched from stream. Controller change is comment-only. - **Resources / imports / concurrency**: per-partition providers are try-with-resources closed; required `java.util` + `StreamConsumerFactory*` imports present in both 3.0 and 4.0; `_partitionIdSubset` is read-only; no new shared mutable state. ### Minor follow-ups (non-blocking) 1. **Test hygiene** — `testComputePartitionGroupMetadataUsesKafkaPartitionIds` (both modules) constructs `new KafkaStreamMetadataProvider(...)` without try-with-resources, leaking the consumer. The new `KafkaStreamMetadataProviderTest` does this correctly; worth matching. 2. **Per-partition fetch fan-out** — in the force-from-stream / empty-status path, each hole/new partition opens+closes a metadata provider sequentially. Identical to the prior SPI-default cost (not a regression), but since the provider already holds a connected `_consumer`, a single `beginningOffsets(allTopicPartitions)` could collapse it to one RPC. Reasonable later optimization. 3. **Vanished-partition parity** — the new path emits only ids returned by `fetchPartitionIds()`, so a partition present in the status list but absent from the live stream is dropped (the old code re-emitted it). Only reachable if a topic is deleted+recreated smaller; treating a non-existent partition as EOL is arguably more correct. Awareness only. ### Tests `KafkaStreamMetadataProviderTest` is a true regression test — it asserts the old size-based algorithm yields `[0,1,3,4,5,6,7,7]`, then asserts the fix yields `[0..7]` **with** the distinguishing offsets (fresh smallest `1002` for the recovered hole vs reused `endOffset` `13` for an existing partition). Added to both 3.0 and 4.0. -- 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]
