Gate Kafka schema changes with Confluent Schema Registry and Kestra

id: kafka-schema-compatibility-gate
namespace: company.team

inputs:
  - id: subject
    type: STRING
    displayName: Schema Registry subject
    description: The subject the candidate schema is checked against, usually <topic>-value.
    defaults: orders-value

  - id: candidate_schema
    type: JSON
    displayName: Candidate Avro schema
    description: The schema the producer is about to deploy. Checked against the
      latest registered version before anything is written.
    defaults: |
      {
        "type": "record",
        "name": "Order",
        "namespace": "shop",
        "fields": [
          {"name": "order_id", "type": "string"},
          {"name": "amount", "type": "double"},
          {"name": "currency", "type": "string", "default": "USD"}
        ]
      }

  - id: sample_record
    type: JSON
    displayName: Sample record
    description: One record matching the candidate schema. Produced then consumed to
      prove the registered schema actually serializes.
    defaults: |
      {"order_id": "ORD-1042", "amount": 129.99, "currency": "EUR"}

  - id: topic
    type: STRING
    displayName: Kafka topic
    description: Topic the round trip writes to and reads back.
    defaults: orders

  - id: kafka_bootstrap_servers
    type: STRING
    displayName: Kafka bootstrap servers
    description: Comma separated broker list, for example broker-1:9092,broker-2:9092.
    defaults: localhost:9092

  - id: schema_registry_url
    type: STRING
    displayName: Schema Registry URL
    description: Base URL of the Confluent Schema Registry, with no trailing slash.
    defaults: http://localhost:8081

  - id: register_on_pass
    type: BOOL
    displayName: Register when compatible
    description: Register the candidate and run the round trip. Set to false for a
      dry run that only reports the verdict.
    defaults: true

  - id: notify_slack
    type: BOOL
    displayName: Notify Slack when blocked
    description: Post the registry reasons to Slack when the gate blocks. Needs the
      SLACK_WEBHOOK_URL secret.
    defaults: false

triggers:
  - id: on_ci_request
    type: io.kestra.plugin.core.trigger.Webhook
    description: CI calls this webhook with the candidate schema before it ships a
      producer. A FAILED execution is the signal to stop the deploy.
    key: "{{ secret('SCHEMA_GATE_WEBHOOK_KEY') }}"

