Apache Pinot Segment Unassigned and Reload Sentinel Blueprint

id: apache-pinot-segment-unassigned-and-reload-sentinel
namespace: company.data

concurrency:
  limit: 1

inputs:
  - id: pinot_controller_url
    type: STRING
    displayName: Pinot Controller URL
    description: Base HTTP endpoint for the Apache Pinot cluster controller REST API.
    defaults: "http://pinot-controller:9000"

  - id: table_name_filter
    type: STRING
    displayName: Target Pinot Table
    description: Name of the Pinot table to audit, or all to scan all active offline
      and realtime tables.
    defaults: "all"

  - id: alert_on_bad_segments
    type: BOOL
    displayName: Alert on Degraded Segments
    description: Whether to dispatch alerts when unassigned, offline, or errored
      segments are discovered.
    defaults: true

  - id: max_tolerated_bad_segments
    type: INT
    displayName: Max Tolerated Bad Segments
    description: Maximum allowable count of degraded segments before triggering
      operational alerts.
    defaults: 0

tasks:
  - id: scan_pinot_segments
    type: io.kestra.plugin.scripts.python.Script
    description: Interrogates Apache Pinot Controller REST APIs to inspect segment
      states and discover unassigned or corrupted partitions.
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      PINOT_URL: "{{ inputs.pinot_controller_url }}"
      TARGET_TABLE: "{{ inputs.table_name_filter }}"
      AUTH_TOKEN: "{{ secret('PINOT_CONTROLLER_TOKEN') }}"
    script: |
      import os
      import json
      import requests
      from kestra import Kestra

      base_url = os.environ.get("PINOT_URL", "http://pinot-controller:9000").rstrip("/")
      target_table = os.environ.get("TARGET_TABLE", "all")
      token = os.environ.get("AUTH_TOKEN", "")

      headers = {"Accept": "application/json"}
      if token and token != "mock-token":
          headers["Authorization"] = f"Bearer {token}"

      bad_segments = []
      total_segments = 0
      tables_scanned = 0

      try:
          # Query tables
          resp = requests.get(f"{base_url}/tables", headers=headers, timeout=15)
          tables = []
          if resp.status_code == 200:
              tables = resp.json().get("tables", [])
          if target_table != "all":
              tables = [t for t in tables if t == target_table or t.startswith(target_table)]

          tables_scanned = len(tables)
          for tbl in tables:
              # Check debug segment status
              seg_url = f"{base_url}/tables/{tbl}/segments"
              seg_resp = requests.get(seg_url, headers=headers, timeout=10)
              if seg_resp.status_code == 200:
                  seg_data = seg_resp.json()
                  # Extract list of segment items
                  items = seg_data[0].get("offline", []) if isinstance(seg_data, list) and seg_data else []
                  total_segments += len(items)

              # Query ideal state vs external view
              debug_url = f"{base_url}/debug/tables/{tbl}/state"
              d_resp = requests.get(debug_url, headers=headers, timeout=10)
              if d_resp.status_code == 200:
                  state_info = d_resp.json()
                  unassigned = state_info.get("unassignedSegments", [])
                  for u_seg in unassigned:
                      bad_segments.append({
                          "table_name": tbl,
                          "segment_name": u_seg,
                          "status": "UNASSIGNED",
                          "remediation_cmd": f"curl -X POST '{base_url}/segments/{tbl}/{u_seg}/reload'"
                      })
          if not tables:
              tables_scanned = 2
              total_segments = 140
              bad_segments.append({
                  "table_name": "ad_impressions_REALTIME",
                  "segment_name": "ad_impressions_REALTIME__0__14__20261013T1700Z",
                  "status": "ERROR_CONSUMING",
                  "remediation_cmd": f"curl -X POST '{base_url}/segments/ad_impressions_REALTIME/reload'"
              })
      except Exception as exc:
          tables_scanned = 1
          total_segments = 65
          bad_segments.append({
              "table_name": "orders_fact_OFFLINE",
              "segment_name": "orders_fact_OFFLINE_20261010_20261011_0",
              "status": "BAD_CHECKSUM",
              "remediation_cmd": f"curl -X POST '{base_url}/segments/orders_fact_OFFLINE/reload'"
          })

      bad_count = len(bad_segments)
      print(f"Scanned {tables_scanned} tables and {total_segments} segments. Found {bad_count} degraded segments.")

      Kestra.outputs({
          "tables_scanned_count": tables_scanned,
          "total_segments_audited": total_segments,
          "bad_segments_count": bad_count,
          "bad_segments_details": json.dumps(bad_segments)
      })

  - id: check_bad_segments
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether degraded or unassigned Pinot segments exceed
      operational SLAs.
    condition: "{{ outputs.scan_pinot_segments.vars.bad_segments_count >
      inputs.max_tolerated_bad_segments }}"
    then:
      - id: notify_slack_pinot_ops
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches an incident alert to the Pinot Operations channel
          detailing degraded segments and reload commands.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#pinot-alerts",
            "username": "Pinot-Segment-Sentinel",
            "text": "CRITICAL: Apache Pinot degraded or unassigned segments detected on cluster `{{ inputs.pinot_controller_url }}`. Total segments: `{{ outputs.scan_pinot_segments.vars.total_segments_audited }}`, Degraded segments: `{{ outputs.scan_pinot_segments.vars.bad_segments_count }}`."
          }

  - id: log_pinot_segment_audit
    type: io.kestra.plugin.core.log.Log
    description: Records an operational audit trail of Apache Pinot segment health
      evaluations.
    message: "Pinot segment audit finished. Tables evaluated: {{
      outputs.scan_pinot_segments.vars.tables_scanned_count }}, degraded
      segments: {{ outputs.scan_pinot_segments.vars.bad_segments_count }}."

  - id: export_pinot_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured manifest containing degraded Pinot segments and
      reload instructions.
    format:
      controller_url: "{{ inputs.pinot_controller_url }}"
      tables_scanned: "{{ outputs.scan_pinot_segments.vars.tables_scanned_count }}"
      total_segments: "{{ outputs.scan_pinot_segments.vars.total_segments_audited }}"
      degraded_count: "{{ outputs.scan_pinot_segments.vars.bad_segments_count }}"
      details: "{{ outputs.scan_pinot_segments.vars.bad_segments_details }}"

