AsyncAPI Breaking Change Gates for Event Streams

id: asyncapi-contract-breaking-change-gate
namespace: company.team
description: |
  Event streams ship breaking message schema changes with no gate because
  REST has OpenAPI checks and event contracts have nothing. A keyed webhook
  accepts the candidate AsyncAPI document from CI, a WorkingDirectory
  materializes it beside the blessed baseline, and a Python comparison extracts
  every channel operation payload and diffs required fields and properties
  against the baseline. Removed channels or required fields alert Slack through
  their own branch, an unreadable candidate escalates separately, and the
  verdict, the breaking changes and the channel count are published as outputs.

triggers:
  - id: post_candidate_spec
    type: io.kestra.plugin.core.trigger.Webhook
    description: CI POSTs the candidate AsyncAPI document here as a raw YAML body
      once the event service branch is built.
    key: "{{ secret('ASYNCAPI_WEBHOOK_KEY') }}"
  - id: nightly_contract_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Fallback net, a nightly contract check at 07:00 UTC, shipped
      disabled until an operator switches it on.
    cron: "0 7 * * *"
    disabled: true

inputs:
  - id: candidate_spec
    type: STRING
    defaults: |
      asyncapi: 2.6.0
      info:
        title: Orders Event API
        version: 1.5.0
        description: Candidate contract CI posted after the orders service refactor.
      channels:
        order/created:
          publish:
            message:
              payload:
                type: object
                required:
                  - order_id
                  - customer_id
                  - total_cents
                properties:
                  order_id:
                    type: string
                  customer_id:
                    type: string
                  total_cents:
                    type: integer
        order/shipped:
          publish:
            message:
              payload:
                type: object
                required:
                  - order_id
                  - shipped_at
                properties:
                  order_id:
                    type: string
                  shipped_at:
                    type: string
        order/refunded:
          publish:
            message:
              payload:
                type: object
                required:
                  - order_id
                  - refund_reason
                properties:
                  order_id:
                    type: string
                  refund_reason:
                    type: string
    description: Candidate AsyncAPI document CI posts, the default fixture drops the
      currency required field and removes the order cancelled channel.
  - id: service_name
    type: STRING
    defaults: orders-events-api
    description: Logical service the candidate contract belongs to, used in the Slack alert.
  - id: ci_build_url
    type: STRING
    defaults: https://ci.example.com/blueprints/orders-events/1042
    description: Build or pipeline URL that produced the candidate contract, linked
      from the Slack alert.

