Minion Task Plugin

Write a custom Minion task plugin with a task generator and task executor.

Minion tasks execute background maintenance and data processing jobs in Pinot. Each task type has two components: a Task Generator (runs on the Controller) that creates task configurations, and a Task Executor (runs on the Minion) that performs the work.

Architecture

Controller                          Minion
┌────────────────────┐             ┌────────────────────┐
│  PinotTaskGenerator│  ─ Helix ─> │ PinotTaskExecutor  │
│  - generateTasks() │  schedules  │ - executeTask()    │
│  - getTaskType()   │             │ - cancel()         │
└────────────────────┘             └────────────────────┘

1. The Controller runs the Task Generator on a schedule or via ad-hoc API call

2. The Generator creates PinotTaskConfig objects describing the work

3. The Controller schedules tasks via Helix

4. Minion instances pick up tasks based on tag/routing

5. The Minion executes the task using the Task Executor

6. Results are returned to the Controller

Step 1: Implement the Task Executor

The executor runs on the Minion and performs the actual work.

Task Executor Factory

import org.apache.pinot.minion.executor.PinotTaskExecutorFactory;
import org.apache.pinot.minion.executor.PinotTaskExecutor;
import org.apache.pinot.spi.annotations.minion.TaskExecutorFactory;

@TaskExecutorFactory(enabled = true)
public class MyTaskExecutorFactory implements PinotTaskExecutorFactory {

  @Override
  public void init(MinionTaskZkMetadataManager zkMetadataManager, MinionConf minionConf) {
    // Initialize resources needed by the executor
  }

  @Override
  public String getTaskType() {
    return "MyCustomTask";  // Must match the generator's task type
  }

  @Override
  public PinotTaskExecutor create() {
    return new MyTaskExecutor();
  }
}

Task Executor

import org.apache.pinot.minion.executor.PinotTaskExecutor;
import org.apache.pinot.core.minion.PinotTaskConfig;

public class MyTaskExecutor implements PinotTaskExecutor {

  private volatile boolean _cancelled = false;

  @Override
  public Object executeTask(PinotTaskConfig pinotTaskConfig) throws Exception {
    String tableName = pinotTaskConfig.getTableName();
    Map<String, String> config = pinotTaskConfig.getConfig();

    // Perform the task work here
    // Check _cancelled periodically for long-running tasks

    return "Task completed successfully";
  }

  @Override
  public void cancel() {
    _cancelled = true;
  }
}

The executor factory class must be annotated with @TaskExecutorFactory(enabled = true) for auto-registration. The class must be in a package matching org.apache.pinot.*.plugin.minion.tasks.*.

Safely replace a segment with metadata push

If a custom task extends BaseSingleSegmentConversionExecutor, rebuilds a segment under the same segment name, and uses push.mode=METADATA, opt in to controller-side copying:

@Override
protected boolean isCopyToDeepStoreForMetadataPush() {
  return true;
}

Also provide output.segment.dir.uri in each task configuration. The Minion uses this location to stage the converted segment, and the Controller must be able to access both the staging location and the table's deep store.

This opt-in path preserves the existing segment download URL. The Minion stages the replacement with a task-specific name, sends the original segment CRC and refresh-only guard to the Controller, and asks the Controller to copy the staged data into the segment's existing deep-store location. After the request completes or fails, the Minion deletes the staged file. A retry of the same task can overwrite its own leftover staging file.

If conversion produces the same CRC as the original segment, Pinot does not stage or copy segment data. It only refreshes the segment metadata against the existing download URL.

Override this method only for same-name replacement tasks. It defaults to false, so existing tasks—including PurgeTask, RefreshSegmentTask, and UpsertCompactionTask—keep their current metadata-push behavior unless their executor explicitly opts in.

Step 2: Implement the Task Generator

The generator runs on the Controller and decides when and how to create tasks.

import org.apache.pinot.controller.helix.core.minion.generator.PinotTaskGenerator;
import org.apache.pinot.spi.annotations.minion.TaskGenerator;
import org.apache.pinot.spi.config.table.TableConfig;
import org.apache.pinot.core.minion.PinotTaskConfig;

@TaskGenerator(enabled = true)
public class MyTaskGenerator implements PinotTaskGenerator {

  private ClusterInfoAccessor _clusterInfoAccessor;

  @Override
  public void init(ClusterInfoAccessor clusterInfoAccessor) {
    _clusterInfoAccessor = clusterInfoAccessor;
  }

  @Override
  public String getTaskType() {
    return "MyCustomTask";  // Must match the executor's task type
  }

  // Called on schedule for all tables with this task type enabled
  @Override
  public List<PinotTaskConfig> generateTasks(List<TableConfig> tableConfigs) {
    List<PinotTaskConfig> tasks = new ArrayList<>();
    for (TableConfig tableConfig : tableConfigs) {
      Map<String, String> taskConfig = new HashMap<>();
      taskConfig.put("myParam", "myValue");

      tasks.add(new PinotTaskConfig(
          getTaskType(),
          tableConfig.getTableName(),
          taskConfig));
    }
    return tasks;
  }

  // Called for ad-hoc task execution via API
  @Override
  public List<PinotTaskConfig> generateTasks(
      TableConfig tableConfig, Map<String, String> taskConfigs) throws Exception {
    return List.of(new PinotTaskConfig(
        getTaskType(),
        tableConfig.getTableName(),
        taskConfigs));
  }
}

Optional Generator Overrides

Method	Default	Description
getTaskTimeoutMs(String minionTag)	3,600,000 (1 hour)	Task timeout in milliseconds
getNumConcurrentTasksPerInstance(String minionTag)	1	Max concurrent tasks per Minion
getMaxAttemptsPerTask(String minionTag)	1 (no retry)	Max retry attempts
getMinionInstanceTag(TableConfig tableConfig)	UNTAGGED_MINION_INSTANCE	Tag for routing tasks to specific Minion instances
validateTaskConfigs(TableConfig, Schema, Map)	no-op	Validate task-specific configs at table creation time

Step 3: Enable the Task in Table Config

Add the task configuration to the table config:

{
  "tableName": "myTable_OFFLINE",
  "task": {
    "taskTypeConfigsMap": {
      "MyCustomTask": {
        "schedule": "0 0 * * * ?",
        "myParam": "myValue"
      }
    }
  }
}

The schedule property uses a CRON expression to define when the task generator runs. Without it, the task can only be triggered via the Controller API.

Step 4: Trigger Tasks via API

To manually trigger task generation:

# Generate tasks for all tables with this task type
POST /tasks/schedule?taskType=MyCustomTask

# Generate tasks for a specific table
POST /tasks/execute?taskType=MyCustomTask&tableName=myTable_OFFLINE

Built-in Task Types

Pinot includes several built-in Minion tasks:

Task Type	Description
RealtimeToOfflineSegmentsTask	Converts realtime segments to offline segments
UpsertCompactionTask	Compacts segments in upsert-enabled tables
UpsertCompactMergeTask	Merges and compacts upsert segments
PurgeTask	Purges records matching a purge function
MergeRollupTask	Merges and rolls up small segments
SegmentGenerationAndPushTask	Generates and pushes segments from external data
RefreshSegmentTask	Refreshes segments from deep store

For more details on built-in tasks, see Minion Merge Rollup Task and Upsert Compaction Task.