[pinot-server] Add consuming-segment decoder observability

[anuragrai16]

Expose the concrete StreamMessageDecoder class of a consuming realtime segment for debugging decode issues:

• New CONSUMING_SEGMENT_DECODER server gauge, tagged by table/topic/ partition/decoder class, emitted alongside LLC_PARTITION_CONSUMING and removed on segment offload.

• Persist the decoder class in segment metadata (custom.decoder.class) and surface decoderClassName via the consuming-segment debug APIs.

• Add a JMX->Prometheus scrape rule for the new gauge.

Backward compatible: SPI addition is a nullable default method, the DTO field is nullable with ignore-unknown, segment metadata is additive, and the scrape rule is anchored to avoid shadowing existing rules.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 76.31579% with 9 lines in your changes missing coverage. Please review.
✅ Project coverage is 67.74%. Comparing base (63b3b1f) to head (deba872).
⚠️ Report is 4 commits behind head on master.

Files with missing lines	Patch %	Lines
...a/manager/realtime/RealtimeSegmentDataManager.java	61.11%	2 Missing and 5 partials ⚠️
...ache/pinot/server/api/resources/DebugResource.java	0.00%	1 Missing ⚠️
...che/pinot/server/api/resources/TablesResource.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #19529      +/-   ##
============================================
+ Coverage     67.70%   67.74%   +0.03%     
  Complexity     1450     1450              
============================================
  Files          3490     3491       +1     
  Lines        224998   225043      +45     
  Branches      35523    35530       +7     
============================================
+ Hits         152335   152453     +118     
+ Misses        60618    60556      -62     
+ Partials      12045    12034      -11     

Flag	Coverage Δ
integration	100.00% <ø> (ø)
integration1	100.00% <ø> (ø)
integration2	?
java-25	67.74% <76.31%> (+0.03%)	⬆️
lane-a	100.00% <ø> (ø)
lane-b	0.00% <ø> (ø)
temurin	67.74% <76.31%> (+0.03%)	⬆️
unittests	67.74% <76.31%> (+0.03%)	⬆️
unittests1	57.85% <51.51%> (+0.05%)	⬆️
unittests2	39.49% <55.26%> (+0.02%)	⬆️

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.

[mcvsubbu]

If I understand right, you are emitting a metric (of value 1) with the current decoder class name. Is this not an overkill? Can you just log the decoder class name? Even log at some regular cadence if needed?

I saw a proposal float by for pinot internal table. Perhaps you could add a row/column in that table?

[anuragrai16]

If I understand right, you are emitting a metric (of value 1) with the current decoder class name. Is this not an overkill? Can you just log the decoder class name? Even log at some regular cadence if needed?

I saw a proposal float by for pinot internal table. Perhaps you could add a row/column in that table?

Hi @mcvsubbu - Thanks for checking the diff. We're trying to migrate 1000s of table decoders on live production tables, each of which consumes from Kafka topics upto 256 partitions, while Pinot has no observability into what decoder was used for a mutable segment, or being used for a consuming segment. A log is ephemeral and not a reliable way to solve this problem.

At the cost/overkill part, we are just emitting one additional gauge per consuming segment, using the existing setValueOfGauge register-once path, no new gauge objects created per tick, just an AtomicLong.set(1). The overhead is negligible compared to what each consuming segment already emits.

As for the proposal for Pinot internal table, it is a larger effort that's still in design. This gauge is a minimal, backward-compatible addition (constant 1 emitted alongside the already-existing LLC_PARTITION_CONSUMING gauge on the same code path, same register-once/AtomicLong mechanism) that solves the immediate operational need without waiting on new infrastructure. Let me know your thoughts, I feel this metadata belongs in the segment as a first class citizen.

[mcvsubbu]

If I understand right, you are emitting a metric (of value 1) with the current decoder class name. Is this not an overkill? Can you just log the decoder class name? Even log at some regular cadence if needed?
I saw a proposal float by for pinot internal table. Perhaps you could add a row/column in that table?

Hi @mcvsubbu - Thanks for checking the diff. We're trying to migrate 1000s of table decoders on live production tables, each of which consumes from Kafka topics upto 256 partitions, while Pinot has no observability into what decoder was used for a mutable segment, or being used for a consuming segment. A log is ephemeral and not a reliable way to solve this problem.

At the cost/overkill part, we are just emitting one additional gauge per consuming segment, using the existing setValueOfGauge register-once path, no new gauge objects created per tick, just an AtomicLong.set(1). The overhead is negligible compared to what each consuming segment already emits.

As for the proposal for Pinot internal table, it is a larger effort that's still in design. This gauge is a minimal, backward-compatible addition (constant 1 emitted alongside the already-existing LLC_PARTITION_CONSUMING gauge on the same code path, same register-once/AtomicLong mechanism) that solves the immediate operational need without waiting on new infrastructure. Let me know your thoughts, I feel this metadata belongs in the segment as a first class citizen.

Every metric you add is an overhead. Especially ones that emit a 0 or 1. We can probably remove the consuming metric as well, because we now have a metric that shows the delay in consumption.
Moreover, the metric will be useless after migration (perhaps), when things are in stable state.

That being said, here are two alternatives:

• Get the information from the table config in zookeeper.
• Create an API on the server to get this information