tasks:
  # --- 1. contract grade ---
  - id: spec_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the shell step that writes both specs and
      the Python step that compares them read the same two files.
    tasks:
      - id: write_specs
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write the blessed baseline with one heredoc and the posted or
          default candidate with a second, then size both files for the log.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: python:3.12-slim
        commands:
          - |
            set -u
            cat > baseline.yaml <<'BASELINE_EOF'
            asyncapi: 2.6.0
            info:
              title: Orders Event API
              version: 1.4.0
              description: Blessed contract every consumer builds against.
            channels:
              order/created:
                publish:
                  message:
                    payload:
                      type: object
                      required:
                        - order_id
                        - customer_id
                        - currency
                        - total_cents
                      properties:
                        order_id:
                          type: string
                        customer_id:
                          type: string
                        currency:
                          type: string
                        total_cents:
                          type: integer
              order/shipped:
                publish:
                  message:
                    payload:
                      type: object
                      required:
                        - order_id
                        - shipped_at
                      properties:
                        order_id:
                          type: string
                        shipped_at:
                          type: string
              order/cancelled:
                publish:
                  message:
                    payload:
                      type: object
                      required:
                        - order_id
                        - reason
                      properties:
                        order_id:
                          type: string
                        reason:
                          type: string
            BASELINE_EOF

            cat > candidate.yaml <<'CANDIDATE_EOF'
            {{ trigger.body ?? inputs.candidate_spec }}
            CANDIDATE_EOF

            echo "baseline bytes $(wc -c < baseline.yaml) candidate bytes $(wc -c < candidate.yaml)"

      - id: compare_specs
        type: io.kestra.plugin.scripts.python.Script
        description: Load both specs, extract every publish and subscribe payload
          required field and property per channel, diff the candidate against
          the baseline, and emit the compatibility verdict through the outputs
          protocol.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        beforeCommands:
          - pip install --quiet pyyaml
        script: |
          import json
          import os
          import yaml

          BASELINE_PATH = "baseline.yaml"
          CANDIDATE_PATH = "candidate.yaml"

          def load_spec(path, label):
              try:
                  if not os.path.exists(path) or os.path.getsize(path) == 0:
                      return None, label + " file is missing or empty"
                  with open(path, "r", encoding="utf-8") as handle:
                      document = yaml.safe_load(handle)
                  if not isinstance(document, dict):
                      return None, label + " is not a YAML mapping"
                  return document, ""
              except yaml.YAMLError:
                  return None, label + " is not valid YAML"

          def operations(spec):
              found = {}
              channels = spec.get("channels")
              if not isinstance(channels, dict):
                  channels = {}
              for name, channel in channels.items():
                  if not isinstance(channel, dict):
                      continue
                  entry = {}
                  for direction in ("publish", "subscribe"):
                      operation = channel.get(direction)
                      if not isinstance(operation, dict):
                          continue
                      message = operation.get("message")
                      payload = message.get("payload") if isinstance(message, dict) else None
                      required, properties = [], []
                      if isinstance(payload, dict):
                          raw_required = payload.get("required")
                          if isinstance(raw_required, list):
                              required = [str(field) for field in raw_required]
                          raw_properties = payload.get("properties")
                          if isinstance(raw_properties, dict):
                              properties = [str(field) for field in raw_properties]
                      entry[direction] = {"required": required, "properties": properties}
                  if entry:
                      found[str(name)] = entry
              return found

          baseline, baseline_problem = load_spec(BASELINE_PATH, "baseline spec")
          candidate, candidate_problem = load_spec(CANDIDATE_PATH, "candidate spec")

          verdict = "compatible"
          error = ""
          breaking = []
          additions = []
          channels_checked = 0
          baseline_version = ""
          candidate_version = ""

          if baseline is None:
              verdict = "error"
              error = baseline_problem
          elif candidate is None:
              verdict = "error"
              error = candidate_problem
          else:
              baseline_version = str((baseline.get("info") or {}).get("version") or "")
              candidate_version = str((candidate.get("info") or {}).get("version") or "")
              base_ops = operations(baseline)
              cand_ops = operations(candidate)
              channels_checked = len(base_ops)
              for name in base_ops:
                  if name not in cand_ops:
                      breaking.append("removed channel " + name)
                      continue
                  for direction, base_detail in base_ops[name].items():
                      if direction not in cand_ops[name]:
                          breaking.append("removed " + direction + " operation on " + name)
                          continue
                      cand_detail = cand_ops[name][direction]
                      flagged = set()
                      for field in base_detail["required"]:
                          if field not in cand_detail["required"]:
                              breaking.append("removed required field " + name + "." + direction + "." + field)
                              flagged.add(field)
                      for field in base_detail["properties"]:
                          if field not in cand_detail["properties"] and field not in flagged:
                              breaking.append("removed property " + name + "." + direction + "." + field)
              for name in cand_ops:
                  if name not in base_ops:
                      additions.append("added channel " + name)
              if breaking:
                  verdict = "breaking"

          payload = {
              "verdict": verdict,
              "breaking_changes": breaking,
              "breaking_count": len(breaking),
              "channels_checked": channels_checked,
              "compatible_additions": additions,
              "baseline_version": baseline_version,
              "candidate_version": candidate_version,
              "error": error,
          }
          print("channels=%d breaking=%d additions=%d verdict=%s" % (
              channels_checked, len(breaking), len(additions), verdict))
          print("::" + json.dumps({"outputs": payload}) + "::")

  # --- 2. verdict gate ---
  - id: gate_contract_changes
    type: io.kestra.plugin.core.flow.If
    description: Escalate breaking contract changes before consumers see them, and
      separate an unreadable candidate from a clean compatibility verdict.
    condition: "{{ outputs.compare_specs.vars.verdict == 'breaking' }}"
    then:
      - id: alert_breaking_change
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the platform channel which channels or required fields the
          candidate dropped, with the CI build to inspect.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "AsyncAPI breaking change for {{ inputs.service_name }} - {{ outputs.compare_specs.vars.breaking_count }} breaking change(s) {{ outputs.compare_specs.vars.breaking_changes | join('; ') }}. Candidate v{{ outputs.compare_specs.vars.candidate_version }} against baseline v{{ outputs.compare_specs.vars.baseline_version }}. Build {{ inputs.ci_build_url }}. Execution {{ execution.id }}."
          }
    else:
      - id: check_parse_error
        type: io.kestra.plugin.core.flow.If
        description: Route an unparseable candidate into its own alert so a broken file
          never reads as a compatible contract.
        condition: "{{ outputs.compare_specs.vars.verdict == 'error' }}"
        then:
          - id: alert_unreadable_spec
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Tell the platform channel the candidate spec could not be parsed
              and name the load problem.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "AsyncAPI candidate unreadable for {{ inputs.service_name }} - {{ outputs.compare_specs.vars.error }}. Build {{ inputs.ci_build_url }}. Execution {{ execution.id }}."
              }
        else:
          - id: log_compatible_contract
            type: io.kestra.plugin.core.log.Log
            description: Record the compatible verdict with the channel count so the
              execution history is the contract evidence.
            message: "asyncapi contract compatible for {{ inputs.service_name }} - {{
              outputs.compare_specs.vars.channels_checked }} baseline channel(s)
              unchanged, candidate v{{
              outputs.compare_specs.vars.candidate_version }} carries {{
              outputs.compare_specs.vars.compatible_additions | length }} added
              channel(s)"

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: compatible when nothing was removed, breaking when a channel,
      operation, required field or property disappeared, error when a spec could
      not be parsed.
    value: "{{ outputs.compare_specs.vars.verdict }}"
  - id: breaking_changes
    type: ARRAY
    description: 'Breaking removals in baseline order, for example ["removed
      required field order/created.publish.currency"].'
    value: "{{ outputs.compare_specs.vars.breaking_changes | toJson }}"
  - id: breaking_count
    type: INT
    description: Number of breaking removals, for example 2.
    value: "{{ outputs.compare_specs.vars.breaking_count }}"
  - id: channels_checked
    type: INT
    description: Number of baseline channels the candidate was compared against, for
      example 3.
    value: "{{ outputs.compare_specs.vars.channels_checked }}"
  - id: compatible_additions
    type: ARRAY
    description: 'Channels the candidate added on top of the baseline, for example
      ["added channel order/refunded"].'
    value: "{{ outputs.compare_specs.vars.compatible_additions | toJson }}"
  - id: candidate_version
    type: STRING
    description: Version string from the candidate info block, for example 1.5.0.
    value: "{{ outputs.compare_specs.vars.candidate_version }}"

