Vamsi-klu opened a new pull request, #19267:
URL: https://github.com/apache/pinot/pull/19267
## Problem
Under Kafka `isolation.level=read_committed`, a partition whose log tail is
only transaction control records never advances `_currentOffset`.
`FreshnessBasedConsumptionStatusChecker` then waits forever on a stable
one-offset gap, and the server never reports GOOD. Rolling restarts of
low-volume exactly-once tables loop.
The reporter case: last visible record at 18895, Pinot sitting at 18896,
Kafka latest at 18897 (the commit marker). `isOffsetCaughtUp` requires `current
>= latest`, so the partition never catches up.
`KafkaPartitionLevelConsumer` already has the right answer. On an empty
`read_committed` poll it snaps `_nextReadOffset` to `KafkaConsumer.position()`
and returns that as `offsetOfNextBatch`. `RealtimeSegmentDataManager` throws it
away because the advance is gated on `getUnfilteredMessageCount() > 0`, and a
control-record-only tail yields zero records.
## What I did
I widened that one guard so the consume loop also advances when the
next-batch offset is already ahead of the current offset. I also documented the
`MessageBatch.getOffsetOfNextBatch()` contract: return the requested start
offset unchanged when the partition had nothing to hand back; only return a
larger offset when the stream itself has moved past offsets this batch will
never deliver.
## How I did it
In `RealtimeSegmentDataManager`:
```
} else if (messageBatch.getUnfilteredMessageCount() > 0
|| messageBatch.getOffsetOfNextBatch().compareTo(_currentOffset) > 0) {
```
`getOffsetOfNextBatch()` is generic SPI. Every other stream returns the
start offset unchanged on a genuinely empty batch, so Kinesis and Pulsar
behavior does not change. A partition that is genuinely lagging also returns
the unchanged start offset and stays not-caught-up.
This is not a `latest - 1` heuristic. I did not change
`DEFAULT_REALTIME_FRESHNESS_IDLE_TIMEOUT_MS`, `endOffsets` for
`read_uncommitted`, or `KafkaStreamMetadataProvider.fetchLatestStreamOffset`.
## Impact
Idle EOS tables under `read_committed` can finish catch-up after a restart.
Servers that hung in `RealtimeConsumptionCatchupServiceStatusCallback` can turn
GOOD. Genuinely lagging partitions, `read_uncommitted`, Kinesis, and Pulsar are
unchanged.
## Testing
`RealtimeSegmentDataManagerTest#testEmptyBatchWithAdvancedNextOffsetMovesCurrentOffset`
fails before this change (stays at the old offset) and passes after. The
sibling `testEmptyBatchWithUnchangedNextOffsetDoesNotInventOffset` asserts an
empty batch whose next offset did not move invents no offset.
`KafkaPartitionLevelConsumerTest#testReadCommittedEmptyPollSnapsOffsetOfNextBatchToPosition`
(kafka 3.0 and 4.0) pins the `position()` snap the guard depends on.
`ExactlyOnceKafkaRealtimeClusterIntegrationTest` now waits for consuming
offsets to reach latest after the final commit marker.
`FreshnessBasedConsumptionStatusCheckerTest#controlRecordTailDoesNotTreatLatestMinusOneAsCaughtUp`
is a behavior lock, not a regression test: it passes either way and exists so
nobody "fixes" readiness with a `latest - 1` heuristic.
```
./mvnw -pl pinot-core -am
-Dtest=RealtimeSegmentDataManagerTest#testEmptyBatchWithAdvancedNextOffsetMovesCurrentOffset,RealtimeSegmentDataManagerTest#testEmptyBatchWithUnchangedNextOffsetDoesNotInventOffset
test
./mvnw -pl pinot-server -am
-Dtest=FreshnessBasedConsumptionStatusCheckerTest#controlRecordTailDoesNotTreatLatestMinusOneAsCaughtUp
test
./mvnw -pl pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0
-Dtest=KafkaPartitionLevelConsumerTest#testReadCommittedEmptyPollSnapsOffsetOfNextBatchToPosition
test
./mvnw -pl pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0
-Dtest=KafkaPartitionLevelConsumerTest#testReadCommittedEmptyPollSnapsOffsetOfNextBatchToPosition
test
./mvnw -pl pinot-integration-tests -am
-Dtest=ExactlyOnceKafkaRealtimeClusterIntegrationTest
-Dsurefire.failIfNoSpecifiedTests=false test
```
Fixes #17962
Made with [Cursor](https://cursor.com)
--
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]
