Apache Pinot Real-Time Ingestion Freshness Sentinel

A stalled Kafka (or other stream) consumer feeding an Apache Pinot real-time table degrades silently - queries still return results, just increasingly stale ones, with no obvious client-side error. This blueprint reads the table's own newest row against Pinot's own now() scalar function to measure that lag directly, classifies the result into a three-state risk level, and alerts on anything but a clean bill of health.

How it works

1. periodic_freshness_check (io.kestra.plugin.core.trigger.Schedule) runs every 10 minutes, explicitly passing pinot_broker_url, pinot_table, timestamp_column, and max_lag_minutes through its own inputs: block so the flow is self-contained when the trigger fires. Shipped disabled so a manual run can be verified first.
2. probe_table_freshness (io.kestra.plugin.jdbc.pinot.Query, fetchType: FETCH_ONE) runs a single SQL statement against the Pinot broker, returning row.row_count (total rows in the table), row.latest_event_time (max of timestamp_column), and row.lag_ms (Pinot's own now() minus that max, in milliseconds).
3. evaluate_row_presence (io.kestra.plugin.core.flow.If) branches on whether row_count > 0.
   • else (EMPTY_TABLE): alert_empty_table reports that the table has never received a row.
   • then (rows exist): nested evaluate_freshness (io.kestra.plugin.core.flow.If) branches on lag_ms > max_lag_minutes * 60000.
     • then (STALE): alert_stale_ingestion reports the lag, threshold, row count, and latest event time to Slack.
     • else (HEALTHY): log_freshness_healthy records a clean check.
4. log_audit_complete always runs last, printing row_count and lag_ms regardless of which branch fired.
5. The errors block alerts Slack separately if the flow itself fails outright - an unreachable broker or bad credentials must not read as EMPTY_TABLE or HEALTHY.

What you get

• A three-state answer (HEALTHY / STALE / EMPTY_TABLE) computed entirely inside Pinot's own query engine via now(), with no external monitoring agent or sidecar.
• A distinct signal for "no data ever arrived" versus "data arrived but stopped", which a single boolean threshold check would conflate.
• One Slack alert carrying the exact lag, row count, and latest event time, instead of running the probe query by hand during an incident.
• A single generic io.kestra.plugin.jdbc.pinot.Query task driving the aggregate probe, reusable against any table via the pinot_table and timestamp_column inputs.

Who it's for

• Platform and data engineering teams running Apache Pinot real-time or hybrid tables who need an early signal when a stream consumer stalls.
• On-call engineers who want a single Slack alert with lag numbers instead of querying Pinot by hand during an incident.
• Anyone evaluating the Pinot JDBC plugin who wants a worked example of Query driving a single-statement aggregate probe, not just a row extract.

Why orchestrate this with Kestra

A one-off Pinot SQL query answers one moment in time; it does not decide a cadence, classify the result into a three-state risk level, or notify anyone. Kestra supplies the schedule, the HEALTHY/STALE/EMPTY_TABLE classification as a first-class branch, and an execution history that shows exactly when ingestion degraded and for how long.

Prerequisites

• A reachable Apache Pinot broker (update pinot_broker_url).
• A target table with a known epoch-millisecond timestamp column (update pinot_table and timestamp_column).
• A Slack incoming webhook for alerts.

Local testing:

docker pull apachepinot/pinot:latest
docker run -d --name pinot-quickstart -p 9000:9000 apachepinot/pinot:latest QuickStart -type batch
docker network connect YOUR_KESTRA_NETWORK pinot-quickstart

Run docker network ls first to find the active Kestra network name. QuickStart -type batch loads the sample airlineStats table used as this blueprint's default, on host port 9000, which does not collide with Kestra's own UI on port 8080.

Secrets

• PINOT_USERNAME / PINOT_PASSWORD: credentials used by io.kestra.plugin.jdbc.pinot.Query. The open-source QuickStart image enforces no authentication, so these can be left as empty secrets for local testing; set them for any secured broker.
• SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_stale_ingestion, alert_empty_table, and the errors block.
• In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded - for example SECRET_SLACK_WEBHOOK_URL=$(echo -n '<your-webhook-url>' | base64) - and read back in flows with {{ secret('NAME') }}.

Inputs

• pinot_broker_url (STRING, default jdbc:pinot://localhost:9000): JDBC URL of the Pinot broker.
• pinot_table (STRING, default airlineStats): table audited for freshness.
• timestamp_column (STRING, default __time): epoch-millisecond event time column used for the lag calculation.
• max_lag_minutes (SELECT: "5", "15", "30", "60", "120"; default "30"): lag ceiling before a non-empty table counts as STALE.

Outputs

• outputs.probe_table_freshness.row.row_count: total rows in pinot_table at probe time.
• outputs.probe_table_freshness.row.latest_event_time: max value of timestamp_column.
• outputs.probe_table_freshness.row.lag_ms: Pinot's now() minus latest_event_time, in milliseconds.

Quick start

1. Start Pinot locally with the command above; it loads the sample airlineStats table.
2. Add the PINOT_USERNAME, PINOT_PASSWORD (can be empty for QuickStart), and SLACK_WEBHOOK_URL secrets.
3. Run the flow manually with the defaults. Because airlineStats is a static batch QuickStart dataset rather than a live stream, lag_ms will be very large and alert_stale_ingestion is expected to fire on the very first run - see Pitfalls.
4. Point pinot_table and timestamp_column at a real-time or hybrid table with an active stream consumer to see log_freshness_healthy fire instead.
5. Enable periodic_freshness_check once you trust the check against your own table.

How to extend

• Loop over several pinot_table values with io.kestra.plugin.core.flow.Loop to audit every real-time table in a cluster in one run.
• Add a second probe comparing row_count across consecutive runs (via a Kestra KV store) to detect a consumer that is connected but no longer advancing, which lag_ms alone would not catch if timestamp_column is not monotonic.
• Route EMPTY_TABLE to a different Slack channel than STALE, since no data ever arriving is typically a configuration problem, not a transient stall.

Pitfalls

• airlineStats is a static batch QuickStart dataset, not a live stream. Running this blueprint against the defaults reports STALE on every run because its __time values are fixed at load time - point pinot_table at a genuinely real-time table to see HEALTHY.
• __time is taken from Kestra's own io.kestra.plugin.jdbc.pinot.Query plugin documentation example, not confirmed as a universal reserved alias for every Pinot deployment. If it does not resolve against your schema, set timestamp_column to your table's actual timeColumnName.
• There is no dedicated "ingestion lag" or "consumer status" task in this plugin. The freshness probe goes through the same generic io.kestra.plugin.jdbc.pinot.Query task used for any other Pinot read.
• io.kestra.plugin.jdbc.pinot.Query supports only a single SQL statement. Use io.kestra.plugin.jdbc.pinot.Queries instead for flows that need more than one statement per task.
• now() returns epoch milliseconds, per Apache Pinot's own datetime functions documentation. If timestamp_column is stored in a different unit (seconds, days), lag_ms and max_lag_minutes will be wrong until the SQL is adjusted.
• This blueprint is UNTESTED against a live Pinot cluster. The now() function and its epoch-millisecond return type are taken verbatim from Apache Pinot's own datetime functions documentation, and the io.kestra.plugin.jdbc.pinot.Query properties and outputs are taken from the plugin's own source on main - but no Docker container or Kestra engine was run to execute this flow end to end.

Links

• Kestra Apache Pinot plugin documentation
• Apache Pinot datetime functions documentation (now())
• Kestra secrets (OSS vs Enterprise)