Supervisor

Apache Druid uses supervisors to manage streaming ingestion from external streaming sources into Druid. Supervisors oversee the state of indexing tasks to coordinate handoffs, manage failures, and ensure that the scalability and replication requirements are maintained. They can also be used to perform automatic compaction after data has been ingested.

This topic uses the Apache Kafka term offset to refer to the identifier for records in a partition. If you are using Amazon Kinesis, the equivalent is sequence number.

Supervisor spec

Druid uses a JSON specification, often referred to as the supervisor spec, to define tasks used for streaming ingestion or auto-compaction. The supervisor spec specifies how Druid should consume, process, and index data from an external stream or Druid itself.

The following table outlines the high-level configuration options for a supervisor spec:

I/O configuration

The following table outlines the ioConfig configuration properties that apply to both Apache Kafka and Amazon Kinesis ingestion methods. For configuration properties specific to Kafka and Kinesis, see Kafka I/O configuration and Kinesis I/O configuration respectively.

Task autoscaler

You can optionally configure autoscaling behavior for ingestion tasks using the autoScalerConfig property of the ioConfig object.

The following table outlines the configuration properties for autoScalerConfig:

Autoscaler strategy

info

Unlike the Kafka indexing service, Kinesis reports lag metrics as the time difference in milliseconds between the current sequence number and the latest sequence number, rather than message count.

The following table outlines the configuration properties related to the lagBased autoscaler strategy:

The following example shows a supervisor spec with lagBased autoscaler:

Click to view the example

{
  "type": "kinesis",
  "dataSchema": {
    "dataSource": "metrics-kinesis",
    "timestampSpec": {
      "column": "timestamp",
      "format": "auto"
    },
    "dimensionsSpec": {
      "dimensions": [],
      "dimensionExclusions": [
        "timestamp",
        "value"
      ]
    },
    "metricsSpec": [
      {
        "name": "count",
        "type": "count"
      },
      {
        "name": "value_sum",
        "fieldName": "value",
        "type": "doubleSum"
      },
      {
        "name": "value_min",
        "fieldName": "value",
        "type": "doubleMin"
      },
      {
        "name": "value_max",
        "fieldName": "value",
        "type": "doubleMax"
      }
    ],
    "granularitySpec": {
      "type": "uniform",
      "segmentGranularity": "HOUR",
      "queryGranularity": "NONE"
    }
  },
  "ioConfig": {
    "stream": "metrics",
    "autoScalerConfig": {
      "enableTaskAutoScaler": true,
      "taskCountMax": 6,
      "taskCountMin": 2,
      "minTriggerScaleActionFrequencyMillis": 600000,
      "autoScalerStrategy": "lagBased",
      "lagCollectionIntervalMillis": 30000,
      "lagCollectionRangeMillis": 600000,
      "scaleOutThreshold": 600000,
      "triggerScaleOutFractionThreshold": 0.3,
      "scaleInThreshold": 100000,
      "triggerScaleInFractionThreshold": 0.9,
      "scaleActionStartDelayMillis": 300000,
      "scaleActionPeriodMillis": 60000,
      "scaleInStep": 1,
      "scaleOutStep": 2
    },
    "inputFormat": {
      "type": "json"
    },
    "endpoint": "kinesis.us-east-1.amazonaws.com",
    "taskCount": 1,
    "replicas": 1,
    "taskDuration": "PT1H"
  },
  "tuningConfig": {
    "type": "kinesis",
    "maxRowsPerSegment": 5000000
  }
}

Tuning configuration

The tuningConfig object is optional. If you don’t specify the tuningConfig object, Druid uses the default configuration settings.

The following table outlines the tuningConfig configuration properties that apply to both Kafka and Kinesis ingestion methods. For configuration properties specific to Kafka and Kinesis, see Kafka tuning configuration and Kinesis tuning configuration respectively.

Start a supervisor

Druid starts a new supervisor when you submit a supervisor spec. You can submit the supervisor spec in the Druid web console data loader or with the Supervisor API.

The following screenshot shows the Supervisors view of the web console for a cluster with two supervisors:

Once started, the supervisor persists in the configured metadata database. There can only be one supervisor per datasource. Submitting a second supervisor spec for the same datasource overwrites the previous one.

