Exposing consumer's record lag in /consumingSegmentsInfo

[navina]

Re-work of the PR : #8280

Changes in this PR:

• /consumingSegmentsInfo API was moved into PinotRealtimeTableResource .
  • This returns additional per-partition map for upstream latest record offset and partition lag (when calculated)
• Introduces new API in StreamMetadataProvider#getCurrentPartitionLagState
  • This PR only adds support for Kafka.
  • Any stream connector that hasn't implemented this method will return NOT_CALCULATED for these new lag status
  • Additionally, connectors can extend PartitionLagState interface to return more custom lag definitions

Things to be addressed in a follow-up PR:

• Define a time-based lag status
• Add partition lag info for kinesis and pulsar stream

[navina]

@KKcorps / @npawar : extending metrics to plugin is taking a while. So, let's get this in while I continue working on that.

[codecov-commenter]

Codecov Report

Merging #9515 (daa4728) into master (0a442b9) will decrease coverage by 6.25%.
The diff coverage is 15.27%.

@@             Coverage Diff              @@
##             master    #9515      +/-   ##
============================================
- Coverage     70.03%   63.78%   -6.26%     
+ Complexity     5288     4952     -336     
============================================
  Files          1937     1889      -48     
  Lines        103498   101377    -2121     
  Branches      15715    15453     -262     
============================================
- Hits          72487    64660    -7827     
- Misses        25907    31986    +6079     
+ Partials       5104     4731     -373     

Flag	Coverage Δ
integration1	?
integration2	?
unittests1	67.31% <0.00%> (-0.10%)	⬇️
unittests2	15.58% <15.27%> (-0.03%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
.../common/restlet/resources/SegmentConsumerInfo.java	0.00% <0.00%> (ø)
...ller/api/resources/PinotRealtimeTableResource.java	0.00% <0.00%> (ø)
...ler/api/resources/PinotSegmentRestletResource.java	10.31% <ø> (-28.13%)	⬇️
...manager/realtime/HLRealtimeSegmentDataManager.java	0.00% <0.00%> (-82.88%)	⬇️
...manager/realtime/LLRealtimeSegmentDataManager.java	52.19% <0.00%> (-17.68%)	⬇️
...a/manager/realtime/RealtimeSegmentDataManager.java	50.00% <ø> (-50.00%)	⬇️
...ugin/stream/kafka20/KafkaConsumerPartitionLag.java	0.00% <0.00%> (ø)
...in/stream/kafka20/KafkaStreamMetadataProvider.java	50.90% <0.00%> (-17.28%)	⬇️
...pache/pinot/spi/stream/ConsumerPartitionState.java	0.00% <0.00%> (ø)
...org/apache/pinot/spi/stream/PartitionLagState.java	0.00% <0.00%> (ø)
... and 488 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[npawar]

minor comments only, looking good!

[npawar]

lgtm!
we need to mark this backward incompat because of the partitionToOffset map being removed? should we keep it around and deprecate it first?

[navina]

lgtm! we need to mark this backward incompat because of the partitionToOffset map being removed? should we keep it around and deprecate it first?

Strictly speaking, this is considered backward incompat. But how do you "deprecate" an API response alone?

[npawar]

lgtm! we need to mark this backward incompat because of the partitionToOffset map being removed? should we keep it around and deprecate it first?

Strictly speaking, this is considered backward incompat. But how do you "deprecate" an API response alone?

all we can do is just mark it deprecated in the java class for devs, and callout in release notes for users in next release. and in release after that we can remove it. we could also add it as API notes so it appears in swagger?