jackjlli commented on a change in pull request #7368:
URL: https://github.com/apache/pinot/pull/7368#discussion_r718009036
##########
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:
How about introducing an extra map to store the table name and its list
of `MetricKey`s? By doing so, you don't have to traverse the whole map. And you
can also introduce a new class so that the logic of adding/removing metrics can
be encapsulated.
--
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