REST contracts grew pull request gates years ago, but an event stream still ships a dropped required field straight into production because nothing compares the new AsyncAPI document against the one consumers built against. This blueprint takes the candidate document from CI, materializes it beside the blessed baseline in a shared workspace, diffs every publish and subscribe payload channel by channel, and escalates removals through Slack with the verdict published as outputs.

How it works

post_candidate_spec (io.kestra.plugin.core.trigger.Webhook, keyed with the ASYNCAPI_WEBHOOK_KEY secret) accepts the candidate document as a raw YAML body from CI; nightly_contract_check (io.kestra.plugin.core.trigger.Schedule, cron 0 7 * * *) ships disabled: true until an operator switches it on.
spec_workspace (io.kestra.plugin.core.flow.WorkingDirectory) opens one shared directory. write_specs (io.kestra.plugin.scripts.shell.Commands on the Docker runner, image python:3.12-slim) writes the blessed baseline with one quoted heredoc and the posted body, or the candidate_spec input default on manual runs, with a second, and compare_specs (io.kestra.plugin.scripts.python.Script) reads exactly those two files. Both children are scripts inside the workspace and every notification sits after the block, because default tasks are stateless and a temp dir is purged when a task completes.
compare_specs installs PyYAML at the start of the container run, walks every channel and its publish and subscribe operations, collects each payload's required fields and properties, and diffs them against the baseline. A removed channel, a removed operation, a required field that left the required list or a property that left the payload is one breaking change, a channel the candidate adds is a compatible addition, and the verdict is printed as compatible, breaking, or error on one ::{"outputs": ...}:: line.
gate_contract_changes (io.kestra.plugin.core.flow.If) escalates breaking into alert_breaking_change (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) with the removals and the CI build, an error run routes through check_parse_error and alert_unreadable_spec, and a clean contract lands in log_compatible_contract. The flow level errors handler reports any workspace or script failure with the execution id.