When an Overlord gains leadership, either by being started or as a result of another Overlord failing, it spawns a supervisor for each supervisor spec in the metadata database. The supervisor then discovers running indexing tasks and attempts to adopt them if they are compatible with the supervisor’s configuration. If they are not compatible, the tasks are terminated and the supervisor creates a new set of tasks. This way, the supervisor ingestion tasks persist across Overlord restarts and failovers.

Schema and configuration changes

To make schema or configuration changes, you must submit a new supervisor spec. The Overlord initiates a graceful shutdown of the existing supervisor. The running supervisor signals its tasks to stop reading and begin publishing, exiting itself. Druid then uses the new configuration to create a new supervisor. Druid submits the updated schema while retaining existing publishing tasks. It also starts new tasks at the previous task offsets. This way, configuration changes can be applied without requiring any pause in ingestion.

Status report

The supervisor status report contains the state of the supervisor tasks and an array of recently thrown exceptions reported as recentErrors. You can control the maximum size of the exceptions using the druid.supervisor.maxStoredExceptionEvents configuration.

To view the supervisor status in the web console, navigate to the Supervisors view and click the supervisor ID to open the Supervisor dialog. Click Status in the left navigation pane to display the status:

The following example shows the status of a supervisor with the name social_media:

Click to view the example

{
  "dataSource": "social_media",
  "stream": "social_media",
  "partitions": 1,
  "replicas": 1,
  "durationSeconds": 3600,
  "activeTasks": [
    {
      "id": "index_kafka_social_media_8ff3096f21fe448_jajnddno",
      "startingOffsets": {
        "0": 0
      },
      "startTime": "2024-01-30T21:21:41.696Z",
      "remainingSeconds": 479,
      "type": "ACTIVE",
      "currentOffsets": {
        "0": 50000
      },
      "lag": {
        "0": 0
      }
    }
  ],
  "publishingTasks": [],
  "latestOffsets": {
    "0": 50000
  },
  "minimumLag": {
    "0": 0
  },
  "aggregateLag": 0,
  "offsetsLastUpdated": "2024-01-30T22:13:19.335Z",
  "suspended": false,
  "healthy": true,
  "state": "RUNNING",
  "detailedState": "RUNNING",
  "recentErrors": []
}

The status report contains two properties that correspond to the state of the supervisor: state and detailedState. The state property contains a small number of generic states that apply to any type of supervisor. The detailedState property contains a more descriptive, implementation-specific state that may provide more insight into the supervisor’s activities.

Possible state values are PENDING, RUNNING, SUSPENDED, STOPPING, UNHEALTHY_SUPERVISOR, and UNHEALTHY_TASKS.

The following table lists detailedState values and their corresponding state mapping:

On each iteration of the supervisor’s run loop, the supervisor completes the following tasks in sequence:

1. Retrieve the list of partitions and determine the starting offset for each partition. If continuing, Druid uses the last processed offset. For new streams, Druid starts from either the beginning or end of the stream, depending on the useEarliestOffset property.
2. Discover any running indexing tasks that are writing to the supervisor’s datasource and adopt them if they match the supervisor’s configuration, else signal them to stop.
3. Send a status request to each supervised task to update the view of the state of the tasks under supervision.
4. Handle tasks that have exceeded taskDuration and should transition from reading to publishing.
5. Handle tasks that have finished publishing and signal redundant replica tasks to stop.
6. Handle tasks that have failed and clean up the supervisor’s internal state.
7. Compare the list of healthy tasks to the requested taskCount and replicas configurations and create additional tasks if required.

The detailedState property shows additional values (marked with “first iteration only” in the preceding table) the first time the supervisor executes this run loop after startup or after resuming from a suspension. This is intended to surface initialization-type issues, where the supervisor is unable to reach a stable state. For example, if the supervisor can’t connect to the stream, if it’s unable to read from the stream, or if it can’t communicate with existing tasks. Once the supervisor is stable; that is, once it has completed a full execution without encountering any issues, detailedState will show a RUNNING state until it is stopped, suspended, or hits a failure threshold and transitions to an unhealthy state.

info

For the Kafka indexing service, Druid may report the consumer lag per partition as a negative value if the supervisor hasn’t received the latest offset response from Kafka. The aggregate lag value is always >= 0.