tasks:
  - id: compatibility_level
    type: io.kestra.plugin.core.http.Request
    description: Read the level the verdict will be measured against. The subject
      may override the global level, so the log records which one applied.
    method: GET
    uri: "{{ inputs.schema_registry_url }}/config/{{ inputs.subject
      }}?defaultToGlobal=true"
    options:
      allowFailed: true

  - id: check_compatibility
    type: io.kestra.plugin.core.http.Request
    description: >
      Ask the registry whether the candidate is compatible with the latest
      registered version. verbose=true is what makes the registry fill in the
      per-field reasons. Without it the response carries the verdict with
      nothing an engineer can act on. The latest alias also makes a first
      registration work: a numeric version against a subject that does not exist
      yet is a version-not-found error instead of a verdict.
    method: POST
    uri: "{{ inputs.schema_registry_url }}/compatibility/subjects/{{ inputs.subject
      }}/versions/latest?verbose=true"
    contentType: application/vnd.schemaregistry.v1+json
    body: "{{ inputs.candidate_schema |
      jq('{schemaType:\"AVRO\",schema:(.|tojson)}') | first | toJson }}"

  - id: verdict
    type: io.kestra.plugin.core.output.OutputValues
    description: Pull the verdict out of the response body so the gate has a named
      value to branch on rather than an HTTP status.
    values:
      level: "{{ outputs.compatibility_level.code == 200 ?
        (outputs.compatibility_level.body | jq('.compatibilityLevel') | first) :
        'unknown' }}"
      is_compatible: "{{ outputs.check_compatibility.body | jq('.is_compatible') | first }}"
      reason_count: "{{ outputs.check_compatibility.body | jq('.messages | length') |
        first }}"
      reasons: "{{ outputs.check_compatibility.body | jq('[.messages[]? |
        select(startswith(\"{errorType\"))] | join(\" | \")') | first }}"

  - id: log_verdict
    type: io.kestra.plugin.core.log.Log
    message: >-
      Subject {{ inputs.subject }} level {{ outputs.verdict.values.level }}:
      HTTP {{ outputs.check_compatibility.code }} with is_compatible={{
      outputs.verdict.values.is_compatible }} and {{
      outputs.verdict.values.reason_count }} message(s). The status code is 200
      either way, so the verdict comes from the body.

  - id: gate
    type: io.kestra.plugin.core.flow.If
    description: >
      The branch reads is_compatible out of the parsed body. An incompatible
      schema is answered with HTTP 200, so a gate written against
      outputs.check_compatibility.code would pass every breaking change.
      Register is the mirror image: that endpoint answers 409, so there the
      status code is the right signal and the default allowFailed handles it.
    condition: "{{ outputs.check_compatibility.body | jq('.is_compatible') | first }}"
    then:
      - id: deploy_decision
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.register_on_pass }}"
        then:
          - id: register_schema
            type: io.kestra.plugin.core.http.Request
            description: Register the candidate. This endpoint answers 409 on an
              incompatible schema, so the default allowFailed false already
              stops the run.
            method: POST
            uri: "{{ inputs.schema_registry_url }}/subjects/{{ inputs.subject }}/versions"
            contentType: application/vnd.schemaregistry.v1+json
            body: "{{ inputs.candidate_schema |
              jq('{schemaType:\"AVRO\",schema:(.|tojson)}') | first | toJson }}"

          - id: create_topic
            type: io.kestra.plugin.kafka.TopicCreate
            description: Make sure the topic exists before the round trip. Existing topics
              are left alone.
            topic: "{{ inputs.topic }}"
            partitions: 1
            replicationFactor: 1
            ifNotExists: true
            properties:
              bootstrap.servers: "{{ inputs.kafka_bootstrap_servers }}"

          - id: produce_record
            type: io.kestra.plugin.kafka.Produce
            description: >
              Write one record with the schema that was just registered.
              transactional is off because a single broker cluster rejects the
              generated transactional id under the default
              transaction.state.log.replication.factor.
            topic: "{{ inputs.topic }}"
            transactional: false
            keySerializer: STRING
            valueSerializer: AVRO
            valueAvroSchema: "{{ inputs.candidate_schema | toJson }}"
            serdeProperties:
              schema.registry.url: "{{ inputs.schema_registry_url }}"
            properties:
              bootstrap.servers: "{{ inputs.kafka_bootstrap_servers }}"
            # One rendered JSON object, so the value reaches the Avro serializer as a map and not as a string.
            from: "{{ {'key': execution.id, 'value': inputs.sample_record} | toJson }}"

          - id: consume_record
            type: io.kestra.plugin.kafka.Consume
            description: Read the record back. The Avro deserializer resolves the schema id
              from the registry, so a successful read proves the registered
              version is usable.
            topic: "{{ inputs.topic }}"
            groupId: kestra-schema-gate
            keyDeserializer: STRING
            valueDeserializer: AVRO
            maxRecords: 1
            pollDuration: PT15S
            serdeProperties:
              schema.registry.url: "{{ inputs.schema_registry_url }}"
            properties:
              bootstrap.servers: "{{ inputs.kafka_bootstrap_servers }}"
              auto.offset.reset: earliest

          - id: readable_record
            type: io.kestra.plugin.serdes.json.IonToJson
            description: Turn the consumed Ion file into JSON so the round trip is readable
              in the logs.
            from: "{{ outputs.consume_record.uri }}"

          - id: round_trip_report
            type: io.kestra.plugin.core.log.Log
            message: >-
              Registered schema id {{ outputs.register_schema.body | jq('.id') |
              first }} as a new version of {{ inputs.subject }}. Produced 1
              record to {{ inputs.topic }} then read {{
              outputs.consume_record.messagesCount }} back through the registry:
              {{ read(outputs.readable_record.uri) }}
        else:
          - id: dry_run_report
            type: io.kestra.plugin.core.log.Log
            message: >-
              Compatible with the latest version of {{ inputs.subject }} under
              {{ outputs.verdict.values.level }}. register_on_pass is false, so
              nothing was registered and no record was produced.
    else:
      - id: log_blocked
        type: io.kestra.plugin.core.log.Log
        description: The registry's own wording, kept verbatim so an engineer can find
          the field without opening the registry.
        message: >-
          Deploy blocked. {{ inputs.subject }} is {{
          outputs.verdict.values.level }} and the candidate breaks it. Registry
          reasons: {{ outputs.verdict.values.reasons }}

      - id: alert_blocked
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: slack_blocked
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "Schema gate blocked a deploy. Subject {{ inputs.subject }} is {{ outputs.verdict.values.level }}. {{ outputs.verdict.values.reasons }} Execution {{ execution.id }}."
              }

      - id: block_deploy
        type: io.kestra.plugin.core.execution.Fail
        description: End the run as FAILED so the CI job that called the webhook stops.
          Nothing was registered and nothing was produced.
        errorMessage: "{{ inputs.subject }} is {{ outputs.verdict.values.level }} and
          the candidate is not compatible with its latest version. {{
          outputs.verdict.values.reasons }}"

