Karapace Schema Compatibility & Drift Gate

id: karapace-schema-compatibility-and-drift-compliance-gate
namespace: company.platform

description: "Audits Karapace Schema Registry compatibility modes, detects
  breaking schema drift, dispatches actionable Slack incident alerts, and
  automatically enforces enterprise backward compatibility policies."

inputs:
  - id: karapace_host
    type: STRING
    defaults: "localhost"
    description: "Hostname or IP address of the Karapace Schema Registry instance."

  - id: karapace_port
    type: INT
    defaults: 8081
    description: "Port number for the Karapace HTTP REST API (default: 8081)."

  - id: enforced_compatibility_level
    type: STRING
    defaults: "BACKWARD_TRANSITIVE"
    description: "Required schema compatibility level to prevent breaking downstream
      consumers (options: BACKWARD, BACKWARD_TRANSITIVE, FULL,
      FULL_TRANSITIVE)."

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: "When enabled and dry_run is false, automatically invokes PUT
      /config to restore global schema compatibility to the enforced standard."

  - id: dry_run
    type: BOOL
    defaults: true
    description: "Safety gate: when true, audits and emits Slack alerts without
      mutating Karapace registry configuration."

  - id: slack_channel
    type: STRING
    defaults: "#data-governance-alerts"
    description: "Target Slack channel for schema drift violations and compliance
      breach alerts."

triggers:
  - id: periodic_schema_compliance_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"
    disabled: true

tasks:
  - id: probe_global_config
    type: io.kestra.plugin.core.http.Request
    description: "Polls Karapace Schema Registry /config endpoint to inspect the
      current active global compatibility mode."
    uri: "http://{{ inputs.karapace_host }}:{{ inputs.karapace_port }}/config"
    method: GET
    options:
      timeout:
        readIdleTimeout: PT15S

  - id: probe_registered_subjects
    type: io.kestra.plugin.core.http.Request
    description: "Polls Karapace /subjects endpoint to enumerate all registered
      event streaming schema topics."
    uri: "http://{{ inputs.karapace_host }}:{{ inputs.karapace_port }}/subjects"
    method: GET
    options:
      timeout:
        readIdleTimeout: PT15S

  - id: classify_schema_compliance
    type: io.kestra.plugin.core.flow.Switch
    description: "Evaluates Karapace registry responses to classify compliance into
      COMPLIANCE_DRIFT_DETECTED, NO_SUBJECTS_REGISTERED, or COMPLIANT."
    value: >-
      {% if '"NONE"' in outputs.probe_global_config.body or '"FORWARD"' in
      outputs.probe_global_config.body %} COMPLIANCE_DRIFT_DETECTED {% elif
      (fromJson(outputs.probe_registered_subjects.body) ?? []) | length == 0 %}
      NO_SUBJECTS_REGISTERED {% else %} COMPLIANT {% endif %}
    cases:
      COMPLIANCE_DRIFT_DETECTED:
        - id: alert_drift_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Dispatches high-priority Slack incident alert when registry
            compatibility level degrades below enterprise policy."
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": ":rotating_light: *Karapace Schema Compliance Breach: Incompatible Evolution Mode*",
              "blocks": [
                {
                  "type": "section",
                  "text": {
                    "type": "mrkdwn",
                    "text": ":rotating_light: *Karapace Schema Compliance Breach: Incompatible Evolution Mode*\n*Registry:* `http://{{ inputs.karapace_host }}:{{ inputs.karapace_port }}`\n*Current Level:* `{{ fromJson(outputs.probe_global_config.body).compatibilityLevel ?? 'UNKNOWN' }}`\n*Enforced Policy:* `{{ inputs.enforced_compatibility_level }}`\n*Impact:* Relaxed compatibility allows breaking schema changes, risking downstream Kafka consumer crashes."
                  }
                }
              ]
            }

        - id: check_auto_remediation
          type: io.kestra.plugin.core.flow.If
          description: "Guards automated configuration updates against accidental
            invocation during dry-run evaluations."
          condition: "{{ inputs.dry_run == false and inputs.auto_remediate == true }}"
          then:
            - id: enforce_compatibility_policy
              type: io.kestra.plugin.core.http.Request
              description: "Updates Karapace global configuration via PUT /config to restore
                enforced compatibility level."
              uri: "http://{{ inputs.karapace_host }}:{{ inputs.karapace_port }}/config"
              method: PUT
              headers:
                Content-Type: "application/vnd.schemaregistry.v1+json"
              body: |
                {
                  "compatibility": "{{ inputs.enforced_compatibility_level }}"
                }
            - id: log_remediation_success
              type: io.kestra.plugin.core.log.Log
              message: "Successfully restored Karapace schema compatibility policy to {{
                inputs.enforced_compatibility_level }}."
          else:
            - id: log_dry_run_remediation
              type: io.kestra.plugin.core.log.Log
              message: "Dry-run active or auto_remediate disabled: schema drift diagnostic
                logged without altering registry state."

      NO_SUBJECTS_REGISTERED:
        - id: alert_empty_registry
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Dispatches warning notification if Karapace reports zero
            registered schema subjects in production."
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": ":warning: *Karapace Schema Registry Warning: Zero Subjects Registered*",
              "blocks": [
                {
                  "type": "section",
                  "text": {
                    "type": "mrkdwn",
                    "text": ":warning: *Karapace Schema Registry Warning: Zero Subjects Registered*\n*Registry:* `http://{{ inputs.karapace_host }}:{{ inputs.karapace_port }}`\n*Status:* No active schema subjects detected. The registry backend may be uninitialized or subject data dropped."
                  }
                }
              ]
            }

      COMPLIANT:
        - id: log_compliant_state
          type: io.kestra.plugin.core.log.Log
          description: "Records compliance verification when compatibility policy matches
            enterprise requirements."
          message: "Karapace Schema Registry compliant on {{ inputs.karapace_host }}:{{
            inputs.karapace_port }}. Compatibility: {{
            fromJson(outputs.probe_global_config.body).compatibilityLevel ??
            'NOMINAL' }} | Total Subjects: {{
            (fromJson(outputs.probe_registered_subjects.body) ?? []) | length
            }}."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: "Emits immutable audit lineage capturing audited subject counts,
      compatibility level, and evaluation outcome."
    message: "Karapace sentinel audit completed for http://{{ inputs.karapace_host
      }}:{{ inputs.karapace_port }}. Status: {{
      outputs.classify_schema_compliance.value | trim }} | Evaluated Subjects:
      {{ (fromJson(outputs.probe_registered_subjects.body) ?? []) | length }} |
      Execution: {{ execution.id }}"

