You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "swaminathanmanish (via GitHub)" <gi...@apache.org> on 2026/08/05 16:07:34 UTC
[PR] Fix false StreamDataLoss on transactional Kafka topics [pinot]
swaminathanmanish opened a new pull request, #19161:
URL: https://github.com/apache/pinot/pull/19161
## 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`):
```java
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:
```java
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.
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19161:
URL: https://github.com/apache/pinot/pull/19161#issuecomment-5195070752
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19161?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 57.11%. Comparing base ([`c5f7bda`](https://app.codecov.io/gh/apache/pinot/commit/c5f7bda6aed8cb3820332e9b4dfc122aec1fd67c?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`0fe66ad`](https://app.codecov.io/gh/apache/pinot/commit/0fe66adda4e0b3e21a203c6cdaf83c3ffbd5d25b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 1 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (c5f7bda) and HEAD (0fe66ad). Click for more details.
>
> <details><summary>HEAD has 16 uploads less than BASE</summary>
>
>| Flag | BASE (c5f7bda) | HEAD (0fe66ad) |
>|------|------|------|
>|unittests|2|1|
>|java-25|5|1|
>|temurin|5|1|
>|unittests2|1|0|
>|integration|3|0|
>|integration1|1|0|
>|integration2|1|0|
>|custom-integration1|1|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #19161 +/- ##
============================================
- Coverage 65.81% 57.11% -8.70%
+ Complexity 1423 1 -1422
============================================
Files 3441 2644 -797
Lines 218320 158666 -59654
Branches 34737 26067 -8670
============================================
- Hits 143677 90630 -53047
+ Misses 63075 60254 -2821
+ Partials 11568 7782 -3786
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19161/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/19161/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/19161/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/19161/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/19161/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-25](https://app.codecov.io/gh/apache/pinot/pull/19161/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.11% <ø> (-8.70%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19161/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.11% <ø> (-8.70%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19161/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.11% <ø> (-8.69%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19161/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.11% <ø> (-0.04%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19161/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/19161?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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on PR #19161:
URL: https://github.com/apache/pinot/pull/19161#issuecomment-5240147211
> Can we add following integration tests if easily possible.
>
> 1. Perform the transaction and see if anything is actually reported or not.
> 2. Delete the offsets and check if the flag has been set to true/ false or a metric is emitted.
Good callout. Added the tests and removed mocks
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "9aman (via GitHub)" <gi...@apache.org>.
9aman commented on PR #19161:
URL: https://github.com/apache/pinot/pull/19161#issuecomment-5239396507
If we want to decouple these calls from the above parameters maybe we can do the following:
1. Fetch which offset will surely be valid 5 minutes from the call being made ie. at `T + 5 minutes. `
2. Use this value for all the calls for the next 5 minutes i.e. cache the above value and use it for comparison
- startOffset should be `>=` offset surely valid at T+5 minutes.
3. If this is false, then make an **actual call** otherwise refresh this at the end of time T + 5.
This way we make a call every 5 minutes or k minutes instead of relying on the batch size etc.
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #19161:
URL: https://github.com/apache/pinot/pull/19161#discussion_r3738658471
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java:
##########
@@ -166,15 +166,36 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
long offsetOfNextBatch = _nextReadOffset;
- // For read_uncommitted (the default), a non-contiguous returned batch implies data
- // loss (records dropped before being read). For read_committed the offset gap is
- // expected because the broker filters aborted transactional records, so we don't flag
- // it as data loss.
- boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset;
+ // A gap between the requested startOffset and the first returned offset does NOT by itself
+ // imply data loss. Transactional producers write commit/abort control records that occupy
+ // offsets but are never delivered to the consumer (even under read_uncommitted), so a
+ // contiguous stream of user records legitimately has offset gaps. Real data loss only
+ // happens when the requested startOffset is below the log's start offset, i.e. the broker
+ // has already deleted (via retention or truncation) records at or after startOffset. For
+ // read_committed we never flag loss because aborted-record gaps are always expected.
+ boolean hasDataLoss = false;
+ if (!_isReadCommitted && firstOffset > startOffset) {
+ hasDataLoss = getLogStartOffset(timeoutMs) > startOffset;
Review Comment:
`poll()` has already been allowed to consume the entire `timeoutMs`, but this second broker call receives a fresh full timeout. On the gap path, `fetchMessages` can therefore block for almost twice the caller's timeout, violating the `PartitionGroupConsumer.fetchMessages` contract that it return within that timeout (`PartitionGroupConsumer.java:43-50`). Track a deadline before polling and pass only the remaining budget here (or skip this best-effort lookup once the budget is exhausted); add a test that verifies the second call cannot extend the overall timeout.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -165,15 +165,36 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
}
long offsetOfNextBatch = _nextReadOffset;
- // For read_uncommitted (the default), a non-contiguous returned batch implies data
- // loss (records dropped before being read). For read_committed the offset gap is
- // expected because the broker filters aborted transactional records, so we don't flag
- // it as data loss.
- boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset;
+ // A gap between the requested startOffset and the first returned offset does NOT by itself
+ // imply data loss. Transactional producers write commit/abort control records that occupy
+ // offsets but are never delivered to the consumer (even under read_uncommitted), so a
+ // contiguous stream of user records legitimately has offset gaps. Real data loss only
+ // happens when the requested startOffset is below the log's start offset, i.e. the broker
+ // has already deleted (via retention or truncation) records at or after startOffset. For
+ // read_committed we never flag loss because aborted-record gaps are always expected.
+ boolean hasDataLoss = false;
+ if (!_isReadCommitted && firstOffset > startOffset) {
+ hasDataLoss = getLogStartOffset(timeoutMs) > startOffset;
Review Comment:
`poll()` has already been allowed to consume the entire `timeoutMs`, but this second broker call receives a fresh full timeout. On the gap path, `fetchMessages` can therefore block for almost twice the caller's timeout, violating the `PartitionGroupConsumer.fetchMessages` contract that it return within that timeout (`PartitionGroupConsumer.java:43-50`). Track a deadline before polling and pass only the remaining budget here (or skip this best-effort lookup once the budget is exhausted); add a test that verifies the second call cannot extend the overall timeout.
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "9aman (via GitHub)" <gi...@apache.org>.
9aman commented on code in PR #19161:
URL: https://github.com/apache/pinot/pull/19161#discussion_r3748645011
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -165,15 +165,36 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
}
long offsetOfNextBatch = _nextReadOffset;
- // For read_uncommitted (the default), a non-contiguous returned batch implies data
- // loss (records dropped before being read). For read_committed the offset gap is
- // expected because the broker filters aborted transactional records, so we don't flag
- // it as data loss.
- boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset;
+ // A gap between the requested startOffset and the first returned offset does NOT by itself
+ // imply data loss. Transactional producers write commit/abort control records that occupy
+ // offsets but are never delivered to the consumer (even under read_uncommitted), so a
+ // contiguous stream of user records legitimately has offset gaps. Real data loss only
+ // happens when the requested startOffset is below the log's start offset, i.e. the broker
+ // has already deleted (via retention or truncation) records at or after startOffset. For
+ // read_committed we never flag loss because aborted-record gaps are always expected.
+ boolean hasDataLoss = false;
+ if (!_isReadCommitted && firstOffset > startOffset) {
+ hasDataLoss = getLogStartOffset(timeoutMs) > startOffset;
+ }
return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata,
hasDataLoss, batchSizeInBytes);
}
+ /// Returns the log start (earliest available) offset for the partition, bounded by the same
+ /// timeout as [#poll]. Returns [Long#MIN_VALUE] when it cannot be determined so the caller
+ /// treats an offset gap as expected (no data loss) rather than raising a false positive.
+ private long getLogStartOffset(int timeoutMs) {
+ try {
+ return _consumer.beginningOffsets(List.of(_topicPartition), Duration.ofMillis(timeoutMs))
+ .getOrDefault(_topicPartition, Long.MIN_VALUE);
+ } catch (Exception e) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Failed to read log start offset for {}", _topicPartition, e);
Review Comment:
Should we at least log the exception ?
It will be hard to debug in case of a genuine data loss. I think the segment metadata will only point to a data loss in case the above call fails.
Is this failure expected often ? If not, shall we log a WARN.
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "9aman (via GitHub)" <gi...@apache.org>.
9aman commented on PR #19161:
URL: https://github.com/apache/pinot/pull/19161#issuecomment-5238813815
I tried to verify this under the following scenarios/ configs.
1. Non-pauseless:
- It should work as expected.
2. Pauseless:
- Normal ingestion: should work as above.
- Re-ingestion during RVM: should work fine flagging only genuine cases
IIUC, we want to separate the presence of data from Pinot's ability to ingest it.
1. Data loss alerting to ensure source has data.
2. NoRealtimeIngestion, Ingestion lag: Inability to ingest data.
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish merged PR #19161:
URL: https://github.com/apache/pinot/pull/19161
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on code in PR #19161:
URL: https://github.com/apache/pinot/pull/19161#discussion_r3765239080
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -165,15 +165,36 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
}
long offsetOfNextBatch = _nextReadOffset;
- // For read_uncommitted (the default), a non-contiguous returned batch implies data
- // loss (records dropped before being read). For read_committed the offset gap is
- // expected because the broker filters aborted transactional records, so we don't flag
- // it as data loss.
- boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset;
+ // A gap between the requested startOffset and the first returned offset does NOT by itself
+ // imply data loss. Transactional producers write commit/abort control records that occupy
+ // offsets but are never delivered to the consumer (even under read_uncommitted), so a
+ // contiguous stream of user records legitimately has offset gaps. Real data loss only
+ // happens when the requested startOffset is below the log's start offset, i.e. the broker
+ // has already deleted (via retention or truncation) records at or after startOffset. For
+ // read_committed we never flag loss because aborted-record gaps are always expected.
+ boolean hasDataLoss = false;
+ if (!_isReadCommitted && firstOffset > startOffset) {
+ hasDataLoss = getLogStartOffset(timeoutMs) > startOffset;
Review Comment:
Not an issue, and the added complexity isn't worth it. This call is per fetch batch and not per record.
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaPartitionLevelConsumer.java:
##########
@@ -166,15 +166,36 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
long offsetOfNextBatch = _nextReadOffset;
- // For read_uncommitted (the default), a non-contiguous returned batch implies data
- // loss (records dropped before being read). For read_committed the offset gap is
- // expected because the broker filters aborted transactional records, so we don't flag
- // it as data loss.
- boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset;
+ // A gap between the requested startOffset and the first returned offset does NOT by itself
+ // imply data loss. Transactional producers write commit/abort control records that occupy
+ // offsets but are never delivered to the consumer (even under read_uncommitted), so a
+ // contiguous stream of user records legitimately has offset gaps. Real data loss only
+ // happens when the requested startOffset is below the log's start offset, i.e. the broker
+ // has already deleted (via retention or truncation) records at or after startOffset. For
+ // read_committed we never flag loss because aborted-record gaps are always expected.
+ boolean hasDataLoss = false;
+ if (!_isReadCommitted && firstOffset > startOffset) {
+ hasDataLoss = getLogStartOffset(timeoutMs) > startOffset;
Review Comment:
Not an issue, and the added complexity isn't worth it. This call is per fetch batch and not per record.
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "9aman (via GitHub)" <gi...@apache.org>.
9aman commented on PR #19161:
URL: https://github.com/apache/pinot/pull/19161#issuecomment-5238975720
Can we add following integration tests if easily possible.
1. Perform the transaction and see if anything is actually reported or not.
2. Delete the offsets and check if the flag has been set to true/ false or a metric is emitted.
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "9aman (via GitHub)" <gi...@apache.org>.
9aman commented on PR #19161:
URL: https://github.com/apache/pinot/pull/19161#issuecomment-5239337795
```
┌────────────────────────────┬──────────────┬──────────────────┬─────────────────────┐
│ Regime │ Records/poll │ Gap probability │ RPC : messages │
├────────────────────────────┼──────────────┼──────────────────┼─────────────────────┤
│ Catch-up, full polls │ 500 │ ≈ marker density │ ≪ 1/500 │
├────────────────────────────┼──────────────┼──────────────────┼─────────────────────┤
│ Tail, high throughput │ large │ ≈ 1 │ ~1/records-per-poll │
├────────────────────────────┼──────────────┼──────────────────┼─────────────────────┤
│ Tail, low rate, 1-rec txns │ 1 │ ≈ 1 │ 1:1 │
└────────────────────────────┴──────────────┴──────────────────┴─────────────────────┘
```
this is a good estimate of the additional calls that we will end up making.
500 is the default max records per poll. There are restrictions on the bytes fetched per call as well.
Number of additional calls is inversely proportional to
1. Number of records fetched in one call.
2. Number of messages within a transaction.
cc: @KKcorps @noob-se7en
--
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] Fix false StreamDataLoss on transactional Kafka topics [pinot]
Posted by "swaminathanmanish (via GitHub)" <gi...@apache.org>.
swaminathanmanish commented on code in PR #19161:
URL: https://github.com/apache/pinot/pull/19161#discussion_r3749422082
##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaPartitionLevelConsumer.java:
##########
@@ -165,15 +165,36 @@ public synchronized KafkaMessageBatch fetchMessages(StreamPartitionMsgOffset sta
}
}
long offsetOfNextBatch = _nextReadOffset;
- // For read_uncommitted (the default), a non-contiguous returned batch implies data
- // loss (records dropped before being read). For read_committed the offset gap is
- // expected because the broker filters aborted transactional records, so we don't flag
- // it as data loss.
- boolean hasDataLoss = !_isReadCommitted && firstOffset > startOffset;
+ // A gap between the requested startOffset and the first returned offset does NOT by itself
+ // imply data loss. Transactional producers write commit/abort control records that occupy
+ // offsets but are never delivered to the consumer (even under read_uncommitted), so a
+ // contiguous stream of user records legitimately has offset gaps. Real data loss only
+ // happens when the requested startOffset is below the log's start offset, i.e. the broker
+ // has already deleted (via retention or truncation) records at or after startOffset. For
+ // read_committed we never flag loss because aborted-record gaps are always expected.
+ boolean hasDataLoss = false;
+ if (!_isReadCommitted && firstOffset > startOffset) {
+ hasDataLoss = getLogStartOffset(timeoutMs) > startOffset;
+ }
return new KafkaMessageBatch(filteredRecords, records.size(), offsetOfNextBatch, firstOffset, lastMessageMetadata,
hasDataLoss, batchSizeInBytes);
}
+ /// Returns the log start (earliest available) offset for the partition, bounded by the same
+ /// timeout as [#poll]. Returns [Long#MIN_VALUE] when it cannot be determined so the caller
+ /// treats an offset gap as expected (no data loss) rather than raising a false positive.
+ private long getLogStartOffset(int timeoutMs) {
+ try {
+ return _consumer.beginningOffsets(List.of(_topicPartition), Duration.ofMillis(timeoutMs))
+ .getOrDefault(_topicPartition, Long.MIN_VALUE);
+ } catch (Exception e) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("Failed to read log start offset for {}", _topicPartition, e);
Review Comment:
Made this a warn msg. Hope it does not flood the log or we need a rate limiter.
--
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