Advance consume offset past read_committed control-record tails

[Vamsi-klu]

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

[codecov-commenter]

Codecov Report

❌ Patch coverage is 50.00000% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 66.96%. Comparing base (388bc64) to head (562d87f).
⚠️ Report is 5 commits behind head on master.

Files with missing lines	Patch %	Lines
...a/manager/realtime/RealtimeSegmentDataManager.java	50.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19267      +/-   ##
============================================
- Coverage     66.97%   66.96%   -0.02%     
  Complexity     1423     1423              
============================================
  Files          3453     3453              
  Lines        218949   218961      +12     
  Branches      34805    34806       +1     
============================================
- Hits         146648   146622      -26     
- Misses        60585    60624      +39     
+ Partials      11716    11715       -1     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	66.96% <50.00%> (-0.02%)	⬇️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	66.96% <50.00%> (-0.02%)	⬇️
unittests	66.95% <50.00%> (-0.02%)	⬇️
unittests1	57.72% <50.00%> (-0.01%)	⬇️
unittests2	39.02% <0.00%> (-0.02%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[Copilot]

Pull request overview

Fixes #17962 by advancing realtime consumption beyond invisible Kafka transactional control-record tails.

Changes:

• Advances _currentOffset for authoritative empty-batch offsets.
• Documents the MessageBatch offset contract.
• Adds unit, Kafka, readiness, and integration coverage.

Reviewed changes

Copilot reviewed 7 out of 7 changed files in this pull request and generated no comments.

Show a summary per file

File	Description
pinot-spi/.../MessageBatch.java	Clarifies next-batch offset semantics.
pinot-core/.../RealtimeSegmentDataManager.java	Advances offsets for empty control-record tails.
pinot-core/.../RealtimeSegmentDataManagerTest.java	Tests advanced and unchanged empty offsets.
pinot-server/.../FreshnessBasedConsumptionStatusCheckerTest.java	Prevents unsafe latest - 1 readiness logic.
pinot-kafka-3.0/.../KafkaPartitionLevelConsumerTest.java	Verifies Kafka 3 position snapping.
pinot-kafka-4.0/.../KafkaPartitionLevelConsumerTest.java	Verifies Kafka 4 position snapping.
pinot-integration-tests/.../ExactlyOnceKafkaRealtimeClusterIntegrationTest.java	Validates end-to-end control-tail catch-up.

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.