Do not count a deliberate consumer stop as a stream consumption error

[krishan1390]

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.

#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

[codecov-commenter]

Codecov Report

❌ Patch coverage is 80.00000% with 3 lines in your changes missing coverage. Please review.
✅ Project coverage is 66.93%. Comparing base (62b4263) to head (cbe2364).
⚠️ Report is 53 commits behind head on master.

Files with missing lines	Patch %	Lines
...a/manager/realtime/RealtimeSegmentDataManager.java	80.00%	1 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19224      +/-   ##
============================================
+ Coverage     65.78%   66.93%   +1.15%     
  Complexity     1423     1423              
============================================
  Files          3441     3452      +11     
  Lines        218303   218550     +247     
  Branches      34728    34744      +16     
============================================
+ Hits         143609   146296    +2687     
+ Misses        63119    60564    -2555     
- Partials      11575    11690     +115     

Flag	Coverage Δ
custom-integration1	?
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	66.93% <80.00%> (+1.15%)	⬆️
lane-a	100.00% <ø> (?)
lane-b	0.00% <ø> (?)
temurin	66.93% <80.00%> (+1.15%)	⬆️
unittests	66.93% <80.00%> (+1.15%)	⬆️
unittests1	57.72% <80.00%> (+0.60%)	⬆️
unittests2	39.02% <33.33%> (+1.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.