SUPERVISORS system table

Druid exposes system information through special system schemas. You can query the sys.supervisors table to retrieve information about the supervisor internals. The following example shows how to retrieve supervisor tasks information filtered by health status:

SELECT * FROM sys.supervisors WHERE healthy=0;

For more information on the supervisors system table, see SUPERVISORS table.

Manage a supervisor

You can manage a supervisor from the web console or with the Supervisor API. In the web console, navigate to the Supervisors view and click the ellipsis in the Actions column. Select the desired action from the menu that appears.

The supervisor must be running for some of these actions to be available.

Suspend

Suspend pauses a running supervisor. The suspended supervisor continues to emit logs and metrics. Indexing tasks remain suspended until you resume the supervisor. For information on how to suspend a supervisor by API, see Supervisors: Suspend a running supervisor.

Set offsets

info

Perform this action with caution as it may result in skipped messages and lead to data loss or duplicate data.

Set offsets resets the offsets for supervisor partitions. This action clears the stored offsets and instructs the supervisor to resume reading data from the specified offsets. If there are no stored offsets, Druid saves the specified offsets in the metadata store. Set offsets terminates and recreates active tasks for the specified partitions to begin reading from the reset offsets. For partitions not specified in this operation, the supervisor resumes from the last stored offset.

For information on how to reset offsets by API, see Supervisors: Reset offsets for a supervisor.

Hard reset

info

Perform this action with caution as it may result in skipped messages and lead to data loss or duplicate data.

Hard reset clears supervisor metadata, causing the supervisor to resume data reading from either the earliest or latest available position, depending on the useEarliestOffset setting. Hard reset terminates and recreates active tasks, so that tasks begin reading from valid positions.

Use this action to recover from a stopped state due to missing offsets.

For information on how to reset a supervisor by API, see Supervisors: Reset a supervisor.

Terminate

Terminate stops a supervisor and its indexing tasks, triggering the publishing of their segments. When you terminate a supervisor, Druid places a tombstone marker in the metadata store to prevent reloading on restart. The terminated supervisor still exists in the metadata store and its history can be retrieved.

For information on how to terminate a supervisor by API, see Supervisors: Terminate a supervisor.

Capacity planning

Indexing tasks run on Middle Managers and are limited by the resources available in the Middle Manager cluster. In particular, you should make sure that you have sufficient worker capacity, configured using the druid.worker.capacity property, to handle the configuration in the supervisor spec. Note that worker capacity is shared across all types of indexing tasks, so you should plan your worker capacity to handle your total indexing load, such as batch processing, streaming tasks, and merging tasks. If your workers run out of capacity, indexing tasks queue and wait for the next available worker. This may cause queries to return partial results but will not result in data loss, assuming the tasks run before the stream purges those sequence numbers.

A running task can be in one of two states: reading or publishing. A task remains in reading state for the period defined in taskDuration, at which point it transitions to publishing state. A task remains in publishing state for as long as it takes to generate segments, push segments to deep storage, and have them loaded and served by a Historical service or until completionTimeout elapses.

The number of reading tasks is controlled by replicas and taskCount. In general, there are replicas * taskCount reading tasks. An exception occurs if taskCount is over the number of shards in Kinesis or partitions in Kafka, in which case Druid uses the number of shards or partitions. When taskDuration elapses, these tasks transition to publishing state and replicas * taskCount new reading tasks are created. To allow for reading tasks and publishing tasks to run concurrently, there should be a minimum capacity of:

workerCapacity = 2 * replicas * taskCount

This value is for the ideal situation in which there is at most one set of tasks publishing while another set is reading. In some circumstances, it is possible to have multiple sets of tasks publishing simultaneously. This would happen if the time-to-publish (generate segment, push to deep storage, load on Historical) is greater than taskDuration. This is a valid and correct scenario but requires additional worker capacity to support. In general, it is a good idea to have taskDuration be large enough that the previous set of tasks finishes publishing before the current set begins.

Learn more

See the following topics for more information:

• Supervisor API for how to manage and monitor supervisors using the API.
• Apache Kafka ingestion to learn about ingesting data from an Apache Kafka stream.
• Amazon Kinesis ingestion to learn about ingesting data from an Amazon Kinesis stream.