A breaking schema change is cheap to catch before a producer ships and expensive once it is on the topic. This flow puts the decision in front of the deploy. CI posts the candidate Avro schema to the webhook, the registry answers whether it is compatible with the subject's latest version. A FAILED execution is what stops the pipeline. When the answer is yes the flow registers the schema, writes one record with it, then reads that record back through the registry so the new version is proven usable rather than merely accepted.

The reason this is worth a flow rather than one curl is that the two registry endpoints it needs report failure in opposite ways:

POST /compatibility/subjects/{subject}/versions/latest answers HTTP 200 when the candidate is incompatible, with is_compatible: false in the body. A gate written against the status code passes every breaking change, silently. So gate branches on the parsed body field instead.
POST /subjects/{subject}/versions answers 409 Conflict on the same schema. There the status code is the right signal, so register_schema leaves allowFailed at its default false and lets the HTTP failure stop the run.

?verbose=true is load bearing, not cosmetic. The registry only fills in messages when it is set. Without it the flow can still block a deploy but cannot say which field broke, so the engineer gets a red pipeline with nothing to fix. With it the log carries the registry's own wording, for example The field 'currency' at path '/fields/1' in the new schema has no default value and is missing in the old schema.

The path uses the latest alias rather than a version number. That makes a brand new subject work with no special casing: the first candidate comes back compatible. A numeric version against a subject that does not exist is a version-not-found error instead of a verdict.

This blueprint was created by zkasuran.

How it works

compatibility_level (io.kestra.plugin.core.http.Request) reads GET /config/{subject}?defaultToGlobal=true with allowFailed: true, so the log records whether the verdict came from the global level or a subject override. The registry applies the level itself, so this task is for the record rather than for the decision.
check_compatibility posts the candidate to /compatibility/subjects/{subject}/versions/latest?verbose=true. The registry wants the schema as a JSON string inside the request body, which is what jq('{schemaType:"AVRO",schema:(.|tojson)}') builds. Pasting the schema in as a nested object is a 42201 that reads like a flow bug.
verdict (io.kestra.plugin.core.output.OutputValues) names the three things the rest of the flow needs: the level, is_compatible plus the reasons joined into one line.
gate (io.kestra.plugin.core.flow.If) branches on is_compatible from the body.
Compatible, with register_on_pass on: register_schema registers the version, create_topic (io.kestra.plugin.kafka.TopicCreate) makes sure the topic exists, produce_record (io.kestra.plugin.kafka.Produce) writes one Avro record, consume_record (io.kestra.plugin.kafka.Consume) reads it back, readable_record converts the Ion output to JSON, then round_trip_report logs the new schema id with the record.
Incompatible: log_blocked prints the registry reasons, alert_blocked posts them to Slack when notify_slack is on, then block_deploy (io.kestra.plugin.core.execution.Fail) ends the run as FAILED. Nothing is registered and nothing is produced.

