navina commented on code in PR #9713:
URL: https://github.com/apache/pinot/pull/9713#discussion_r1191553503
##########
pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProvider.java:
##########
@@ -189,6 +192,24 @@ private boolean
consumedEndOfShard(StreamPartitionMsgOffset startCheckpoint,
}
@Override
- public void close() {
+ public Map<String, PartitionLagState> getCurrentPartitionLagState(
+ Map<String, ConsumerPartitionState> currentPartitionStateMap) {
+ Map<String, PartitionLagState> perPartitionLag = new HashMap<>();
+ for (Map.Entry<String, ConsumerPartitionState> entry:
currentPartitionStateMap.entrySet()) {
+ ConsumerPartitionState partitionState = entry.getValue();
+ // Compute record-availability
+ String recordAvailabilityLag = "UNKNOWN";
+ RowMetadata lastProcessedMessageMetadata =
partitionState.getLastProcessedRowMetadata();
+ if (lastProcessedMessageMetadata != null &&
partitionState.getLastProcessedTimeMs() > 0) {
+ long availabilityLag = partitionState.getLastProcessedTimeMs()
+ - lastProcessedMessageMetadata.getRecordIngestionTimeMs();
+ recordAvailabilityLag = String.valueOf(availabilityLag);
+ }
+ perPartitionLag.put(entry.getKey(), new
KinesisConsumerPartitionLag(recordAvailabilityLag));
Review Comment:
Yeah. So, the metrics that is actually used for monitoring is
`ingestionDelayMs`. This PR provides an API to compute the point in time
ingestion lag (This API predates the ingestion lag) and doesn't actually create
any metric. So, I think we don't need to worry about false alerts here.
We do enable the `ingestionDelayLagMs` metric at a partition level. I am not
entirely sure how to remove a metric emitted by the realtime engine, when the
connector manages the set of active partitions.. Need to look at the code
again. but this problem should is out of scope for this PR.
--
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-unsubscr...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
