You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "xiangfu0 (via GitHub)" <gi...@apache.org> on 2026/10/24 01:21:56 UTC
[PR] Adopt the stream position of an empty batch in RealtimeSegmentDataManager [pinot]
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Adopt the stream position of an empty batch in RealtimeSegmentDataManager [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4366204270
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -656,12 +734,16 @@ private boolean processStreamEvents(MessageBatch messageBatch, long idlePipeSlee
throw new RuntimeException("Realtime segment full");
}
- // Decode message
StreamMessage<?> streamMessage = messageBatch.getStreamMessage(index);
- StreamDataDecoderResult decodedRow = _streamDataDecoder.decode(streamMessage);
StreamMessageMetadata metadata = streamMessage.getMetadata();
StreamPartitionMsgOffset offset = metadata.getOffset();
StreamPartitionMsgOffset nextOffset = metadata.getNextOffset();
+ if (reachedFinalOffsetBeforeMessage(messageBatch, nextOffset)) {
Review Comment:
Non-blocking: when this settles on `_finalOffset` mid-batch, the `getUnfilteredMessageCount() > 0` branch below still calls `updateIngestionMetrics(lastMetadata)` with the *last* (unconsumed, skipped) message's timestamp, so freshness briefly reflects a row we didn't index. This matches the existing endCriteria-mid-batch exit, so it's not a regression — just worth confirming it's intended during catch-up.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Adopt the stream position of an empty batch in RealtimeSegmentDataManager [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4366260608
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -656,12 +734,16 @@ private boolean processStreamEvents(MessageBatch messageBatch, long idlePipeSlee
throw new RuntimeException("Realtime segment full");
}
- // Decode message
StreamMessage<?> streamMessage = messageBatch.getStreamMessage(index);
- StreamDataDecoderResult decodedRow = _streamDataDecoder.decode(streamMessage);
StreamMessageMetadata metadata = streamMessage.getMetadata();
StreamPartitionMsgOffset offset = metadata.getOffset();
StreamPartitionMsgOffset nextOffset = metadata.getNextOffset();
+ if (reachedFinalOffsetBeforeMessage(messageBatch, nextOffset)) {
Review Comment:
Intended, and the same as the existing mid-batch exits. When `endCriteriaReached()` fires inside a batch today (row or time limit), the loop also breaks before the remaining messages and this block still reports `getLastMessageMetadata()`, so the semantics of the post-loop metrics have always been "the latest message the consumer fetched", not "the latest message this segment indexed". Settling on `_finalOffset` adds one more way to stop inside a batch and inherits that.
Why I left it alone for the catch-up case: the ingestion delay tracker is keyed by partition, not by segment, and ages the delay at read time (`now - ingestionTimeMs`). The skipped message is by construction the first message of the successor segment (it starts at exactly `_finalOffset`), so the only effect is that, between this stop and the successor reporting its first batch, the partition metric is based on a message one position ahead of what this replica indexed: offset lag understated by one and the delay by the timestamp gap between the two messages. The handoff interval itself dominates that error since the tracker keeps aging the last reported timestamp meanwhile. Making the post-loop block use the last processed message instead would be a reasonable tightening, but it changes the existing premature-exit path too, so I would rather do it separately than inside this change.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Adopt the stream position of an empty batch in RealtimeSegmentDataManager [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20024:
URL: https://github.com/apache/pinot/pull/20024#issuecomment-6378600415
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/20024?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 0.00%. Comparing base ([`910f9d5`](https://app.codecov.io/gh/apache/pinot/commit/910f9d5e4942010621c7d791557e0b4c4ffddcfd?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`5236f98`](https://app.codecov.io/gh/apache/pinot/commit/5236f98cacb04f62ca0d828ce9adf0062c404413?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 148 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (910f9d5) and HEAD (5236f98). Click for more details.
>
> <details><summary>HEAD has 18 uploads less than BASE</summary>
>
>| Flag | BASE (910f9d5) | HEAD (5236f98) |
>|------|------|------|
>|unittests1|1|0|
>|unittests|2|0|
>|java-25|6|2|
>|temurin|6|2|
>|lane-a|2|1|
>|integration|4|2|
>|integration1|2|0|
>|lane-b|2|1|
>|unittests2|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20024 +/- ##
=============================================
- Coverage 68.63% 0.00% -68.64%
=============================================
Files 3526 3 -3523
Lines 230143 6 -230137
Branches 36562 0 -36562
=============================================
- Hits 157963 0 -157963
+ Misses 59837 6 -59831
+ Partials 12343 0 -12343
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [integration](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-100.00%)` | :arrow_down: |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-68.64%)` | :arrow_down: |
| [lane-a](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-100.00%)` | :arrow_down: |
| [lane-b](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (ø)` | |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `0.00% <ø> (-68.64%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/20024/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/pinot/pull/20024?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
Re: [PR] Adopt the stream position of an empty batch in RealtimeSegmentDataManager [pinot]
Posted by "xiangfu0 (via GitHub)" <gi...@apache.org>.
xiangfu0 commented on code in PR #20024:
URL: https://github.com/apache/pinot/pull/20024#discussion_r4369051914
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +616,65 @@ protected boolean consumeLoop()
return true;
}
+ // TODO Issue 5359 Need to find a way to bump metrics without getting actual offset value.
+ private void updateHighestStreamOffsetConsumedGauge() {
+ if (_currentOffset instanceof LongMsgOffset) {
+ // TODO: only LongMsgOffset supplies long offset value.
+ _serverMetrics.setValueOfTableGauge(_clientId, ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED,
+ ((LongMsgOffset) _currentOffset).getOffset());
+ }
+ }
+
+ /// Adopts an empty batch's advanced consumer position for INITIAL consumption or ONLINE catch-up.
+ /// Transaction markers and aborted records can move a stream's position without delivering rows, so adoption
+ /// keeps position and lag current. But a position, including a batch with `hasDataLoss() == false`, does not prove
+ /// the skipped interval contained no rows: validation may be unavailable, or the topic may have been compacted.
+ /// Remember that uncertainty for local-reuse CRC validation even when the new position is below the final offset.
+ ///
+ /// Protocol CATCHING_UP can subsequently become the committer, so it must not synthesize progress from an empty
+ /// batch. ONLINE catch-up can cap the position at the committed final offset, because any local reuse is guarded
+ /// by the committed CRC. A reported loss is not capped; ordinary overshoot recovery still applies.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ if (_state == State.CATCHING_UP) {
+ return false;
+ }
+ StreamPartitionMsgOffset nextBatchOffset = messageBatch.getOffsetOfNextBatch();
+ if (nextBatchOffset == null || nextBatchOffset.compareTo(_currentOffset) <= 0) {
+ return false;
+ }
+ if (_state == State.CONSUMING_TO_ONLINE && _finalOffset != null && !messageBatch.hasDataLoss()
+ && nextBatchOffset.compareTo(_finalOffset) > 0) {
+ nextBatchOffset = _finalOffset;
+ }
+ if (_segmentLogger.isDebugEnabled()) {
+ _segmentLogger.debug("Empty batch moved the stream position. Advancing from {} to {}", _currentOffset,
+ nextBatchOffset);
+ }
+ _hasUnverifiedOffsetAdvance = true;
+ _currentOffset = nextBatchOffset;
+ return true;
+ }
+
+ /// ONLINE catch-up can stop before a message ending past the committed final offset. This only establishes
+ /// offset equality, not row equality, so record the unverified advance before publishing the final offset.
+ /// Protocol CATCHING_UP must retain ordinary record advancement and overshoot handling instead of manufacturing
+ /// a target match that could make this replica the committer. Reported loss also retains overshoot recovery.
+ ///
+ /// @return `true` if consumption must stop before the message
+ private boolean reachedFinalOffsetBeforeMessage(MessageBatch messageBatch, StreamPartitionMsgOffset nextOffset) {
+ if (_state != State.CONSUMING_TO_ONLINE || _finalOffset == null || messageBatch.hasDataLoss()
+ || _currentOffset.compareTo(_finalOffset) >= 0 || nextOffset.compareTo(_finalOffset) <= 0) {
Review Comment:
Nit: when a real message straddles _finalOffset (offset < final < nextOffset), this settles at _finalOffset without indexing that message, which is only safe because _hasUnverifiedOffsetAdvance then forces the CRC-guarded download. That coupling is correct but subtle; a one-line note here pointing at the _hasUnverifiedOffsetAdvance guard would help the next reader.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeSegmentDataManager.java:
##########
@@ -605,6 +616,65 @@ protected boolean consumeLoop()
return true;
}
+ // TODO Issue 5359 Need to find a way to bump metrics without getting actual offset value.
+ private void updateHighestStreamOffsetConsumedGauge() {
+ if (_currentOffset instanceof LongMsgOffset) {
+ // TODO: only LongMsgOffset supplies long offset value.
+ _serverMetrics.setValueOfTableGauge(_clientId, ServerGauge.HIGHEST_STREAM_OFFSET_CONSUMED,
+ ((LongMsgOffset) _currentOffset).getOffset());
+ }
+ }
+
+ /// Adopts an empty batch's advanced consumer position for INITIAL consumption or ONLINE catch-up.
+ /// Transaction markers and aborted records can move a stream's position without delivering rows, so adoption
+ /// keeps position and lag current. But a position, including a batch with `hasDataLoss() == false`, does not prove
+ /// the skipped interval contained no rows: validation may be unavailable, or the topic may have been compacted.
+ /// Remember that uncertainty for local-reuse CRC validation even when the new position is below the final offset.
+ ///
+ /// Protocol CATCHING_UP can subsequently become the committer, so it must not synthesize progress from an empty
+ /// batch. ONLINE catch-up can cap the position at the committed final offset, because any local reuse is guarded
+ /// by the committed CRC. A reported loss is not capped; ordinary overshoot recovery still applies.
+ ///
+ /// @return `true` if `_currentOffset` was advanced
+ private boolean adoptNextBatchOffsetOfEmptyBatch(MessageBatch messageBatch) {
+ if (_state == State.CATCHING_UP) {
Review Comment:
Excluding CATCHING_UP from empty-batch adoption is the right safety call. One liveness edge worth a confirming thought: if a committer committed at a marker-only offset, a CATCHING_UP follower can't reach _finalOffset by consuming records and will poll empty batches (100ms sleep each) until the Helix ONLINE transition arrives, which then recovers via the CONSUMING_TO_ONLINE adoption + CRC-guarded download. Self-recovering and low-CPU, so fine — just flagging in case you want a comment noting the ONLINE path is the intended escape.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org