Retry consuming-segment stream init on the consumer thread

[Vamsi-klu]

Problem

A retryable Kafka/DNS failure during consuming-segment init still runs on the Helix state-transition thread. That path uses Kafka's short 5x2s retry, then marks the replica OFFLINE after about 10s. Healthy replicas keep consuming. The partition stays under-replicated until the next natural flush.

#17062 already retries mid-consume recreate on the consumer thread with CONSUMER_RECREATE_RETRY_POLICY (about 8 minutes). First create did not use that path.

#19083 tried a controller-side auto force-commit. Review feedback was that one bad server must not force healthy replicas to commit, and that the remaining hole is this init path, not another default-off controller flag.

What I did

Move first makeStreamConsumer("Starting") from the Helix constructor onto PartitionConsumer.run(), after the ready-wait and consumer semaphore. First create now uses the same CONSUMER_RECREATE_RETRY_POLICY as mid-consume recreate (10 attempts, 1s, factor 2).

No force-commit. No new controller config. #17754 (controller.realtime.segment.partialOfflineReplicaRepairEnabled, default false) is unchanged. All-OFFLINE recreate is unchanged.

How

• Helix constructor still builds the decoder, transform pipeline, mutable segment, and partition metadata provider. It does not open the stream consumer.
• Exhausted init still takes the existing postStopConsumedMsg / ERROR / OFFLINE path.
• stop() for CONSUMING to ONLINE keeps a just-created consumer so catchup can use it. Offload during in-flight create closes the uninstalled client.
• Catchup requires both the consumer semaphore and a live consumer. Otherwise the replica downloads.
• Close leaves the reference in place so COMMIT/CATCH_UP can still checkpoint a closed instance (default is identity).

Impact

A transient Kafka/DNS blip on one server retries on that server's consumer thread instead of going OFFLINE after about 10s. Other replicas are not force-committed. Brokers do not start preferring a replica that is hours behind.

Helix state-transition threads are not blocked for the long retry window.

Not in this PR

createPartitionMetadataProvider("Starting") still runs on the Helix constructor. For Kafka that opens a metadata-provider client on the short 5x2s path. A retryable DNS failure there can still OFFLINE a replica after about 10s. Fetch fallbacks already exist for offset lookup; provider construction itself can still throw.

Testing

RealtimeSegmentDataManagerTest: 33 tests, 0 failures.

• ctor does not create a stream consumer
• first create uses CONSUMER_RECREATE_RETRY_POLICY
• exhaustion posts stop-consumed and goes ERROR
• offload with never-created consumer succeeds
• stop during init keeps the consumer for catchup
• offload during init closes the created client
• ONLINE with semaphore but no consumer downloads

./mvnw -pl pinot-core -am -Dtest=RealtimeSegmentDataManagerTest -Dsurefire.failIfNoSpecifiedTests=false test

Please tag bug and testing. No release-notes (no new config).

Related

• Automatically repair a problematic consuming segment #15897
• Error in Realtime Ingestion Marks Segment as Offline in Ideal State #11314
• Improves consumer recreation retry policy #17062
• resume partially offline replicas in controller validation #17754
• Alternative to Auto force-commit partial OFFLINE consuming segments #19083 (no force-commit, no new controller flag)

cc @noob-se7en

Was generative AI tooling used to co-author this PR?

• Yes

[codecov-commenter]

Codecov Report

❌ Patch coverage is 62.79070% with 16 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.46%. Comparing base (d40dceb) to head (ed1761b).

Files with missing lines	Patch %	Lines
...a/manager/realtime/RealtimeSegmentDataManager.java	62.79%	10 Missing and 6 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19370      +/-   ##
============================================
+ Coverage     67.44%   67.46%   +0.02%     
  Complexity     1430     1430              
============================================
  Files          3485     3485              
  Lines        223874   223899      +25     
  Branches      35300    35308       +8     
============================================
+ Hits         150987   151051      +64     
+ Misses        60890    60852      -38     
+ Partials      11997    11996       -1     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	0.00% <ø> (ø)
java-25	67.46% <62.79%> (+0.02%)	⬆️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	67.46% <62.79%> (+0.02%)	⬆️
unittests	67.46% <62.79%> (+0.02%)	⬆️
unittests1	57.56% <62.79%> (+0.01%)	⬆️
unittests2	39.30% <32.55%> (+<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.