What you get

A contract gate for event streams that works the way OpenAPI gates work for REST, run before the branch merges instead of after consumers break.
The breaking removals as typed outputs, each one a removed channel, operation, required field or property in baseline order.
Escalation where the team already works, a Slack alert with the removals, both spec versions and the CI build to inspect.
A fail safe posture, an unreadable candidate escalates instead of reading as a compatible contract.
An execution history per candidate document that doubles as the contract review evidence.

Who it's for

Platform teams that own shared event schemas across many producing and consuming services.
Release engineers who need a contract verdict in the pipeline before a branch with schema changes merges.
API and event contract reviewers who want removals named in a channel instead of buried in a diff review.

Why orchestrate this with Kestra

AsyncAPI documents change in the same repositories as the services that emit them, and nothing stops a dropped field from reaching every consumer at once. Kestra gives the check a keyed webhook for CI, a shared workspace where the baseline and the candidate sit next to each other, the comparison logic versioned in the flow file, a verdict branch that separates breaking, unreadable and compatible, and an execution history per candidate that any parent pipeline can branch on. A lint step in CI gives you none of that, it only protects the repositories that remember to run it.

Prerequisites

A worker with Docker access, the comparison runs in python:3.12-slim and installs PyYAML at the start of each run.
A CI step that POSTs the candidate document as a raw YAML or JSON body, for example curl --data-binary @asyncapi.yaml against the webhook URL.
A Slack incoming webhook for the breaking change alert, the unreadable candidate alert and the failure handler.

Secrets

ASYNCAPI_WEBHOOK_KEY: key guarding the candidate webhook URL, anyone holding it can post a contract.
SLACK_WEBHOOK_URL: Slack incoming webhook URL used by the breaking change alert and the errors handler.

Quick start

Add the two secrets to your Kestra namespace.
Run the flow once with the default inputs and confirm verdict is breaking with breaking_count at 2, the fixture drops currency from order/created and removes the whole order/cancelled channel while adding order/refunded.
POST the baseline document itself to the webhook and watch verdict flip to compatible with breaking_changes empty in the outputs.
Point the event service pipeline at /api/v1/{tenant}/executions/webhook/{namespace}/asyncapi-contract-breaking-change-gate/{ASYNCAPI_WEBHOOK_KEY} and enable nightly_contract_check once every producer posts through it.

How to extend

Add server variable and message example checks to compare_specs so a renamed server or a dropped example also blocks the merge.
Gate many contracts at once with io.kestra.plugin.core.flow.ForEachItem over a spec list, each subflow posting its own candidate document.
Keep the previous breaking changes in the Kestra KV store and alert only when a new removal appears, so a known exception does not page every build.
Stop the promoting pipeline by adding io.kestra.plugin.core.execution.Fail under gate_contract_changes, and archive every candidate with io.kestra.plugin.aws.s3.Upload for the contract evidence pack.

Gate Breaking AsyncAPI Changes Before Consumers See Them

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?

Gate Breaking AsyncAPI Changes Before Consumers See Them

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

Arcmira: YouTube Transcript Search to CSV

Amazon DynamoDB Throttled Requests and Capacity Starvation Sentinel

Automated Customer Feedback Sentiment Analysis and Reporting

Glue ETL, Athena Validation, DynamoDB Cache

Autoscale EMR Serverless Spark on a CloudWatch Backlog

Catch AWS Spend Spikes Before Standup

New to Kestra?