Fix false StreamDataLoss on transactional Kafka topics

[swaminathanmanish]

Problem

Real-time tables ingesting from transactional Kafka topics raise false
StreamDataLoss alerts. KafkaPartitionLevelConsumer flagged data loss whenever the
first returned record offset was greater than the requested startOffset (under
read_uncommitted):

boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset;

Transactional producers write commit/abort control records that consume offsets in
the log but are never delivered to the consumer. A perfectly healthy, contiguous stream
of user records therefore has offset gaps, which this check misread as data loss.

Fix

Only report data loss when the requested startOffset is actually below the broker's
log start offset (Consumer.beginningOffsets) — i.e. records at/after startOffset
were genuinely deleted by retention or truncation:

boolean hasDataLoss = false;
if (!_isReadCommitted && firstOffset > startOffset) {
  hasDataLoss = getLogStartOffset(timeoutMs) > startOffset;
}

• The extra broker round-trip only happens on the rare gap path, never on the contiguous
  hot path or under read_committed.
• If the log start offset can't be determined, we default to no data loss so a
  transient broker hiccup can't manufacture a false alert.
• This is the resolution anticipated by the pre-existing // TODO: fetch earliest offset from topic comment.

Applied to both pinot-kafka-3.0 and pinot-kafka-4.0.

Tests

New KafkaPartitionLevelConsumerDataLossTest in each module (Mockito harness mirroring
KafkaPartitionLevelConsumerSeekTest), 5 cases each:

Scenario	Expected
Transactional gap, data still retained (logStart <= startOffset)	no loss
startOffset below log start (real truncation)	loss
Contiguous batch (asserts beginningOffsets is never called)	no loss
read_committed gap (never called)	no loss
beginningOffsets lookup fails	no loss

Verified the tests fail against the old code (the two false-positive cases flip to
true) and pass with the fix. checkstyle and spotless clean on both modules.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 77.77778% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 65.79%. Comparing base (c5f7bda) to head (0fe66ad).
⚠️ Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
...in/stream/kafka30/KafkaPartitionLevelConsumer.java	77.77%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19161      +/-   ##
============================================
- Coverage     65.81%   65.79%   -0.02%     
  Complexity     1423     1423              
============================================
  Files          3441     3441              
  Lines        218320   218328       +8     
  Branches      34737    34739       +2     
============================================
- Hits         143677   143650      -27     
- Misses        63075    63100      +25     
- Partials      11568    11578      +10     

Flag	Coverage Δ
custom-integration1	100.00% <ø> (ø)
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	65.79% <77.77%> (-0.02%)	⬇️
temurin	65.79% <77.77%> (-0.02%)	⬇️
unittests	65.79% <77.77%> (-0.02%)	⬇️
unittests1	57.11% <ø> (-0.04%)	⬇️
unittests2	38.03% <77.77%> (+<0.01%)	⬆️

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 false StreamDataLoss alerts caused by Kafka transactional control-record offset gaps.

Changes:

• Confirms loss using Kafka’s log start offset.
• Adds five regression scenarios for Kafka 3.0 and 4.0.
• Preserves fast paths for contiguous and read_committed consumption.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

File	Description
pinot-kafka-3.0/.../KafkaPartitionLevelConsumer.java	Refines data-loss detection.
pinot-kafka-3.0/.../KafkaPartitionLevelConsumerDataLossTest.java	Adds regression coverage.
pinot-kafka-4.0/.../KafkaPartitionLevelConsumer.java	Refines data-loss detection.
pinot-kafka-4.0/.../KafkaPartitionLevelConsumerDataLossTest.java	Adds regression coverage.