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: [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]
