swaminathanmanish commented on code in PR #13298:
URL: https://github.com/apache/pinot/pull/13298#discussion_r1630651998
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTracker.java:
##########
@@ -174,6 +188,22 @@ private long getIngestionDelayMs(long ingestionTimeMs) {
return agedIngestionDelayMs;
}
+ private long getPartitionOffsetLag(IngestionOffsets offset) {
+ if (offset == null) {
Review Comment:
Curious, can this even be null?
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTracker.java:
##########
@@ -174,6 +188,22 @@ private long getIngestionDelayMs(long ingestionTimeMs) {
return agedIngestionDelayMs;
}
+ private long getPartitionOffsetLag(IngestionOffsets offset) {
+ if (offset == null) {
+ return 0;
+ }
+ StreamPartitionMsgOffset msgOffset = offset._offset;
Review Comment:
This is current offset right?
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTracker.java:
##########
@@ -174,6 +188,22 @@ private long getIngestionDelayMs(long ingestionTimeMs) {
return agedIngestionDelayMs;
}
+ private long getPartitionOffsetLag(IngestionOffsets offset) {
Review Comment:
This comment applies to time based reporting as well and can be taken as
follow up.
Should this belong to individual consumer types instead of common code just
like how getCurrentPartitionLagState is interfaced out, because reporting the
metric for all types here might not make any sense ?
Like you said, only for Kafka, this is relevant.
##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTracker.java:
##########
@@ -174,6 +188,22 @@ private long getIngestionDelayMs(long ingestionTimeMs) {
return agedIngestionDelayMs;
}
+ private long getPartitionOffsetLag(IngestionOffsets offset) {
+ if (offset == null) {
+ return 0;
+ }
+ StreamPartitionMsgOffset msgOffset = offset._offset;
+ StreamPartitionMsgOffset latestOffset = offset._latestOffset;
+
+ // Compute aged delay for current partition
+ // TODO: Support other types of offsets
+ if (!(msgOffset instanceof LongMsgOffset && latestOffset instanceof
LongMsgOffset)) {
+ return 0;
Review Comment:
What are other types that can fall here? Basically we need to be able to
explain this metric.
--
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]
