xiangfu0 opened a new pull request, #20020:
URL: https://github.com/apache/pinot/pull/20020
## Problem
A real-time table on a **low-traffic transactional Kafka topic** paged
`StreamDataLoss` (P0) three times in 46 minutes, every time on a one-offset gap
(196 → 197, 480 → 481). The committed segment metadata accounted for every
offset and the controller's retention-loss meter stayed at zero: nothing was
lost. The same pattern recurs daily on another production table.
What happens: a transactional producer writes a commit marker after each
transaction. The marker occupies an offset but the Kafka client strips control
batches under both isolation levels, so after consuming record `N` the consumer
is parked at `N + 1` (the marker) and the next record lands at `N + 2`.
`RealtimeSegmentDataManager` keeps requesting `N + 1` because no record was
ever delivered there, so with one record per transaction every new transaction
shows up as `firstOffset > startOffset`. Before #19161 that was reported as
data loss on every poll. The per-table workaround
`stream.kafka.isolation.level: read_committed` suppresses it, but a customer
config must not be able to fire a false P0, so detection has to be right in
code.
Internal tracking: DATA-2966.
## What master already has
#19161 (merged 2026-08-12, after the pages above) replaced the contiguity
assumption with the retention-truncation signal: a gap is loss only when the
broker's **log start offset** (`beginningOffsets`) is above the requested
`startOffset`, i.e. records at/after `startOffset` were actually deleted. That
is the correct signal and this PR keeps it. Deployments pinned before #19161
still produce the false alerts; this PR hardens what sits around that check.
## Changes
**`RealtimeSegmentDataManager.reportDataLoss`** (pinot-core)
- The `STREAM_DATA_LOSS` gauge is still set for every batch that reports
loss (that is the alert signal).
- The `ERROR` log line and the segment error entry are now emitted at most
**once per minute per segment**, and carry the number of batches with loss
since the previous report. Previously both fired on every poll.
**`KafkaPartitionLevelConsumer`** (pinot-kafka-3.0 and pinot-kafka-4.0,
identical)
- After an **empty poll**, `_nextReadOffset` is snapped to
`KafkaConsumer.position()` under both isolation levels, not only
`read_committed`. The old comment claimed a `read_uncommitted` empty poll can
never move the position; it can, past control records (the client returns an
empty poll as soon as its position advanced over a control batch).
`getOffsetOfNextBatch()` of an empty batch now reports where consumption really
resumes. `position()` is answered from the client's local state once a position
is known; only while a partition is awaiting an out-of-range reset does it
block, bounded by the fetch timeout, which is the bound master already had for
`read_committed` empty polls and for the gap lookup.
- Two callers consume that offset. `RealtimeSegmentDataManager` ignores it
for batches with zero unfiltered messages, so its `_currentOffset` and the loss
decision are unchanged. `StatelessRealtimeSegmentWriter` (re-ingestion) adopts
it after every fetch, so under `read_uncommitted` it now steps past a skipped
commit marker on the empty poll instead of on the next data poll; nothing but
control records (and aborted records under `read_committed`, as before) can be
stepped over, and an out-of-range reset lands it where the following poll would
have put it anyway.
- The `WARN` for a failed log-start-offset lookup (added in #19161, where
the reviewer asked whether it needed a rate limiter) is throttled to once per
minute with a suppressed count; suppressed failures still log at `DEBUG`.
- Comments spell out why the gap is compared against the caller's
`startOffset` and why the broker lookup is the cheapest correct signal (below).
No config, API, wire or serialization change. `reportDataLoss` goes from
`private` to package-private (`@VisibleForTesting`).
## Why the detection algorithm stays as in #19161
I looked for a cheaper way to tell a skipped control record from real loss:
- **Control flag**: not exposed. `ConsumerRecord` has no control bit;
`CompletedFetch` drops control batches and only bumps `nextFetchOffset`.
- **Consumer position as the baseline**: a jump in `position()` that no
delivered record explains is either skipped control/aborted batches (benign) or
an `auto.offset.reset` after `OffsetOutOfRange` (real loss). Both look
identical through the client API. Validating a jump needs the same
`beginningOffsets` round trip, so on the ticket's topic it only moves the
lookup from the data poll to the preceding empty poll: same count (one per
transaction), more state. Using the position as the baseline *without*
validating jumps would hide real loss whenever a lagging consumer's reset
completes inside an empty poll. So the check stays keyed on the caller's
`startOffset`, which is stale but conservative.
- **Cost today**: one `ListOffsets` round trip per gap, never on the
contiguous hot path. On a low-traffic transactional topic that is one per
transaction; while catching up, at most one per poll that happens to start at a
marker.
- **Zero-cost design (follow-up, not done here)**: handle
`OffsetOutOfRangeException` in Pinot itself (`auto.offset.reset=none` towards
the client, apply the configured policy by hand) and flag loss exactly when it
fires. That changes how the configured reset policy is applied and deserves its
own review.
## Should `read_committed` become the default on transactional topics?
Recommendation only, not implemented:
- It is a **semantic change**: `read_uncommitted` delivers records of
aborted and in-flight transactions; flipping the default silently drops rows
that existing tables ingest today.
- It is an **availability change**: under `read_committed` the consumer
cannot read past the last stable offset, so an open or hung transaction stalls
the partition until `transaction.timeout.ms` (60 s by default, up to the
broker's 15 min cap).
- There is **no consumer-side way to detect a transactional topic** (the
control batches are stripped), and a topic can become transactional after the
table is created.
- Since #19161 the detection no longer depends on the isolation level, so
there is no correctness reason to flip it.
`stream.kafka.isolation.level=read_committed` stays the right setting for
tables that want exactly-once semantics, not a workaround for the alert.
## `recordsLag` pinned at 1 (documented, not fixed)
On an idle transactional partition the last offset in the log is a commit
marker, so the high watermark (and the last stable offset) is `N + 2` while the
segment's current offset is `N + 1`: nobody can ever consume `N + 1`, so the
lag reads 1. The consumer now reports the real position (`N + 2`) as
`getOffsetOfNextBatch()` of the empty batch, but `RealtimeSegmentDataManager`
only adopts that offset when a batch had messages that were all filtered out.
Adopting it for empty batches would fix the lag and also remove the
stale-`startOffset` round trip above, but it changes offset-advancement
semantics for every stream plugin (Kinesis, Pulsar), so it is left for a
follow-up.
## Tests
Both Kafka plugins (embedded KRaft broker for 3.0, Testcontainers
`apache/kafka:4.0.0` for 4.0):
| Test | Scenario | Expected |
|---|---|---|
|
`DataLossIntegrationTest.testCaughtUpConsumerOnLowTrafficTransactionalTopicIsNotDataLoss`
(new) | one-record transactions; consumer caught up at the marker; next record
one past the requested offset, twice | no loss; empty batch reports position
past the marker |
|
`DataLossIntegrationTest.testTransactionalGapWithRetainedDataIsNotDataLoss`
(existing) | real commit-marker gap, data retained | no loss |
| `DataLossIntegrationTest.testTruncatedStartOffsetIsDataLoss` (existing) |
`deleteRecords` moved the log start past `startOffset` | **loss** |
| `DataLossIntegrationTest.testReadCommittedGapIsNotDataLoss` (existing) |
`read_committed` gap | no loss |
| `DataLossCheckTest` (new, mocked) | contiguous batch / retained gap /
truncated gap / `read_committed` / lookup failure | `beginningOffsets` never /
once / once / never / twice; loss only for the truncated gap |
|
`SeekTest.testRepeatedStartOffsetAfterReadUncommittedEmptyFetchDoesNotSeekAgain`
(new) | empty poll moved the position past a marker | `getOffsetOfNextBatch()`
= position, no re-seek, no loss |
pinot-core: `RealtimeSegmentDataManagerTest.testDataLossReportIsRateLimited`
(new) drives `reportDataLoss` with a fake clock: first detection reported, 10
more within the minute suppressed, the next one after the minute reported with
"batches with loss since last report: 11", batches without loss never reported.
Local runs (JDK 25, Kafka clients 3.9.2 / 4.2.0, `apache/kafka:4.0.0`
container for the 4.0 suite):
| Module | Result |
|---|---|
| `pinot-kafka-3.0`, all `*Kafka*` classes | 68 tests, 0 failures (new: 5
`DataLossCheckTest`, 1 integration scenario, 1 seek case) |
| `pinot-kafka-4.0`, all `*Kafka*` classes | 51 tests, 0 failures (same new
cases; integration suite against a real 4.0 broker) |
| `pinot-kafka-base` | 47 tests, 0 failures |
| `pinot-core` `RealtimeSegmentDataManagerTest` | 39 tests, 0 failures (new:
`testDataLossReportIsRateLimited`) |
`spotless`, `checkstyle` and `license` are clean on the three touched
modules.
## Runbook
Once this is merged the `StreamDataLoss` runbook should get an entry: when
the gauge fires, read the segment error ("Message loss detected ...
batchFirstOffset ... batches with loss since last report"), compare the
partition's `beginningOffsets` with the segment's start offset, and treat
one-offset gaps on transactional topics from versions without #19161 as known
false positives.
--
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]
