[GitHub] [pinot] jackjlli commented on a change in pull request #7368: Add mergeRollupTask delay metrics

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

jackjlli commented on a change in pull request #7368:
URL: https://github.com/apache/pinot/pull/7368#discussion_r717978921



##########
File path: pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/MergeRollupMinionClusterIntegrationTest.java
##########
@@ -306,6 +306,8 @@ public void testSingleLevelConcat()
     // Check total tasks
     assertEquals(numTasks, 5);
 
+    Assert.assertTrue(_controllerStarter.getControllerMetrics().containsGauge("myTable1_OFFLINE.100days"));

Review comment:
       nit: you don't need the `Assert.` prefix if `assertTrue` shows up as static import, similarly to other assertion methods in the test.

##########
File path: pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/mergerollup/MergeRollupTaskGenerator.java
##########
@@ -463,4 +541,57 @@ private long getWatermarkMs(long minStartTimeMs, long bucketMs, String mergeLeve
 
     return pinotTaskConfigs;
   }
+
+  private long getMergeRollupTaskDelayInNumTimeBuckets(long watermarkMs, long bufferTimeMs, long bucketTimeMs) {
+    if (bufferTimeMs == 0) {
+      return 0;
+    }
+    return (long) Math.floor((System.currentTimeMillis() - watermarkMs - bufferTimeMs) / (double) bucketTimeMs);
+  }
+
+  private void setWatermarkMs(String offlineTableName, String mergeLevel, long watermarkMs, long bufferTimeMs,
+      long bucketTimeMs) {
+    LOGGER.info(
+        "Setting wartermark for table: {} and mergeLevel: {} is {} (watermarkMs={}, bufferTimeMs={}, bucketTimeMs={})",
+        offlineTableName, mergeLevel, getMergeRollupTaskDelayInNumTimeBuckets(watermarkMs, bufferTimeMs, bucketTimeMs),
+        watermarkMs, bucketTimeMs, bucketTimeMs);
+
+    ControllerMetrics controllerMetrics = _clusterInfoAccessor.getControllerMetrics();
+    PinotMetricsRegistry metricsRegistry = controllerMetrics.getMetricsRegistry();
+    if (metricsRegistry == null) {
+      return;
+    }
+
+    // Update gauge value that indicates the delay in terms of the number of time buckets.
+    MetricKey metricKey = new MetricKey(offlineTableName, mergeLevel);
+    _mergeRollupWatermarks.compute(metricKey, (k, v) -> {
+      if (v == null) {
+        controllerMetrics.addCallbackGaugeIfNeeded(getMetricNameForTaskDelay(metricKey.toString()),
+            (() -> getMergeRollupTaskDelayInNumTimeBuckets(_mergeRollupWatermarks.getOrDefault(k, 0L), bufferTimeMs,
+                bucketTimeMs)));
+      }
+      return watermarkMs;
+    });
+  }
+
+  private void resetWatermarkMs(String tableNameWithType) {
+    ControllerMetrics controllerMetrics = _clusterInfoAccessor.getControllerMetrics();
+    PinotMetricsRegistry metricsRegistry = controllerMetrics.getMetricsRegistry();
+    if (metricsRegistry == null) {
+      return;
+    }
+    Iterator<MetricKey> it = _mergeRollupWatermarks.keySet().iterator();
+    while (it.hasNext()) {
+      MetricKey metricKey = it.next();
+      if (metricKey.getTableNameWithType().equals(tableNameWithType)) {
+        it.remove();
+        controllerMetrics.removeCallbackGauge(getMetricNameForTaskDelay(metricKey.toString()));

Review comment:
       Can we simply return if this if statement is qualified? By doing so, you don't have to traverse the whole map.




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