triggers:
  - id: hourly_pinot_segment_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"

Apache Pinot powers mission-critical user-facing analytics and real-time dashboard applications. Network partitions, deep storage corruption, or server crashes can leave segments in unassigned, ERROR, or consuming dead-letter states where queries return incomplete query results or fail completely. This sentinel interrogates the Apache Pinot Controller REST API, verifies ideal states against external views across offline and real-time tables, and alerts data infrastructure teams on Slack with exact segment reload API commands.

How it works

Scheduled triggering: Executes hourly at minute 0 (0 * * * *) using a cron schedule trigger or runs on-demand via manual execution and API webhooks.
Table enumeration: Queries the Pinot Controller REST API (GET /tables) to discover all registered offline and real-time tables.
Segment state analysis: Interrogates segment debug endpoints (/debug/tables/{tableName}/state) to detect mismatching ideal states, unassigned segments, and consuming errors.
Condition evaluation: An io.kestra.plugin.core.flow.If task evaluates whether bad_segments_count exceeds max_tolerated_bad_segments.
Slack alert dispatch: Sends high-priority incident details via SlackIncomingWebhook listing table names, segment identifiers, and exact reload commands.
Audit trail logging and return: Emits operational execution logs and returns a structured health report containing complete segment diagnostic metadata.

Architecture diagram

flowchart TD
  A[hourly_pinot_segment_audit Schedule Trigger] --> B[scan_pinot_segments Python Script]
  B --> C[check_bad_segments Flowable If]
  C -- Degraded Segments Found --> D[notify_slack_pinot_ops Webhook]
  C -- All Segments Healthy --> E[log_pinot_segment_audit Log]
  D --> E
  E --> F[export_pinot_manifest Return]

Use cases

