[GitHub] [pinot] rino-kadijk commented on a diff in pull request #8280: [WIP][NOT_FOR_REVIEW] Expose consuming segment's offset lag through /consumingSegmentsInfo API

[GitBox <gi...@apache.org>]

rino-kadijk commented on code in PR #8280:
URL: https://github.com/apache/pinot/pull/8280#discussion_r969420702


##########
pinot-plugins/pinot-stream-ingestion/pinot-kafka-2.0/src/main/java/org/apache/pinot/plugin/stream/kafka20/KafkaStreamMetadataProvider.java:
##########
@@ -72,6 +77,26 @@ public StreamPartitionMsgOffset fetchStreamPartitionOffset(OffsetCriteria offset
     }
   }
 
+  @Override
+  public PartitionLagInfo computePartitionLagInfo(ConsumerPartitionStatus currentConsumerStatus,
+      ConsumerPartitionStatus upstreamStatus) {
+    if (currentConsumerStatus != null && upstreamStatus != null) {
+      PartitionLagInfo partitionLagInfo = new PartitionLagInfo();
+      partitionLagInfo.setOffsetLag(computeOffsetLag(currentConsumerStatus.getMsgOffset(),
+          upstreamStatus.getMsgOffset()));
+      return partitionLagInfo;
+    } else {
+      LOGGER.warn("Unable to compute partition lag for {}", _topicPartition.toString());
+      return PartitionLagInfo.empty();
+    }
+  }
+
+  private String computeOffsetLag(StreamPartitionMsgOffset currentOffset,

Review Comment:
   When the lag crosses a threshold it would be nice to see if this is due to a spike in messages or a slow backend. Is it possible to expose currentOffset and latestUpstreamOffset as counters?



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org