errors:
  - id: alert_karapace_probe_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Dispatches emergency notification if Karapace Schema Registry REST
      API is unreachable or timing out."
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": ":rotating_light: *Karapace Schema Registry Probe Failure*",
        "blocks": [
          {
            "type": "section",
            "text": {
              "type": "mrkdwn",
              "text": ":rotating_light: *Karapace Schema Registry Probe Failure*\nFailed to connect to Karapace server at `http://{{ inputs.karapace_host }}:{{ inputs.karapace_port }}`. The container may be offline, network partitioned, or port misconfigured."
            }
          }
        ]
      }

Aiven Karapace is the leading 100% open-source, Apache 2.0 licensed drop-in replacement for Kafka Schema Registry and REST Proxy. In enterprise event-driven architectures, schema registries enforce contracts between producers and consumers. A common operational failure occurs when registry compatibility modes are accidentally relaxed to NONE or FORWARD during ad-hoc testing or rogue CI/CD deployments. When compatibility guards are dropped, producers can publish backward-incompatible schemas that cause downstream streaming pipelines (such as Apache Flink, Spark Streaming, and Kafka consumer microservices) to fail on deserialization. This blueprint continuously audits Karapace global and subject compatibility settings, detects policy drift, alerts data governance teams via Slack, and provides guarded automated remediation to re-enforce strict compatibility rules.

How it works

