You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "krishan1390 (via GitHub)" <gi...@apache.org> on 2026/08/12 10:59:22 UTC
[PR] Do not count a deliberate consumer stop as a stream consumption error [pinot]
krishan1390 opened a new pull request, #19224:
URL: https://github.com/apache/pinot/pull/19224
## Problem
`RealtimeSegmentDataManager.stop()` sets a stop flag and then interrupts the consumer thread. Helix reaches this method on two paths:
- `CONSUMING` to `OFFLINE` and `CONSUMING` to `DROPPED`, through `offloadSegment`. A rebalance, a server shutdown, a table disable, or a table delete causes these transitions.
- `CONSUMING` to `ONLINE`, while the replica still consumes.
A consumer waits inside the stream fetch for most of its life. The interrupt therefore arrives inside `fetchMessages`. Kafka wraps it in an unchecked `InterruptException` with an `InterruptedException` cause.
`consumeLoop` does not recognize this exception. The generic catch block sends it to `handleTransientStreamErrors`, which does three things:
1. It increments the `realtimeConsumptionExceptions` meter, both global and per table.
2. It sleeps for 1 second.
3. It builds a new stream consumer for the segment.
The loop then exits, because the stop flag is set. `doOffload` closes the consumer that step 3 built.
This has three results:
- Each stopped partition adds one count to an error meter, although nothing failed.
- Each stopped partition adds about 1 second, plus one client close and create, to the Helix transition thread.
- If the new consumer fails to build, `streamConsumerCreateExceptions` also increases.
An operation that moves or stops partitions across many tables produces a steady trickle of these counts. An alert on the meter reads that trickle as a stream fault.
apache/pinot#15660 already finds this case. It uses the result only to lower the log level from WARN to DEBUG. The meter increment, the sleep, and the consumer rebuild stay.
`catchupToFinalOffset` sets `_shouldStop` back to false before it consumes again. A condition on `_shouldStop` therefore cannot cover the catch-up path.
## Fix
A new field `_stopping` records the stop. `stop()` sets it, and nothing clears it. It survives the reset inside `catchupToFinalOffset`.
A new method `isDeliberateStopInterrupt` returns true when `_stopping` is set and the cause chain holds an `InterruptedException` or a `ClosedByInterruptException`. The second type covers NIO-based clients.
`consumeLoop` calls this method for each exception before it calls `handleTransientStreamErrors`. On a match it writes one log line and leaves the loop.
`handleTransientStreamErrors` now holds only the code for real errors. The old `_shouldStop` branch is unreachable after this change, so this change removes it.
## Behavior after the change
A deliberate stop adds no count, no sleep, and no new consumer.
- On the offload path, the consumer thread ends.
- On the catch-up path, `catchupToFinalOffset` returns false and the replica downloads the segment. This is the same result as before, without five retries first.
A real stream error keeps the old behavior. It still increments the meter, sleeps, and builds a new consumer.
One log line changes level. A transient error that is not an interrupt now writes WARN during a stop. It wrote DEBUG before.
## Tests
`RealtimeConsumerStopTest` is new. It uses a real table data manager, a real consumer thread, and the real `offloadSegment` call that the Helix state model makes. The test supplies the stream through the `StreamConsumerFactory` extension point. The consumer is then always inside `fetchMessages` when the interrupt arrives.
The class holds three tests:
- An offload of a consuming segment adds no count and builds no new consumer.
- An interrupt on the catch-up path adds no count and stops after one fetch.
- A real stream failure still adds a count and still builds a new consumer.
Without the fix, the first two tests fail. They record 1 and 6 unwanted counts.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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] Do not count a deliberate consumer stop as a stream consumption error [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #19224:
URL: https://github.com/apache/pinot/pull/19224#issuecomment-5266395377
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/19224?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:x: Patch coverage is `80.00000%` with `3 lines` in your changes missing coverage. Please review.
:white_check_mark: Project coverage is 57.72%. Comparing base ([`62b4263`](https://app.codecov.io/gh/apache/pinot/commit/62b42630a0b32998ca39d4c09a0114704b09367a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`cbe2364`](https://app.codecov.io/gh/apache/pinot/commit/cbe23647809bded2b8e070d06b32f3559f26ecf5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 53 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pinot/pull/19224?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...a/manager/realtime/RealtimeSegmentDataManager.java](https://app.codecov.io/gh/apache/pinot/pull/19224?src=pr&el=tree&filepath=pinot-core%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpinot%2Fcore%2Fdata%2Fmanager%2Frealtime%2FRealtimeSegmentDataManager.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29yZS9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29yZS9kYXRhL21hbmFnZXIvcmVhbHRpbWUvUmVhbHRpbWVTZWdtZW50RGF0YU1hbmFnZXIuamF2YQ==) | 80.00% | [1 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pinot/pull/19224?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
> :exclamation: There is a different number of reports uploaded between BASE (62b4263) and HEAD (cbe2364). Click for more details.
>
> <details><summary>HEAD has 16 uploads less than BASE</summary>
>
>| Flag | BASE (62b4263) | HEAD (cbe2364) |
>|------|------|------|
>|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 #19224 +/- ##
============================================
- Coverage 65.78% 57.72% -8.07%
+ Complexity 1423 1 -1422
============================================
Files 3441 2655 -786
Lines 218303 158830 -59473
Branches 34728 26056 -8672
============================================
- Hits 143609 91678 -51931
+ Misses 63119 59370 -3749
+ Partials 11575 7782 -3793
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/19224/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/19224/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/19224/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/19224/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/19224/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/19224/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.72% <80.00%> (-8.07%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/19224/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.72% <80.00%> (-8.07%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/19224/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.72% <80.00%> (-8.07%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/19224/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `57.72% <80.00%> (+0.60%)` | :arrow_up: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/19224/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/19224?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