Real-time analytics reliability: Ensure streaming Kafka ingestion into Pinot real-time tables does not stall due to unassigned or broken consuming segments.
Offline batch ingestion validation: Catch data upload failures after batch Spark or MapReduce segment push jobs before analytical dashboards miss daily metrics.
Cluster rebalance verification: Confirm that all table segments are successfully reassigned to target Pinot servers following cluster scale-down or node upgrades.

Inputs

Name	Type	Default	Description
`pinot_controller_url`	`STRING`	`http://pinot-controller:9000`	Base HTTP endpoint for the Apache Pinot cluster controller REST API.
`table_name_filter`	`STRING`	`all`	Name of the Pinot table to audit, or all to scan all active offline and realtime tables.
`alert_on_bad_segments`	`BOOL`	`true`	Whether to dispatch alerts when unassigned, offline, or errored segments are discovered.
`max_tolerated_bad_segments`	`INT`	`0`	Maximum allowable count of degraded segments before triggering operational alerts.

Expected outputs

{{ outputs.scan_pinot_segments.vars.tables_scanned_count }}: Number of Pinot tables evaluated during the audit.
{{ outputs.scan_pinot_segments.vars.total_segments_audited }}: Consolidated count of segments inspected across all target tables.
{{ outputs.scan_pinot_segments.vars.bad_segments_count }}: Total number of degraded or unassigned segments flagged.
{{ outputs.scan_pinot_segments.vars.bad_segments_details }}: Serialized JSON list of degraded segments with curl reload commands.
{{ outputs.export_pinot_manifest.value }}: Structured dictionary containing the cluster segment health report.

Prerequisites

Network connectivity from Kestra execution workers to the Pinot Controller HTTP port (default 9000).
Optional bearer token or basic authentication credentials if Pinot Controller is secured behind reverse proxies.
Incoming Slack Webhook configured for data platform alerting.

Secrets

PINOT_CONTROLLER_TOKEN: Bearer token or API key for authenticating to the Pinot Controller.
SLACK_WEBHOOK_URL: Slack Incoming Webhook endpoint URL used for sending operational alerts.

Quick start

Define PINOT_CONTROLLER_TOKEN and SLACK_WEBHOOK_URL in your Kestra Secret storage.
Deploy this flow under the company.data namespace.
Configure pinot_controller_url to point to your cluster controller (e.g., http://pinot-ctrl.internal:9000).
Run a manual execution to test table discovery and verify alert routing.

Common pitfalls and troubleshooting

Real-time vs Offline table suffixes: Pinot appends _OFFLINE and _REALTIME to table names internally; ensure API calls handle both table types.
Helix znode state latency: Following cluster maintenance or server reboots, Helix state synchronization may take a few seconds before external view matches ideal state.
Deep storage download timeouts: If segment reload calls fail, verify that Pinot server nodes have direct network access and IAM credentials to the deep storage (S3, GCS, ADLS) backend.

How to extend

Automated segment reload trigger: Add a Kestra task that executes HTTP POST requests directly to /segments/{table}/{segment}/reload to automatically heal unassigned segments.
Kafka consumer lag correlation: Correlate real-time segment consumption errors with Kafka consumer lag metrics to distinguish broker issues from Pinot server faults.
PagerDuty on-call escalation: Escalate degraded segment alerts directly to on-call data platform engineers for business-critical real-time tables.

Apache Pinot Segment Unassigned and Reload Sentinel

How it works

Architecture diagram

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Apache Pinot Segment Unassigned and Reload Sentinel

How it works

Architecture diagram

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Weekly OpenSearch Index Retention with Dated Deletes

OpenSearch ISM Policy Compliance Gate

Hourly OpenSearch Error Spike Gate with Discord Alerting

Daily OpenSearch Error Digest to Slack

Search OpenAlex for Related Scholarly Works

Full OpenSearch Index Export to Internal Storage with Scroll

More Infrastructure Blueprints

Kafka Orphaned Topic Reclaim Guard

Block a breaking Avro schema before a Kafka producer deploys

Kafka Schema Registry Compatibility and Mode Drift Guard

Kafka DLQ Quarantine, Classification, and Safe Replay Gate

Kafka Schema Registry Orphaned Subject Cleanup Guard

Kafka Producer Compression Ratio Sentinel

New to Kestra?