periodic_schema_compliance_audit (io.kestra.plugin.core.trigger.Schedule) initiates scheduled checks every 15 minutes. Shipped disabled so initial manual verification can be performed first.
probe_global_config (io.kestra.plugin.core.http.Request) invokes GET /config on the Karapace REST API to inspect active global compatibility settings.
probe_registered_subjects (io.kestra.plugin.core.http.Request) invokes GET /subjects to retrieve all registered topic schema subjects.
classify_schema_compliance (io.kestra.plugin.core.flow.Switch) analyzes probe outputs:
- COMPLIANCE_DRIFT_DETECTED: Fires if global compatibility level is set to NONE or FORWARD, dispatching a Slack alert via io.kestra.plugin.slack.notifications.SlackIncomingWebhook. If auto_remediate is enabled and dry_run is false, enforce_compatibility_policy executes PUT /config to restore BACKWARD_TRANSITIVE compliance.
- NO_SUBJECTS_REGISTERED: Warns the engineering team via Slack if zero subjects are found, highlighting potential storage or connectivity loss.
- COMPLIANT: Logs healthy schema contract enforcement and total subject counts.
log_audit_summary (io.kestra.plugin.core.log.Log) records an immutable audit trail documenting evaluated subject counts and compliance outcome.
Global errors handler catches connection timeouts or unreachable Karapace endpoints, dispatching an immediate incident notification to Slack.

What you get

Automated continuous surveillance of Kafka schema evolution policies across all registered event topics.
Early detection of breaking schema drift before incompatible payloads reach production streaming consumers.
Guarded automated remediation that safely restores BACKWARD_TRANSITIVE compatibility via the Karapace REST API.
Zero unexpected state changes: dry_run defaults to true to ensure full evaluation safety during onboarding.

Who it's for

Data Platform Engineers and Event Streaming Architects managing Apache Kafka and Karapace clusters.
Data Governance Teams responsible for event contract stability and schema lifecycle management.
SREs and Platform Teams maintaining real-time stream processing pipelines and preventing deserialization outages.

Why orchestrate this with Kestra

Karapace provides REST endpoints but lacks automated scheduling, alerting, or drift remediation mechanics. Standalone shell scripts lack secret masking, structured retries, execution timelines, and auditable governance. Kestra delivers a declarative, version-controlled YAML blueprint with native scheduling, secure credential handling, conditional routing, and complete execution observability.

Prerequisites

A running Karapace Schema Registry instance (version 3.0+) accessible over HTTP. Karapace publishes an official open-source Docker image (ghcr.io/aiven/karapace:latest) requiring no license or account.
An incoming Slack webhook URL for operational notifications.

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook URL for dispatching schema compliance alerts.

Quick start

Start a local Karapace Schema Registry container:

docker run -d --name karapace-test -p 8081:8081 -e KARAPACE_PORT=8081 ghcr.io/aiven/karapace:latest

Configure the SLACK_WEBHOOK_URL secret in your Kestra namespace.
Import this blueprint into Kestra and trigger an execution with default inputs (dry_run: true).
Verify that log_compliant_state logs nominal compatibility.

Test drift detection by temporarily relaxing the compatibility mode:

curl -X PUT http://localhost:8081/config -H "Content-Type: application/vnd.schemaregistry.v1+json" -d '{"compatibility": "NONE"}'

Re-run the flow to observe alert_drift_slack firing.
Enable periodic_schema_compliance_audit to activate recurring 15-minute surveillance.

How to extend

Add per-subject compatibility iteration using io.kestra.plugin.core.flow.EachSequential to audit specific tier-1 subjects (orders-value, payments-value).
Integrate with PagerDuty via io.kestra.plugin.pagerduty.notifications.PagerDutyAlert for tier-1 streaming pipeline schema drift.
Push schema version metrics to Prometheus via io.kestra.plugin.prometheus for tracking schema evolution velocity over time.

Karapace Schema Registry Subject Compatibility & Drift Compliance Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Karapace Schema Registry Subject Compatibility & Drift Compliance Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Write-Audit-Publish (WAP) and Atomic Merge Gate for Apache Iceberg with Project Nessie

Object Storage Integrity and Streaming Replica Parity Audit

Detect and Resume Stuck Paused NSQ Channels with Accumulating Backlog

Stop a Bad Monthly Load with SLA Assertions on NYC Taxi Data in DuckDB

NFS New File Extension Compliance Gate

Automated OpenLineage Metadata Extraction and Marquez Governance Pipeline

New to Kestra?