You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "abhishekbafna (via GitHub)" <gi...@apache.org> on 2025/11/03 07:11:05 UTC
[PR] Fix Minion task metrics collection gaps [pinot]
abhishekbafna opened a new pull request, #17128:
URL: https://github.com/apache/pinot/pull/17128
# Fix Minion Task Metrics Collection Gaps
## Problem Statement
The Minion subtask metrics collection had two scenarios where metrics were not being reported accurately:
1. **Previously in-progress tasks that completed between collection cycles**: Tasks that were in-progress in one collection run but completed before the next run were not included in metrics, causing their final states (especially errors) to be missed.
2. **Short-lived tasks that started and completed between collection cycles**: Tasks that started and completed entirely between two 15-minute collection cycles were never captured in metrics.
## Solution
### 1. Track Previously In-Progress Tasks
- Added `_previousInProgressTasks` map to maintain state of tasks that were in-progress in the previous execution cycle
- Compare previous and current in-progress tasks to detect completed tasks
- Include completed tasks in current cycle's metrics collection
### 2. Detect Short-Lived Tasks
- Added `getTasksStartedAfter()` method in `PinotHelixTaskResourceManager` that uses `WorkflowContext.getJobStartTimes()` to efficiently find tasks that started after a timestamp
- Filter out tasks already tracked to avoid duplicates
- Include short-lived tasks in metrics collection
### 3. Dynamic Timestamp Initialization
- Replaced static `_initializationTimestamp` (set at object construction) with dynamic calculation using `currentTime - frequencyMs` on first execution
- Ensures accurate timestamps when controller becomes leader for the first time or after leader changes
- Each task type maintains its own timestamp for independent lifecycle tracking
## Key Changes
### Files Modified
1. **TaskMetricsEmitter.java**
- Added `_previousInProgressTasks` map to track tasks across cycles
- Changed from global `_previousExecutionTimestamp` to per-task-type `_previousExecutionTimestamps` map
- Added dynamic timestamp initialization using `_taskMetricsEmitterFrequencyMs`
- Implemented logic to detect and include completed tasks
- Implemented logic to detect and include short-lived tasks
- Added cleanup for removed task types
2. **PinotHelixTaskResourceManager.java**
- Added `getTasksStartedAfter(taskType, afterTimestampMs)` method
- Uses `WorkflowContext.getJobStartTimes()` for efficient batch retrieval
- Returns tasks that started after the specified timestamp
3. **TaskMetricsEmitterTest.java**
- Added comprehensive test for previously in-progress tasks that completed between runs
- Added comprehensive test for short-lived tasks that started and completed between runs
- Updated all comments to use contextual naming instead of "gap" terminology
- Updated variable names for clarity (`taskCompletedBetweenRuns`, `taskShortLived`)
## Testing
- ✅ All 11 tests pass (including 2 new tests for the gaps)
- ✅ No checkstyle violations
- ✅ Comprehensive test coverage for both scenarios
## Benefits
1. **Complete Metrics**: No metrics are missed for tasks that complete quickly or between cycles
2. **Accurate Error Reporting**: Errors in fast-failing tasks are now captured
3. **Leader Change Resilience**: Dynamic timestamp initialization handles controller leadership changes correctly
4. **Per-Task-Type Independence**: Each task type maintains its own tracking, preventing cross-contamination
## Backward Compatibility
✅ Fully backward compatible - existing functionality is preserved, only extended to cover the missing scenarios.
--
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
Re: [PR] Fix Minion task metrics collection gaps [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17128:
URL: https://github.com/apache/pinot/pull/17128#discussion_r2485515234
##########
pinot-controller/src/test/java/org/apache/pinot/controller/helix/core/minion/TaskMetricsEmitterTest.java:
##########
@@ -336,4 +336,141 @@ public void removeOldTaskTypeAddNewTaskType() {
oneSingleTaskTypeWithTwoTables();
taskType2WithOneTable();
}
+
+ /**
+ * Test for previously in-progress tasks that completed between runs:
+ * Tasks that were in-progress in the previous run but completed before the current run
+ * should still have their metrics reported in the current run.
+ *
+ * Scenario:
+ * - Run 1: Task "taskCompletedBetweenRuns" is in-progress with 1 error subtask
+ * - Run 2: Task "taskCompletedBetweenRuns" has completed and is no longer in getTasksInProgress()
+ *
+ * Expected: Metrics for "taskCompletedBetweenRuns" should still be emitted in Run 2 by detecting it via
+ * _previousInProgressTasks tracking. The emitter maintains state of tasks that were in-progress
+ * in the previous execution cycle and includes completed tasks in the current cycle's metrics.
+ */
+ @Test
+ public void testReportsPreviouslyInProgressTasksThatCompletedBetweenRuns() {
+ String taskType = "SegmentGenerationAndPushTask";
+ String taskName = "taskCompletedBetweenRuns";
+ String tableName = "testTable_OFFLINE";
+
+ Mockito.when(_pinotHelixTaskResourceManager.getTaskTypes()).thenReturn(ImmutableSet.of(taskType));
+
+ // Run 1: Task is in-progress with 1 error subtask
+ Mockito.when(_pinotHelixTaskResourceManager.getTasksInProgress(taskType))
+ .thenReturn(ImmutableSet.of(taskName));
+
+ // Ensure getTasksStartedAfter returns empty for this test (not relevant for this scenario)
+ Mockito.when(_pinotHelixTaskResourceManager.getTasksStartedAfter(
+ Mockito.eq(taskType), Mockito.anyLong()))
+ .thenReturn(ImmutableSet.of());
+
+ PinotHelixTaskResourceManager.TaskCount taskCount = new PinotHelixTaskResourceManager.TaskCount();
+ taskCount.addTaskState(TaskPartitionState.TASK_ERROR);
+ Mockito.when(_pinotHelixTaskResourceManager.getTableTaskCount(taskName))
+ .thenReturn(Map.of(tableName, taskCount));
+
+ _taskMetricsEmitter.runTask(null);
+
+ // Verify metrics were emitted in Run 1
+ PinotMetricsRegistry metricsRegistry = _controllerMetrics.getMetricsRegistry();
+ Assert.assertEquals(((YammerSettableGauge<?>) metricsRegistry.allMetrics().get(
+ new YammerMetricName(ControllerMetrics.class,
+ "pinot.controller.numMinionSubtasksError." + taskType))
+ .getMetric()).value(), 1L);
Review Comment:
This metric retrieval pattern with multiple type casts is repeated identically at lines 379-382, 395-398, 438-441, and 471-474. Extract this into a test helper method like `getMetricValue(String metricName, String taskType)` to improve maintainability and reduce duplication.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskMetricsEmitter.java:
##########
@@ -90,20 +106,71 @@ protected final void runTask(Properties periodicTaskProperties) {
TaskCount taskTypeAccumulatedCount = new TaskCount();
Map<String, TaskCount> tableAccumulatedCount = new HashMap<>();
try {
- Set<String> tasksInProgress = _helixTaskResourceManager.getTasksInProgress(taskType);
- final int numRunningTasks = tasksInProgress.size();
- for (String task : tasksInProgress) {
- Map<String, TaskCount> tableTaskCount = _helixTaskResourceManager.getTableTaskCount(task);
- tableTaskCount.forEach((tableNameWithType, taskCount) -> {
- taskTypeAccumulatedCount.accumulate(taskCount);
- tableAccumulatedCount.compute(tableNameWithType, (name, count) -> {
- if (count == null) {
- count = new TaskCount();
- }
- count.accumulate(taskCount);
- return count;
+ // Capture the current execution timestamp for this task type collection cycle
+ long currentExecutionTimestamp = System.currentTimeMillis();
+ // For task types encountered for the first time, use current time minus one collection cycle
+ // as the initial timestamp to ensure we don't miss tasks that started recently
+ long previousExecutionTimestamp = _previousExecutionTimestamps.computeIfAbsent(taskType,
+ k -> currentExecutionTimestamp - _taskMetricsEmitterFrequencyMs);
+
+ // Get currently in-progress tasks
+ Set<String> currentInProgressTasks = _helixTaskResourceManager.getTasksInProgress(taskType);
+
+ // Get tasks that were in-progress during the previous collection cycle
+ Set<String> previouslyInProgressTasks =
+ _previousInProgressTasks.getOrDefault(taskType, Collections.emptySet());
+
+ // Start with currently in-progress tasks
+ Set<String> tasksToReport = new HashSet<>(currentInProgressTasks);
+
+ // Include tasks that were in-progress previously but are no longer in-progress
+ // These tasks completed between collection cycles and need their final metrics reported
+ for (String taskName : previouslyInProgressTasks) {
+ if (!currentInProgressTasks.contains(taskName)) {
+ LOGGER.debug("Including task {} that completed between collection cycles for taskType: {}",
+ taskName, taskType);
+ tasksToReport.add(taskName);
+ }
+ }
+
+ // Include tasks that started and completed between collection cycles
+ Set<String> tasksStartedAfter = _helixTaskResourceManager.getTasksStartedAfter(
+ taskType, previousExecutionTimestamp);
+
+ // Filter out tasks we already know about to avoid duplicates
+ Set<String> shortLivedTasks = new HashSet<>();
+ for (String taskName : tasksStartedAfter) {
+ if (!currentInProgressTasks.contains(taskName) && !previouslyInProgressTasks.contains(taskName)) {
+ shortLivedTasks.add(taskName);
+ }
+ }
+
+ if (!shortLivedTasks.isEmpty()) {
+ LOGGER.debug("Including {} short-lived tasks that started and completed between cycles for taskType: {}",
+ shortLivedTasks.size(), taskType);
+ tasksToReport.addAll(shortLivedTasks);
+ }
+
+ final int numRunningTasks = currentInProgressTasks.size();
+
+ // Process all tasks that need metrics reported
+ for (String task : tasksToReport) {
+ try {
+ Map<String, TaskCount> tableTaskCount = _helixTaskResourceManager.getTableTaskCount(task);
+ tableTaskCount.forEach((tableNameWithType, taskCount) -> {
+ taskTypeAccumulatedCount.accumulate(taskCount);
+ tableAccumulatedCount.compute(tableNameWithType, (name, count) -> {
+ if (count == null) {
+ count = new TaskCount();
+ }
+ count.accumulate(taskCount);
+ return count;
+ });
});
- });
+ } catch (Exception e) {
+ LOGGER.warn("Failed to get task count for task: {} of type: {} (task may have been purged from DAG)",
Review Comment:
The error message references 'DAG' (Directed Acyclic Graph) which is Helix-specific terminology that may not be immediately clear to all developers or operators reading logs. Consider clarifying this as 'Helix workflow DAG' or adding a brief explanation of what it means for a task to be purged from the DAG.
```suggestion
LOGGER.warn("Failed to get task count for task: {} of type: {} (task may have been purged from the Helix workflow DAG, i.e., removed from the workflow graph)",
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskMetricsEmitter.java:
##########
@@ -90,20 +106,71 @@ protected final void runTask(Properties periodicTaskProperties) {
TaskCount taskTypeAccumulatedCount = new TaskCount();
Map<String, TaskCount> tableAccumulatedCount = new HashMap<>();
try {
- Set<String> tasksInProgress = _helixTaskResourceManager.getTasksInProgress(taskType);
- final int numRunningTasks = tasksInProgress.size();
- for (String task : tasksInProgress) {
- Map<String, TaskCount> tableTaskCount = _helixTaskResourceManager.getTableTaskCount(task);
- tableTaskCount.forEach((tableNameWithType, taskCount) -> {
- taskTypeAccumulatedCount.accumulate(taskCount);
- tableAccumulatedCount.compute(tableNameWithType, (name, count) -> {
- if (count == null) {
- count = new TaskCount();
- }
- count.accumulate(taskCount);
- return count;
+ // Capture the current execution timestamp for this task type collection cycle
+ long currentExecutionTimestamp = System.currentTimeMillis();
+ // For task types encountered for the first time, use current time minus one collection cycle
+ // as the initial timestamp to ensure we don't miss tasks that started recently
+ long previousExecutionTimestamp = _previousExecutionTimestamps.computeIfAbsent(taskType,
+ k -> currentExecutionTimestamp - _taskMetricsEmitterFrequencyMs);
+
+ // Get currently in-progress tasks
+ Set<String> currentInProgressTasks = _helixTaskResourceManager.getTasksInProgress(taskType);
+
+ // Get tasks that were in-progress during the previous collection cycle
+ Set<String> previouslyInProgressTasks =
+ _previousInProgressTasks.getOrDefault(taskType, Collections.emptySet());
+
+ // Start with currently in-progress tasks
+ Set<String> tasksToReport = new HashSet<>(currentInProgressTasks);
+
+ // Include tasks that were in-progress previously but are no longer in-progress
+ // These tasks completed between collection cycles and need their final metrics reported
+ for (String taskName : previouslyInProgressTasks) {
+ if (!currentInProgressTasks.contains(taskName)) {
+ LOGGER.debug("Including task {} that completed between collection cycles for taskType: {}",
+ taskName, taskType);
+ tasksToReport.add(taskName);
+ }
+ }
+
+ // Include tasks that started and completed between collection cycles
+ Set<String> tasksStartedAfter = _helixTaskResourceManager.getTasksStartedAfter(
+ taskType, previousExecutionTimestamp);
+
+ // Filter out tasks we already know about to avoid duplicates
+ Set<String> shortLivedTasks = new HashSet<>();
+ for (String taskName : tasksStartedAfter) {
+ if (!currentInProgressTasks.contains(taskName) && !previouslyInProgressTasks.contains(taskName)) {
+ shortLivedTasks.add(taskName);
+ }
+ }
Review Comment:
This filtering logic duplicates the membership checks against currentInProgressTasks and previouslyInProgressTasks. Consider using Set operations (e.g., removeAll) for cleaner and more efficient code: `Set<String> shortLivedTasks = new HashSet<>(tasksStartedAfter); shortLivedTasks.removeAll(currentInProgressTasks); shortLivedTasks.removeAll(previouslyInProgressTasks);`
```suggestion
Set<String> shortLivedTasks = new HashSet<>(tasksStartedAfter);
shortLivedTasks.removeAll(currentInProgressTasks);
shortLivedTasks.removeAll(previouslyInProgressTasks);
```
--
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
Re: [PR] Fix Minion task metrics collection gaps [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps merged PR #17128:
URL: https://github.com/apache/pinot/pull/17128
--
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
Re: [PR] Fix Minion task metrics collection gaps [pinot]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #17128:
URL: https://github.com/apache/pinot/pull/17128#discussion_r2488867732
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskMetricsEmitter.java:
##########
@@ -90,20 +106,59 @@ protected final void runTask(Properties periodicTaskProperties) {
TaskCount taskTypeAccumulatedCount = new TaskCount();
Map<String, TaskCount> tableAccumulatedCount = new HashMap<>();
try {
- Set<String> tasksInProgress = _helixTaskResourceManager.getTasksInProgress(taskType);
- final int numRunningTasks = tasksInProgress.size();
- for (String task : tasksInProgress) {
- Map<String, TaskCount> tableTaskCount = _helixTaskResourceManager.getTableTaskCount(task);
- tableTaskCount.forEach((tableNameWithType, taskCount) -> {
- taskTypeAccumulatedCount.accumulate(taskCount);
- tableAccumulatedCount.compute(tableNameWithType, (name, count) -> {
- if (count == null) {
- count = new TaskCount();
- }
- count.accumulate(taskCount);
- return count;
+ // Capture the current execution timestamp for this task type collection cycle
+ long currentExecutionTimestamp = System.currentTimeMillis();
+ // For task types encountered for the first time, use current time minus one collection cycle
+ // as the initial timestamp to ensure we don't miss tasks that started recently
+ long previousExecutionTimestamp = _previousExecutionTimestamps.computeIfAbsent(taskType,
+ k -> currentExecutionTimestamp - _taskMetricsEmitterFrequencyMs);
+
+ // Get currently in-progress tasks (for metrics)
+ Set<String> currentInProgressTasks = _helixTaskResourceManager.getTasksInProgress(taskType);
+
+ // Get tasks that were in-progress during the previous collection cycle
+ Set<String> previouslyInProgressTasks =
+ _previousInProgressTasks.getOrDefault(taskType, Collections.emptySet());
+
+ // Get all tasks including those that started after the previous execution timestamp
+ // This combines in-progress tasks and short-lived tasks that started and completed between cycles
+ // in a single Helix call, avoiding duplicate getWorkflowConfig/getWorkflowContext calls
+ Set<String> tasksIncludingShortLived = _helixTaskResourceManager.getTasksInProgressAndRecent(
+ taskType, previousExecutionTimestamp);
+
+ // Start with all tasks that need reporting (in-progress + short-lived)
+ Set<String> tasksToReport = new HashSet<>(tasksIncludingShortLived);
+
+ // Include tasks that were in-progress previously but are no longer in-progress
+ // These tasks completed between collection cycles and need their final metrics reported
+ for (String taskName : previouslyInProgressTasks) {
+ if (!tasksIncludingShortLived.contains(taskName)) {
+ LOGGER.debug("Including task {} that completed between collection cycles for taskType: {}",
+ taskName, taskType);
+ tasksToReport.add(taskName);
+ }
+ }
+
+ final int numRunningTasks = currentInProgressTasks.size();
+
+ // Process all tasks that need metrics reported
+ for (String task : tasksToReport) {
+ try {
+ Map<String, TaskCount> tableTaskCount = _helixTaskResourceManager.getTableTaskCount(task);
+ tableTaskCount.forEach((tableNameWithType, taskCount) -> {
+ taskTypeAccumulatedCount.accumulate(taskCount);
+ tableAccumulatedCount.compute(tableNameWithType, (name, count) -> {
+ if (count == null) {
+ count = new TaskCount();
+ }
+ count.accumulate(taskCount);
+ return count;
+ });
});
- });
+ } catch (Exception e) {
+ LOGGER.warn("Failed to get task count for task: {} of type: {} (task may have been purged from DAG)",
+ task, taskType, e);
Review Comment:
The error message indicates the task 'may have been purged from DAG', but this is speculative. The exception could occur for multiple reasons (network issues, Helix unavailability, etc.). Consider either logging at debug level with the current message, or using a more general warning message like 'Failed to get task count for task: {} of type: {}' without speculating about the cause.
```suggestion
LOGGER.warn("Failed to get task count for task: {} of type: {}", task, taskType, e);
```
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java:
##########
@@ -441,24 +441,63 @@ public synchronized Map<String, TaskCount> getTableTaskCount(String taskName) {
* @return Set of task names
*/
public synchronized Set<String> getTasksInProgress(String taskType) {
- WorkflowConfig workflowConfig = _taskDriver.getWorkflowConfig(getHelixJobQueueName(taskType));
+ return getTasksInProgressAndRecent(taskType, 0);
+ }
+
+ /**
+ * Returns a set of Task names (in the form "Task_<taskType>_<uuid>_<timestamp>") that are in progress or not started
+ * yet, and optionally includes recent tasks that started after a given timestamp.
+ * NOTE: For tasks just submitted without the context created, count them as NOT_STARTED.
+ * This method combines in-progress tasks and recent tasks in a single Helix call to avoid duplicate calls.
+ *
+ * @param taskType Task type
+ * @param afterTimestampMs If > 0, also include tasks that started after this timestamp (in milliseconds).
+ * This is used to detect short-lived tasks that started and completed between cycles.
+ * @return Set of task names that are in-progress, and optionally recent tasks that started after the timestamp
+ */
+ public synchronized Set<String> getTasksInProgressAndRecent(String taskType, long afterTimestampMs) {
+ String helixJobQueueName = getHelixJobQueueName(taskType);
+ WorkflowConfig workflowConfig = _taskDriver.getWorkflowConfig(helixJobQueueName);
if (workflowConfig == null) {
return Collections.emptySet();
}
Set<String> helixJobs = workflowConfig.getJobDag().getAllNodes();
if (helixJobs.isEmpty()) {
return Collections.emptySet();
}
- WorkflowContext workflowContext = _taskDriver.getWorkflowContext(getHelixJobQueueName(taskType));
+ WorkflowContext workflowContext = _taskDriver.getWorkflowContext(helixJobQueueName);
if (workflowContext == null) {
- return helixJobs.stream().map(PinotHelixTaskResourceManager::getPinotTaskName).collect(Collectors.toSet());
- } else {
- Map<String, TaskState> helixJobStates = workflowContext.getJobStates();
- return helixJobs.stream().filter(helixJobName -> {
- TaskState taskState = helixJobStates.get(helixJobName);
- return taskState == null || taskState == TaskState.NOT_STARTED || taskState == TaskState.IN_PROGRESS;
- }).map(PinotHelixTaskResourceManager::getPinotTaskName).collect(Collectors.toSet());
+ // If no context, return all jobs as in-progress (backward compatible behavior)
+ Set<String> result = helixJobs.stream()
+ .map(PinotHelixTaskResourceManager::getPinotTaskName)
+ .collect(Collectors.toSet());
+ // If timestamp is specified, we can't filter by start time without context, so return all
Review Comment:
This comment suggests that when a timestamp is specified but no context exists, all tasks are returned without filtering. However, this behavior may not align with the caller's expectations when requesting tasks after a specific timestamp. Consider documenting this limitation in the method's Javadoc to make it clear that timestamp filtering requires a valid WorkflowContext.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskMetricsEmitter.java:
##########
@@ -90,20 +106,59 @@ protected final void runTask(Properties periodicTaskProperties) {
TaskCount taskTypeAccumulatedCount = new TaskCount();
Map<String, TaskCount> tableAccumulatedCount = new HashMap<>();
try {
- Set<String> tasksInProgress = _helixTaskResourceManager.getTasksInProgress(taskType);
- final int numRunningTasks = tasksInProgress.size();
- for (String task : tasksInProgress) {
- Map<String, TaskCount> tableTaskCount = _helixTaskResourceManager.getTableTaskCount(task);
- tableTaskCount.forEach((tableNameWithType, taskCount) -> {
- taskTypeAccumulatedCount.accumulate(taskCount);
- tableAccumulatedCount.compute(tableNameWithType, (name, count) -> {
- if (count == null) {
- count = new TaskCount();
- }
- count.accumulate(taskCount);
- return count;
+ // Capture the current execution timestamp for this task type collection cycle
+ long currentExecutionTimestamp = System.currentTimeMillis();
+ // For task types encountered for the first time, use current time minus one collection cycle
+ // as the initial timestamp to ensure we don't miss tasks that started recently
+ long previousExecutionTimestamp = _previousExecutionTimestamps.computeIfAbsent(taskType,
+ k -> currentExecutionTimestamp - _taskMetricsEmitterFrequencyMs);
+
+ // Get currently in-progress tasks (for metrics)
+ Set<String> currentInProgressTasks = _helixTaskResourceManager.getTasksInProgress(taskType);
+
+ // Get tasks that were in-progress during the previous collection cycle
+ Set<String> previouslyInProgressTasks =
+ _previousInProgressTasks.getOrDefault(taskType, Collections.emptySet());
+
+ // Get all tasks including those that started after the previous execution timestamp
+ // This combines in-progress tasks and short-lived tasks that started and completed between cycles
+ // in a single Helix call, avoiding duplicate getWorkflowConfig/getWorkflowContext calls
+ Set<String> tasksIncludingShortLived = _helixTaskResourceManager.getTasksInProgressAndRecent(
+ taskType, previousExecutionTimestamp);
+
+ // Start with all tasks that need reporting (in-progress + short-lived)
+ Set<String> tasksToReport = new HashSet<>(tasksIncludingShortLived);
+
+ // Include tasks that were in-progress previously but are no longer in-progress
+ // These tasks completed between collection cycles and need their final metrics reported
+ for (String taskName : previouslyInProgressTasks) {
+ if (!tasksIncludingShortLived.contains(taskName)) {
+ LOGGER.debug("Including task {} that completed between collection cycles for taskType: {}",
Review Comment:
[nitpick] This debug log message may be helpful during development, but in production it could generate significant log volume for active task types. Consider whether this debug logging is necessary, or if it should be at trace level. If kept, ensure it provides sufficient value for troubleshooting in production environments.
```suggestion
LOGGER.trace("Including task {} that completed between collection cycles for taskType: {}",
```
--
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
Re: [PR] Fix Minion task metrics collection gaps [pinot]
Posted by "krishan1390 (via GitHub)" <gi...@apache.org>.
krishan1390 commented on code in PR #17128:
URL: https://github.com/apache/pinot/pull/17128#discussion_r2486271796
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/TaskMetricsEmitter.java:
##########
@@ -90,20 +106,71 @@ protected final void runTask(Properties periodicTaskProperties) {
TaskCount taskTypeAccumulatedCount = new TaskCount();
Map<String, TaskCount> tableAccumulatedCount = new HashMap<>();
try {
- Set<String> tasksInProgress = _helixTaskResourceManager.getTasksInProgress(taskType);
- final int numRunningTasks = tasksInProgress.size();
- for (String task : tasksInProgress) {
- Map<String, TaskCount> tableTaskCount = _helixTaskResourceManager.getTableTaskCount(task);
- tableTaskCount.forEach((tableNameWithType, taskCount) -> {
- taskTypeAccumulatedCount.accumulate(taskCount);
- tableAccumulatedCount.compute(tableNameWithType, (name, count) -> {
- if (count == null) {
- count = new TaskCount();
- }
- count.accumulate(taskCount);
- return count;
+ // Capture the current execution timestamp for this task type collection cycle
+ long currentExecutionTimestamp = System.currentTimeMillis();
+ // For task types encountered for the first time, use current time minus one collection cycle
+ // as the initial timestamp to ensure we don't miss tasks that started recently
+ long previousExecutionTimestamp = _previousExecutionTimestamps.computeIfAbsent(taskType,
+ k -> currentExecutionTimestamp - _taskMetricsEmitterFrequencyMs);
+
+ // Get currently in-progress tasks
+ Set<String> currentInProgressTasks = _helixTaskResourceManager.getTasksInProgress(taskType);
+
+ // Get tasks that were in-progress during the previous collection cycle
+ Set<String> previouslyInProgressTasks =
+ _previousInProgressTasks.getOrDefault(taskType, Collections.emptySet());
+
+ // Start with currently in-progress tasks
+ Set<String> tasksToReport = new HashSet<>(currentInProgressTasks);
+
+ // Include tasks that were in-progress previously but are no longer in-progress
+ // These tasks completed between collection cycles and need their final metrics reported
+ for (String taskName : previouslyInProgressTasks) {
+ if (!currentInProgressTasks.contains(taskName)) {
+ LOGGER.debug("Including task {} that completed between collection cycles for taskType: {}",
+ taskName, taskType);
+ tasksToReport.add(taskName);
+ }
+ }
+
+ // Include tasks that started and completed between collection cycles
+ Set<String> tasksStartedAfter = _helixTaskResourceManager.getTasksStartedAfter(
+ taskType, previousExecutionTimestamp);
+
+ // Filter out tasks we already know about to avoid duplicates
+ Set<String> shortLivedTasks = new HashSet<>();
Review Comment:
not clear why we're creating shortLivedTasks and then adding it to tasksToReport
instead of lines 140 to 152, we can just do tasksToReport.addAll(tasksStartedAfter);
That will take care of deduplication etc
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java:
##########
@@ -461,6 +461,41 @@ public synchronized Set<String> getTasksInProgress(String taskType) {
}
}
+ /**
+ * Returns tasks that started after the given timestamp.
+ * This is used to detect short-lived tasks that started and completed between metric collection cycles.
+ *
+ * @param taskType Task type
+ * @param afterTimestampMs Only include tasks that started after this timestamp (in milliseconds)
+ * @return Set of task names that started after the timestamp
+ */
+ public synchronized Set<String> getTasksStartedAfter(String taskType, long afterTimestampMs) {
Review Comment:
rather than fetching tasks twice from helix for the same task type (in getTasksInProgress and getTasksStartedAfter), we can just update getTasksInProgress() to accept an optional start timestamp.
the method is anyways used only in metrics emitter so its fine to make the change
--
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
Re: [PR] Fix Minion task metrics collection gaps [pinot]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #17128:
URL: https://github.com/apache/pinot/pull/17128#issuecomment-3479310722
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/17128?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 56.14%. Comparing base ([`eb4c342`](https://app.codecov.io/gh/apache/pinot/commit/eb4c34299834ede4203e57cac717e68be8bc8ae9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`fe9b113`](https://app.codecov.io/gh/apache/pinot/commit/fe9b11307214b6da672078eb0429a1d91afdab51?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 9 commits behind head on master.
> :exclamation: There is a different number of reports uploaded between BASE (eb4c342) and HEAD (fe9b113). Click for more details.
>
> <details><summary>HEAD has 36 uploads less than BASE</summary>
>
>| Flag | BASE (eb4c342) | HEAD (fe9b113) |
>|------|------|------|
>|unittests1|2|1|
>|unittests|4|1|
>|java-11|5|0|
>|temurin|10|1|
>|java-21|5|1|
>|unittests2|2|0|
>|integration|6|0|
>|integration2|2|0|
>|integration1|2|0|
>|custom-integration1|2|0|
></details>
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #17128 +/- ##
============================================
- Coverage 63.24% 56.14% -7.11%
+ Complexity 1428 701 -727
============================================
Files 3104 2420 -684
Lines 183423 135421 -48002
Branches 28112 21641 -6471
============================================
- Hits 116010 76032 -39978
+ Misses 58442 52999 -5443
+ Partials 8971 6390 -2581
```
| [Flag](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [custom-integration1](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration1](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [integration2](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-11](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
| [java-21](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.14% <ø> (-7.05%)` | :arrow_down: |
| [temurin](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.14% <ø> (-7.11%)` | :arrow_down: |
| [unittests](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.14% <ø> (-7.10%)` | :arrow_down: |
| [unittests1](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `56.14% <ø> (-0.08%)` | :arrow_down: |
| [unittests2](https://app.codecov.io/gh/apache/pinot/pull/17128/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `?` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/pinot/pull/17128?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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
Re: [PR] Fix Minion task metrics collection gaps [pinot]
Posted by "KKcorps (via GitHub)" <gi...@apache.org>.
KKcorps commented on code in PR #17128:
URL: https://github.com/apache/pinot/pull/17128#discussion_r2493135613
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java:
##########
@@ -441,24 +441,63 @@ public synchronized Map<String, TaskCount> getTableTaskCount(String taskName) {
* @return Set of task names
*/
public synchronized Set<String> getTasksInProgress(String taskType) {
- WorkflowConfig workflowConfig = _taskDriver.getWorkflowConfig(getHelixJobQueueName(taskType));
+ return getTasksInProgressAndRecent(taskType, 0);
+ }
+
+ /**
+ * Returns a set of Task names (in the form "Task_<taskType>_<uuid>_<timestamp>") that are in progress or not started
+ * yet, and optionally includes recent tasks that started after a given timestamp.
+ * NOTE: For tasks just submitted without the context created, count them as NOT_STARTED.
+ * This method combines in-progress tasks and recent tasks in a single Helix call to avoid duplicate calls.
+ *
+ * @param taskType Task type
+ * @param afterTimestampMs If > 0, also include tasks that started after this timestamp (in milliseconds).
+ * This is used to detect short-lived tasks that started and completed between cycles.
+ * @return Set of task names that are in-progress, and optionally recent tasks that started after the timestamp
+ */
+ public synchronized Set<String> getTasksInProgressAndRecent(String taskType, long afterTimestampMs) {
+ String helixJobQueueName = getHelixJobQueueName(taskType);
+ WorkflowConfig workflowConfig = _taskDriver.getWorkflowConfig(helixJobQueueName);
if (workflowConfig == null) {
return Collections.emptySet();
}
Set<String> helixJobs = workflowConfig.getJobDag().getAllNodes();
if (helixJobs.isEmpty()) {
return Collections.emptySet();
}
- WorkflowContext workflowContext = _taskDriver.getWorkflowContext(getHelixJobQueueName(taskType));
+ WorkflowContext workflowContext = _taskDriver.getWorkflowContext(helixJobQueueName);
if (workflowContext == null) {
- return helixJobs.stream().map(PinotHelixTaskResourceManager::getPinotTaskName).collect(Collectors.toSet());
- } else {
- Map<String, TaskState> helixJobStates = workflowContext.getJobStates();
- return helixJobs.stream().filter(helixJobName -> {
- TaskState taskState = helixJobStates.get(helixJobName);
- return taskState == null || taskState == TaskState.NOT_STARTED || taskState == TaskState.IN_PROGRESS;
- }).map(PinotHelixTaskResourceManager::getPinotTaskName).collect(Collectors.toSet());
+ // If no context, return all jobs as in-progress (backward compatible behavior)
+ Set<String> result = helixJobs.stream()
+ .map(PinotHelixTaskResourceManager::getPinotTaskName)
+ .collect(Collectors.toSet());
+ // If timestamp is specified, we can't filter by start time without context, so return all
+ return result;
+ }
+
+ Set<String> result = new HashSet<>();
+ Map<String, TaskState> helixJobStates = workflowContext.getJobStates();
+ Map<String, Long> jobStartTimes = afterTimestampMs > 0 ? workflowContext.getJobStartTimes() : null;
Review Comment:
What if `workflowContext.getJobStartTimes()` is null as well here? it can happen if task queue hasn't been created yet
--
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
Re: [PR] Fix Minion task metrics collection gaps [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #17128:
URL: https://github.com/apache/pinot/pull/17128#discussion_r2493370538
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java:
##########
@@ -441,24 +441,63 @@ public synchronized Map<String, TaskCount> getTableTaskCount(String taskName) {
* @return Set of task names
*/
public synchronized Set<String> getTasksInProgress(String taskType) {
- WorkflowConfig workflowConfig = _taskDriver.getWorkflowConfig(getHelixJobQueueName(taskType));
+ return getTasksInProgressAndRecent(taskType, 0);
+ }
+
+ /**
+ * Returns a set of Task names (in the form "Task_<taskType>_<uuid>_<timestamp>") that are in progress or not started
+ * yet, and optionally includes recent tasks that started after a given timestamp.
+ * NOTE: For tasks just submitted without the context created, count them as NOT_STARTED.
+ * This method combines in-progress tasks and recent tasks in a single Helix call to avoid duplicate calls.
+ *
+ * @param taskType Task type
+ * @param afterTimestampMs If > 0, also include tasks that started after this timestamp (in milliseconds).
+ * This is used to detect short-lived tasks that started and completed between cycles.
+ * @return Set of task names that are in-progress, and optionally recent tasks that started after the timestamp
+ */
+ public synchronized Set<String> getTasksInProgressAndRecent(String taskType, long afterTimestampMs) {
+ String helixJobQueueName = getHelixJobQueueName(taskType);
+ WorkflowConfig workflowConfig = _taskDriver.getWorkflowConfig(helixJobQueueName);
if (workflowConfig == null) {
return Collections.emptySet();
}
Set<String> helixJobs = workflowConfig.getJobDag().getAllNodes();
if (helixJobs.isEmpty()) {
return Collections.emptySet();
}
- WorkflowContext workflowContext = _taskDriver.getWorkflowContext(getHelixJobQueueName(taskType));
+ WorkflowContext workflowContext = _taskDriver.getWorkflowContext(helixJobQueueName);
if (workflowContext == null) {
- return helixJobs.stream().map(PinotHelixTaskResourceManager::getPinotTaskName).collect(Collectors.toSet());
- } else {
- Map<String, TaskState> helixJobStates = workflowContext.getJobStates();
- return helixJobs.stream().filter(helixJobName -> {
- TaskState taskState = helixJobStates.get(helixJobName);
- return taskState == null || taskState == TaskState.NOT_STARTED || taskState == TaskState.IN_PROGRESS;
- }).map(PinotHelixTaskResourceManager::getPinotTaskName).collect(Collectors.toSet());
+ // If no context, return all jobs as in-progress (backward compatible behavior)
+ Set<String> result = helixJobs.stream()
+ .map(PinotHelixTaskResourceManager::getPinotTaskName)
+ .collect(Collectors.toSet());
+ // If timestamp is specified, we can't filter by start time without context, so return all
+ return result;
+ }
+
+ Set<String> result = new HashSet<>();
+ Map<String, TaskState> helixJobStates = workflowContext.getJobStates();
+ Map<String, Long> jobStartTimes = afterTimestampMs > 0 ? workflowContext.getJobStartTimes() : null;
Review Comment:
That would not happen.
1. Similar to `workflowContext.getJobStartTimes()` is `workflowContext.getJobStates()` and it did not have the null check before.
2. Both the methods, return empty map and not null as default value.
<img width="774" height="660" alt="Screenshot 2025-11-05 at 1 15 42 PM" src="https://github.com/user-attachments/assets/0bbc13ab-fbdf-46ed-aa48-85a4509810a5" />
--
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
Re: [PR] Fix Minion task metrics collection gaps [pinot]
Posted by "abhishekbafna (via GitHub)" <gi...@apache.org>.
abhishekbafna commented on code in PR #17128:
URL: https://github.com/apache/pinot/pull/17128#discussion_r2493370538
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/minion/PinotHelixTaskResourceManager.java:
##########
@@ -441,24 +441,63 @@ public synchronized Map<String, TaskCount> getTableTaskCount(String taskName) {
* @return Set of task names
*/
public synchronized Set<String> getTasksInProgress(String taskType) {
- WorkflowConfig workflowConfig = _taskDriver.getWorkflowConfig(getHelixJobQueueName(taskType));
+ return getTasksInProgressAndRecent(taskType, 0);
+ }
+
+ /**
+ * Returns a set of Task names (in the form "Task_<taskType>_<uuid>_<timestamp>") that are in progress or not started
+ * yet, and optionally includes recent tasks that started after a given timestamp.
+ * NOTE: For tasks just submitted without the context created, count them as NOT_STARTED.
+ * This method combines in-progress tasks and recent tasks in a single Helix call to avoid duplicate calls.
+ *
+ * @param taskType Task type
+ * @param afterTimestampMs If > 0, also include tasks that started after this timestamp (in milliseconds).
+ * This is used to detect short-lived tasks that started and completed between cycles.
+ * @return Set of task names that are in-progress, and optionally recent tasks that started after the timestamp
+ */
+ public synchronized Set<String> getTasksInProgressAndRecent(String taskType, long afterTimestampMs) {
+ String helixJobQueueName = getHelixJobQueueName(taskType);
+ WorkflowConfig workflowConfig = _taskDriver.getWorkflowConfig(helixJobQueueName);
if (workflowConfig == null) {
return Collections.emptySet();
}
Set<String> helixJobs = workflowConfig.getJobDag().getAllNodes();
if (helixJobs.isEmpty()) {
return Collections.emptySet();
}
- WorkflowContext workflowContext = _taskDriver.getWorkflowContext(getHelixJobQueueName(taskType));
+ WorkflowContext workflowContext = _taskDriver.getWorkflowContext(helixJobQueueName);
if (workflowContext == null) {
- return helixJobs.stream().map(PinotHelixTaskResourceManager::getPinotTaskName).collect(Collectors.toSet());
- } else {
- Map<String, TaskState> helixJobStates = workflowContext.getJobStates();
- return helixJobs.stream().filter(helixJobName -> {
- TaskState taskState = helixJobStates.get(helixJobName);
- return taskState == null || taskState == TaskState.NOT_STARTED || taskState == TaskState.IN_PROGRESS;
- }).map(PinotHelixTaskResourceManager::getPinotTaskName).collect(Collectors.toSet());
+ // If no context, return all jobs as in-progress (backward compatible behavior)
+ Set<String> result = helixJobs.stream()
+ .map(PinotHelixTaskResourceManager::getPinotTaskName)
+ .collect(Collectors.toSet());
+ // If timestamp is specified, we can't filter by start time without context, so return all
+ return result;
+ }
+
+ Set<String> result = new HashSet<>();
+ Map<String, TaskState> helixJobStates = workflowContext.getJobStates();
+ Map<String, Long> jobStartTimes = afterTimestampMs > 0 ? workflowContext.getJobStartTimes() : null;
Review Comment:
That would not happen.
1. Similar to `workflowContext.getJobStartTimes()` is `workflowContext.getJobStates()` and it did not have the null check before.
2. Both the methods, return empty map and not null as default value.
--
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