xiangfu0 opened a new pull request, #20024: URL: https://github.com/apache/pinot/pull/20024
Stacked on #20020 (the first commit here is that PR's commit; this PR's own change is the second commit). Will rebase once #20020 merges. ## Problem On a transactional Kafka topic every committed transaction leaves a control record (the commit marker) that occupies an offset but is never delivered to the client. After record N a caught-up consumer sits at the marker offset N+1, and `RealtimeSegmentDataManager._currentOffset` stays there because it never moves on an empty batch. Two symptoms remain after #19161 and #20020: 1. `recordsLag` from `KafkaStreamMetadataProvider.getCurrentPartitionLagState` reads 1 while idle (high watermark N+2, current offset N+1), and the committed end offset points at an offset nobody can consume. 2. Every new transaction shows up to `KafkaPartitionLevelConsumer.fetchMessages` as `firstOffset > startOffset` and pays one `beginningOffsets` round trip to rule out data loss. ## Change **`RealtimeSegmentDataManager.consumeLoop`**: when a batch delivered no messages and `getOffsetOfNextBatch()` is ahead of `_currentOffset`, adopt it (`adoptNextBatchOffsetOfEmptyBatch`). Null-checked and `compareTo > 0` guarded. The adoption counts as stream activity for the idle timer and updates the highest-consumed-offset gauge: the partition did produce something, so the consumer is not stuck; ingestion delay and freshness are untouched since no row was consumed (`processStreamEvents` already reports an empty batch as caught up). The rate-limited data-loss report now also prints the batch's next offset, since the loss can be reported on an empty batch. **Catch-up with an end offset inside a skipped region.** A committer can now stop at marker+1, an offset no message's next offset produces, so a replica catching up to it has to be able to settle there rather than overshoot (`endCriteriaReached` throws "Past max offset" on an overshoot in CATCHING_UP, and CONSUMING_TO_ONLINE turns a local build into a download). Two rules, both only in CATCHING_UP / CONSUMING_TO_ONLINE and only when the batch does not report data loss: - An empty batch's adopted offset is capped at `_finalOffset`. - In `processStreamEvents`, a message whose next offset lies past `_finalOffset` is not indexed; `_currentOffset` settles on `_finalOffset` and the message is left to the next segment (what `StatelessRealtimeSegmentWriter` already does). Every offset in between was skipped by the stream, so the replica holds the same rows as the committer. When the batch does report data loss the skipped offsets held rows this replica never saw, so neither rule applies: the overshoot surfaces as before and the committed segment is downloaded, instead of settling on an end offset the replica cannot vouch for. **`MessageBatch.getOffsetOfNextBatch` (SPI Javadoc only)**: documents the contract this relies on: for an empty batch the requested start offset, unless the stream moved past offsets it will never deliver; never past an undelivered message. **`KafkaPartitionLevelConsumer`** (kafka-3.0 and kafka-4.0, byte-identical modulo package): tracks the client position after every poll (`_positionAfterLastPoll`, `>= _nextReadOffset`). - A move the client made while delivering records (the gap sits inside one answered fetch response, before or after the delivered records) can never be an `auto.offset.reset`, since a reset re-positions the client before a fetch is answered. So the data-poll gap check compares the first delivered offset with the position the poll started from instead of the caller's `startOffset`, and the next empty poll does not re-validate a move the client made while delivering the previous records. - A move that first shows up on an empty poll is the one case a reset can hide in once the caller adopts the reported position: the next data poll would then start exactly where it was requested and show no gap. That move is validated against the broker's log start offset (`!readCommitted && logStart > positionBeforePoll`) and reported as `hasDataLoss` on the empty batch. If that lookup fails, the move stays unaccounted for and the next poll validates it again, so a broker hiccup cannot hide a reset for good (the WARN for a failed lookup is already rate-limited). ## Per-plugin audit of `getOffsetOfNextBatch()` on an empty batch Adopting the offset must never skip undelivered records. Every implementation under `pinot-plugins/pinot-stream-ingestion` plus the test fake was checked: | Implementation | Empty batch returns | Adoption effect | |---|---|---| | `KafkaMessageBatch` (3.0/4.0 consumer) | the client position after the poll (`position()`), which only moves past control records, aborted records (read_committed) or compacted-away offsets, or after an out-of-range reset (now reported as loss) | moves past skipped offsets only | | `KinesisMessageBatch` (`KinesisConsumer`) | the requested `startOffset`, on every empty path (rate-limit timeout, request-limiter timeout, null shard iterator, empty `GetRecords`) | no-op (`compareTo == 0`) | | `PulsarMessageBatch` (`PulsarPartitionLevelConsumer`) | the requested `startOffset` | no-op | | `FakeStreamMessageBatch` (pinot-core test fake) | the requested `startOffset` once the data is exhausted | no-op | | `StatelessRealtimeSegmentWriter` (reference, not a batch) | already adopts `getOffsetOfNextBatch()` unconditionally after every batch | same semantics as this change | ## What the `beginningOffsets` round trip does now - Data poll of a caught-up consumer: gone. The consumer parks past the marker, so the next record is delivered at exactly the offset the poll started from. - Marker in the same fetch response as its record (transaction fully committed before the consumer asks, or any lagging consumer): gone entirely. The record poll sees the client already past the marker; the following empty poll finds no new move. - Marker in a later fetch than its record (the typical caught-up timing, and what the real-broker test observes from the second transaction on: the broker answers the consumer's parked fetch as soon as the record is appended, the marker lands milliseconds later): the empty poll that skips the marker still pays one lookup, because a position move during an empty poll cannot be told apart from an `auto.offset.reset` without asking the broker. Before this change the same transaction paid the lookup on the data poll instead, so the count per transaction is unchanged in this case and lower in the others; it is never higher. Removing that remaining lookup soundly needs the client to stop resetting silently (`auto.offset.reset=none` towards the client, with Pinot emulating the configured policy on `OffsetOutOfRangeException`). That changes how the consumer handles out-of-range offsets and is left as a follow-up. ## Rolling upgrade A committed end offset can now sit one past a commit marker (the high watermark) instead of one past the last record. A server that does not have this change cannot reach such an offset by consuming: it parks one before it. Mixed-version behaviour during a rolling upgrade, when a new server wins the commit and an old replica is told to catch up to its end offset: - Idle partition: the old replica stays in CATCHING_UP until the ONLINE transition, then spends the 31-second CONSUMING_TO_ONLINE budget and downloads the committed segment. - A record arrives first: the old replica consumes it, overshoots the end offset, its consumer thread ends with "Past max offset" and the segment is downloaded on the ONLINE transition. Both end in the committed segment being downloaded, so no data is lost or duplicated, at the cost of a download and an error log line per affected segment on old replicas while the upgrade is in flight. Old servers committing (end offset one past the last record) are unaffected: new replicas never move past a catch-up offset. Pauseless re-ingestion of such a segment (`StatelessRealtimeSegmentWriter`) on a server without #20020's empty-batch position snap keeps polling while the topic is idle, since its empty batch never moves; it completes as soon as the next record arrives. ## Tests - `RealtimeSegmentDataManagerTest`: drives the real `consumeLoop()` with scripted batches: empty batches ahead, equal, null and behind (offsets requested from the consumer and the final current offset asserted); while catching up, an empty batch ahead of the final offset is capped there and no further fetch is issued, an empty batch reporting data loss is not capped and surfaces "Past max offset", a message ending past the final offset is not indexed and the offset settles on the final offset, and the same message with data loss is indexed and surfaces the overshoot. - `KafkaPartitionLevelConsumerDataLossCheckTest` (both plugins, mocked client): a move inside a record poll needs no lookup; a gap skipped inside an earlier record poll is not re-checked when the next data poll starts inside it (the false positive the old baseline produced once the log start moved onto the marker); a move during an empty poll is validated once and the following data poll needs none; a reset during an empty poll is reported as loss on the empty batch and not again on the next data poll; a failed validation is retried on the next poll and then reports the loss; no move means no lookup; read_committed never looks anything up. - `KafkaPartitionLevelConsumerDataLossIntegrationTest` (both plugins, real broker): replays the data manager loop over three single-record transactions and asserts the adopted offset reaches the high watermark with `recordsLag` 0 (vs 1 at the marker) after every transaction, that the data poll never looks anything up, and that the empty poll pays one `beginningOffsets` call only when the marker arrived in a later fetch than its record (the client position after the data poll, observed through a `Consumer` proxy around the real `KafkaConsumer`, tells which case happened; the first transaction, fetched with nothing in flight, is always the same-fetch case). The late-marker scenario (record visible before the transaction commits) pins down exactly one lookup, no loss, offset at the high watermark, and no lookup on the idle polls after it. An emptied topic (log start = high watermark) asserts a reset inside an empty poll is reported as loss on the empty batch. The truncated-topic test now accepts the loss being reported on the empty batch that observed the reset when the reset and the fetch do not complete in the same poll. Local evidence (JDK 25): `RealtimeSegmentDataManagerTest` 41/41; pinot-kafka-3.0 76/76 (embedded broker); pinot-kafka-4.0 59/59 (real 4.0 broker via Testcontainers); pinot-kafka-base 47/47; `spotless:apply` and `checkstyle:check` clean on pinot-core, pinot-kafka-3.0 and pinot-kafka-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]
