This is an automated email from the ASF dual-hosted git repository.
snlee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 09bb91c Fix merge delay metric when there is empty bucket (#7761)
09bb91c is described below
commit 09bb91cecc8cff459a9d7814a7fe61aa67f2a393
Author: Seunghyun Lee <[email protected]>
AuthorDate: Tue Nov 16 16:57:44 2021 -0800
Fix merge delay metric when there is empty bucket (#7761)
when we have the empty buckets (no more data push
for the table). However, we still have some edge
cases not covered. We found the issue where the
table has some empty buckets and new fresh data
that is not eligible for merge due to buffer time.
In this case, we still fall back to the original
algorithm and report the wrong delay. This PR
addresses such case.
---
.../plugin/minion/tasks/mergerollup/MergeRollupTaskGenerator.java | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/mergerollup/MergeRollupTaskGenerator.java
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/mergerollup/MergeRollupTaskGenerator.java
index 3b59a6f..ece07be 100644
---
a/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/mergerollup/MergeRollupTaskGenerator.java
+++
b/pinot-plugins/pinot-minion-tasks/pinot-minion-builtin-tasks/src/main/java/org/apache/pinot/plugin/minion/tasks/mergerollup/MergeRollupTaskGenerator.java
@@ -254,7 +254,11 @@ public class MergeRollupTaskGenerator implements
PinotTaskGenerator {
// but the metrics are not available until the controller schedules a
valid task
long maxEndTimeMs = Long.MIN_VALUE;
for (SegmentZKMetadata preSelectedSegment : preSelectedSegments) {
- maxEndTimeMs = Math.max(maxEndTimeMs,
preSelectedSegment.getEndTimeMs());
+ long currentEndTimeMs = preSelectedSegment.getEndTimeMs();
+ // Compute maxEndTimeMs among segments that are valid for merge
+ if (currentEndTimeMs < System.currentTimeMillis() - bufferMs) {
+ maxEndTimeMs = Math.max(maxEndTimeMs, currentEndTimeMs);
+ }
}
createOrUpdateDelayMetrics(offlineTableName, mergeLevel, null,
watermarkMs, maxEndTimeMs,
bufferMs, bucketMs);
@@ -548,7 +552,7 @@ public class MergeRollupTaskGenerator implements
PinotTaskGenerator {
private long getMergeRollupTaskDelayInNumTimeBuckets(long watermarkMs, long
maxEndTimeMsOfCurrentLevel,
long bufferTimeMs, long bucketTimeMs) {
- if (watermarkMs == -1) {
+ if (watermarkMs == -1 || maxEndTimeMsOfCurrentLevel == Long.MIN_VALUE) {
return 0;
}
return (Math.min(System.currentTimeMillis() - bufferTimeMs,
maxEndTimeMsOfCurrentLevel) - watermarkMs)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
