[GitHub] [pinot] sajjad-moradi commented on a diff in pull request #10121: Add a tracker for end-to-end consumption delay of events.

[via GitHub]

sajjad-moradi commented on code in PR #10121:
URL: https://github.com/apache/pinot/pull/10121#discussion_r1084449315


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTracker.java:
##########
@@ -283,8 +299,27 @@ public void markPartitionForVerification(int 
partitionGroupId) {
    */
   public long getPartitionIngestionDelayMs(int partitionGroupId) {
     // Not protected as this will only be invoked when metric is installed 
which happens after server ready
-    Long currentMeasure = _partitionToIngestionTimeMsMap.get(partitionGroupId);
-    return getIngestionDelayMs(currentMeasure);
+    IngestionTimestamps currentMeasure = 
_partitionToIngestionTimestampsMap.get(partitionGroupId);
+    if (currentMeasure == null) {
+      return 0;
+    }

Review Comment:
   Could you add some comments here on when we get into the situation of 
currentMeasure being null?



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/IngestionDelayTracker.java:
##########
@@ -283,8 +299,27 @@ public void markPartitionForVerification(int 
partitionGroupId) {
    */
   public long getPartitionIngestionDelayMs(int partitionGroupId) {
     // Not protected as this will only be invoked when metric is installed 
which happens after server ready
-    Long currentMeasure = _partitionToIngestionTimeMsMap.get(partitionGroupId);
-    return getIngestionDelayMs(currentMeasure);
+    IngestionTimestamps currentMeasure = 
_partitionToIngestionTimestampsMap.get(partitionGroupId);
+    if (currentMeasure == null) {
+      return 0;
+    }
+    return getIngestionDelayMs(currentMeasure._ingestionTimeMs);
+  }
+
+  /*
+   * Method to get end to end ingestion delay for a given partition.
+   *
+   * @param partitionGroupId partition for which we are retrieving the delay
+   *
+   * @return End to end ingestion delay in milliseconds for the given 
partition ID.
+   */
+  public long getPartitionEndToEndIngestionDelayMs(int partitionGroupId) {
+    // Not protected as this will only be invoked when metric is installed 
which happens after server ready
+    IngestionTimestamps currentMeasure = 
_partitionToIngestionTimestampsMap.get(partitionGroupId);
+    if (currentMeasure == null) {
+      return 0;
+    }

Review Comment:
   Same here.



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/RealtimeTableDataManager.java:
##########
@@ -240,8 +240,8 @@ protected void doShutdown() {
    * @param ingestionTimeMs Ingestion delay being reported.
    * @param partitionGroupId Partition ID for which delay is being updated.
    */
-  public void updateIngestionDelay(long ingestionTimeMs, int partitionGroupId) 
{
-    _ingestionDelayTracker.updateIngestionDelay(ingestionTimeMs, 
partitionGroupId);
+  public void updateIngestionDelay(long ingestionTimeMs, long creationTimeMs, 
int partitionGroupId) {
+    _ingestionDelayTracker.updateIngestionDelay(ingestionTimeMs, 
creationTimeMs, partitionGroupId);

Review Comment:
   s/creationTimeMs/firstStreamIngestionTimeMs/



-- 
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