Inputs

subject (STRING, default orders-value): the subject to check against.
candidate_schema (JSON): the schema about to be deployed.
sample_record (JSON): one record matching it, used for the round trip.
topic (STRING, default orders): topic for the round trip.
kafka_bootstrap_servers (STRING): broker list.
schema_registry_url (STRING): registry base URL with no trailing slash.
register_on_pass (BOOL, default true): set to false for a verdict-only dry run.
notify_slack (BOOL, default false): Slack alert on a block.

Prerequisites

A Kafka cluster plus a Confluent Schema Registry reachable from Kestra. A registry keeps its state in a Kafka topic, so it cannot run without a broker. This is why the blueprint is not marked as a demo.
Write permission on the subject for the registry user, plus produce and consume permission on the topic.
On a single broker test cluster, keep transactional: false on produce_record. The default transactional producer generates a transactional id that the default transaction.state.log.replication.factor of 3 cannot satisfy.
valueAvroSchema is supplied on produce_record alongside serdeProperties. The Avro serializer resolves its schema before any registry block is applied, so it needs the schema text as well as the registry URL.

Secrets

SCHEMA_GATE_WEBHOOK_KEY: the webhook key CI calls the flow with.
SLACK_WEBHOOK_URL: only needed when notify_slack is true.

Move kafka_bootstrap_servers plus schema_registry_url to secret() values if your registry sits behind credentials.

Quick start

Add the flow, then set kafka_bootstrap_servers plus schema_registry_url to your own endpoints.
Run it with the defaults against a subject that does not exist yet. The candidate comes back compatible, gets registered as version 1, then one record round trips.
Run it again with a candidate that drops amount or adds a field with no default. The gate blocks, the log names the field, the execution ends FAILED.
Point your CI at the webhook: curl -X POST -d '{"candidate_schema": ...}' <kestra>/api/v1/executions/webhook/company.team/kafka-schema-compatibility-gate/<key>. A non-zero result is the stop signal.

Expected outputs

outputs.verdict.values.is_compatible: true or false, read from the body rather than the status code.
outputs.verdict.values.reasons: the registry's errorType entries joined into one line, empty when compatible.
outputs.register_schema.body: {"id": <n>}, the global schema id of the newly registered version.
outputs.consume_record.messagesCount: 1 on a successful round trip.
A FAILED execution with the registry wording in the error message when the gate blocks.

How to extend

Swap schemaType to PROTOBUF or JSON to gate those contracts with the same two endpoints.
Check several subjects in one run by wrapping the gate in io.kestra.plugin.core.flow.Loop over a list of subject plus schema pairs.
Raise the level to FULL_TRANSITIVE for the subject with PUT /config/{subject} before the check, so consumers that lag behind are covered too.
Have the else branch open a pull request comment with the reasons instead of only failing.
Add io.kestra.plugin.core.trigger.Schedule beside the webhook to re-check the schema in git every night, which catches someone lowering the subject's compatibility level.

Block a breaking Avro schema before a Kafka producer deploys

How it works

Inputs

Prerequisites

Secrets

Quick start

Expected outputs

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Block a breaking Avro schema before a Kafka producer deploys

How it works

Inputs

Prerequisites

Secrets

Quick start

Expected outputs

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Invoice Math Discrepancy Gate with In-Memory DuckDB Reconciliation

AI Data Quality Anomaly Advisor and SQL Remediation Generator

Glue ETL, Athena Validation, DynamoDB Cache

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

Automate Cross-Cloud ETL from Azure Blob Storage to BigQuery with dbt Cloud

Autoscale EMR Serverless Spark on a CloudWatch Backlog

More Infrastructure Blueprints

Automatically Review GitHub Pull Requests with OpenAI GPT-4o and Post Comments

AI Incident Post-Mortem and Root Cause Generator

AWS Cost Explorer Spend Anomaly Guard

GitOps CI/CD to AWS EKS with Kaniko, ECR and Discord Alerts

AI Security Incident Threat Advisor and MITRE ATT&CK Containment Generator

New to Kestra?