Kafka Schema Registry Compatibility and Mode Drift Guard with Kestra

id: kafka-schema-registry-compatibility-mode-drift-guard
namespace: company.team

description: |
  Audit every Confluent Schema Registry subject backing a live Kafka topic against a governed
  compatibility level and a governed subject mode, then pin the drifted ones back behind an approval
  gate and re-read both settings to prove it landed. A subject left at `NONE` accepts a breaking
  schema in silence, so the guard that stops a producer from poisoning its consumers is simply off.
  A subject left in `READONLY` or `IMPORT` after a restore refuses every write with a 422 that reads
  like a client bug. The registry answers both requests normally either way, so nothing reports it.
  Compatibility and mode are checked independently, because they drift independently. A subject with
  no override inherits the global setting, which is reported as inherited and judged by the effective
  value.

inputs:
  - 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: bootstrap_servers
    type: STRING
    displayName: Kafka bootstrap servers
    description: Comma separated broker list, used to read the live topic set that
      scopes the audit.
    defaults: localhost:9092

  - id: baseline_compatibility
    type: SELECT
    displayName: Governed compatibility level
    description: The level every audited subject must resolve to.
    values:
      - BACKWARD
      - BACKWARD_TRANSITIVE
      - FORWARD
      - FORWARD_TRANSITIVE
      - FULL
      - FULL_TRANSITIVE
      - NONE
    defaults: BACKWARD

  - id: baseline_mode
    type: SELECT
    displayName: Governed subject mode
    description: The mode every audited subject must resolve to. READWRITE is the
      only mode that accepts producer writes.
    values:
      - READWRITE
      - READONLY
      - IMPORT
    defaults: READWRITE

  - id: inherited_is_compliant
    type: BOOL
    displayName: Treat an inherited setting as compliant
    description: True means a subject with no override is judged by the global
      setting. False means every audited subject must carry its own explicit
      override.
    defaults: true

  - id: subjects
    type: ARRAY
    itemType: STRING
    displayName: Subjects to audit
    description: Leave empty to audit every registered subject that backs a live
      topic under the TopicNameStrategy. Name subjects here to audit those
      instead.
    defaults: []

  - id: exempt_subjects
    type: ARRAY
    itemType: STRING
    displayName: Subjects exempt from the baseline
    description: Subjects allowed to sit off baseline, for example one deliberately
      held at NONE for a migration. Never read, never changed.
    defaults: []

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, for any setting to be written.
    defaults: true

variables:
  audit_key: "sr_drift_audit_{{ inputs.schema_registry_url | slugify }}"
  verify_key: "sr_drift_verify_{{ inputs.schema_registry_url | slugify }}"

tasks:
  - id: list_subjects
    type: io.kestra.plugin.core.http.Request
    description: Read-only. Every subject the registry serves right now.
    method: GET
    uri: "{{ inputs.schema_registry_url }}/subjects"

  - id: list_topics
    type: io.kestra.plugin.kafka.TopicList
    description: Read-only AdminClient call. The live topic set is what scopes the
      audit, so a subject whose topic is gone is never read or written.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: read_global_config
    type: io.kestra.plugin.core.http.Request
    description: Read-only. The global compatibility level, which every subject
      without an override inherits.
    method: GET
    uri: "{{ inputs.schema_registry_url }}/config"

  - id: read_global_mode
    type: io.kestra.plugin.core.http.Request
    description: Read-only. The global mode, which every subject without an override
      inherits.
    method: GET
    uri: "{{ inputs.schema_registry_url }}/mode"

  - id: globals
    type: io.kestra.plugin.core.output.OutputValues
    description: The two inherited values, pulled out once so every subject row is
      judged against the same pair.
    values:
      config_level: "{{ fromJson(outputs.read_global_config.body).compatibilityLevel
        ?? 'UNKNOWN' }}"
      mode: "{{ fromJson(outputs.read_global_mode.body).mode ?? 'UNKNOWN' }}"

  - id: scope
    type: io.kestra.plugin.core.output.OutputValues
    description: Audited subjects are the registered ones that back a live topic (or
      the named list) minus the exempt ones. Internal topics are dropped.
    values:
      subject_list: >-
        {{ {'reg': fromJson(outputs.list_subjects.body), 'topics':
        outputs.list_topics.topics, 'sel': inputs.subjects, 'ex':
        inputs.exempt_subjects} | jq('.reg as $reg | .sel as $sel | .ex as $ex |
        ([ .topics[] | select(startswith("_") | not) ]) as $pt | (($pt | map(. +
        "-value")) + ($pt | map(. + "-key"))) as $exp | (if ($sel | length) > 0
        then ($reg - ($reg - $sel)) else ($reg - ($reg - $exp)) end) as $base |
        (($base - $ex) | sort)') | first | toJson }}

  - id: in_scope
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.scope.values.subject_list) | length > 0 }}"
    then:
      - id: reset_audit
        type: io.kestra.plugin.core.kv.Set
        description: The audit accumulator is rebuilt from scratch every run. No ttl, so
          the read back inside the run can never hit an expired key.
        key: "{{ render(vars.audit_key) }}"
        kvType: JSON
        value: "{}"

      - id: audit
        type: io.kestra.plugin.core.flow.Loop
        description: Read both settings per subject. One at a time, because each
          iteration folds its row into a single accumulator.
        concurrencyLimit: 1
        values: "{{ outputs.scope.values.subject_list }}"
        tasks:
          - id: read_config
            type: io.kestra.plugin.core.http.Request
            description: Read-only. A 404 here is not an error, it means the subject has no
              override and inherits the global level.
            method: GET
            uri: "{{ inputs.schema_registry_url }}/config/{{ item.value }}"
            options:
              allowFailed: true

          - id: read_mode
            type: io.kestra.plugin.core.http.Request
            description: Read-only. A 404 here means the subject has no mode override and
              inherits the global mode.
            method: GET
            uri: "{{ inputs.schema_registry_url }}/mode/{{ item.value }}"
            options:
              allowFailed: true

          - id: row_merge
            type: io.kestra.plugin.core.output.OutputValues
            description: Observed facts only. The baseline comparison happens once, in
              assess.
            values:
              merged: >-
                {{ {'cur': (kv(render(vars.audit_key), errorOnMissing=false) ??
                '{}'), 'subject': item.value, 'ccode': outputs.read_config.code,
                'cbody': outputs.read_config.body, 'mcode':
                outputs.read_mode.code, 'mbody': outputs.read_mode.body, 'gc':
                outputs.globals.values.config_level, 'gm':
                outputs.globals.values.mode} | jq('(.cur | if type=="string"
                then (if .=="" then {} else fromjson end) else . end) as $c |
                {subject: .subject, config_override: (.ccode == 200),
                config_effective: (if .ccode == 200 then (.cbody | fromjson |
                .compatibilityLevel) else .gc end), mode_override: (.mcode ==
                200), mode_effective: (if .mcode == 200 then (.mbody | fromjson
                | .mode) else .gm end)} as $row | $c + {(.subject): $row}') |
                first | toJson }}

          - id: row_save
            type: io.kestra.plugin.core.kv.Set
            key: "{{ render(vars.audit_key) }}"
            kvType: JSON
            value: "{{ outputs.row_merge.values.merged }}"

      - id: assess
        type: io.kestra.plugin.core.output.OutputValues
        description: A subject drifts when its effective value differs from the
          baseline. It also drifts when an override is required and it has none.
          The two settings are judged separately.
        values:
          result: >-
            {{ {'rows': (kv(render(vars.audit_key), errorOnMissing=false) ??
            '{}'), 'bc': inputs.baseline_compatibility, 'bm':
            inputs.baseline_mode, 'ok': inputs.inherited_is_compliant} |
            jq('(.rows | if type=="string" then (if .=="" then {} else fromjson
            end) else . end) as $r | .bc as $bc | .bm as $bm | .ok as $ok | [
            $r[] | {subject: .subject, config_override: .config_override,
            config_effective: .config_effective, mode_override: .mode_override,
            mode_effective: .mode_effective, fix_config: ((.config_effective !=
            $bc) or (($ok | not) and (.config_override | not))), fix_mode:
            ((.mode_effective != $bm) or (($ok | not) and (.mode_override |
            not)))} ] as $all | ([ $all[] | select(.fix_config or .fix_mode) ] |
            sort_by(.subject)) as $d | {audited: ($r | length), count: ($d |
            length), config_count: ([ $d[] | select(.fix_config) ] | length),
            mode_count: ([ $d[] | select(.fix_mode) ] | length), subjects: ($d |
            map(.subject)), drift: $d, summary: ([ $d[] | .subject + "
            [compatibility " + .config_effective + (if .config_override then "
            override" else " inherited" end) + (if .fix_config then " DRIFT"
            else " ok" end) + ", mode " + .mode_effective + (if .mode_override
            then " override" else " inherited" end) + (if .fix_mode then "
            DRIFT" else " ok" end) + "]" ] | join("; "))}') | first | toJson }}

      - id: route
        type: io.kestra.plugin.core.flow.If
        condition: "{{ fromJson(outputs.assess.values.result).count > 0 }}"
        then:
          - id: log_findings
            type: io.kestra.plugin.core.log.Log
            message: >-
              Governance scan on {{ inputs.schema_registry_url }}: {{
              fromJson(outputs.assess.values.result).audited }} subjects
              audited, {{ fromJson(outputs.assess.values.result).count }} off
              baseline ({{ fromJson(outputs.assess.values.result).config_count
              }} on compatibility, {{
              fromJson(outputs.assess.values.result).mode_count }} on mode).
              Baseline is compatibility {{ inputs.baseline_compatibility }} and
              mode {{ inputs.baseline_mode }}. Global settings are compatibility
              {{ outputs.globals.values.config_level }} and mode {{
              outputs.globals.values.mode }}. Drift: {{
              fromJson(outputs.assess.values.result).summary }}.
              auto_remediate={{ inputs.auto_remediate }} dry_run={{
              inputs.dry_run }}.

          - id: alert
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: >-
              {{ {'text': 'Schema Registry ' ~ inputs.schema_registry_url ~ ': '
              ~ (fromJson(outputs.assess.values.result).count) ~ ' of ' ~
              (fromJson(outputs.assess.values.result).audited) ~ ' subjects are
              off the governed baseline (compatibility ' ~
              inputs.baseline_compatibility ~ ', mode ' ~ inputs.baseline_mode ~
              '). ' ~ (fromJson(outputs.assess.values.result).summary) ~ '.
              auto_remediate=' ~ inputs.auto_remediate ~ ' dry_run=' ~
              inputs.dry_run ~ '.'} | toJson }}

          - id: remediate_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
            then:
              - id: approve
                type: io.kestra.plugin.core.flow.Pause
                description: Resume with approved=true to pin the subjects listed by
                  log_findings back to the baseline. Resume without it or kill
                  the execution to cancel.
                onResume:
                  - id: approved
                    type: BOOL
                    displayName: Approve pinning the drifted subjects back to the baseline
                    required: false

              - id: do_fix
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.approve.onResume.approved ?? false }}"
                then:
                  - id: remediate_pin
                    type: io.kestra.plugin.core.flow.Loop
                    description: Remediation. Write back only the setting that drifted, one subject
                      at a time so the registry never takes concurrent config
                      writes.
                    concurrencyLimit: 1
                    values: "{{ fromJson(outputs.assess.values.result).drift | toJson }}"
                    tasks:
                      - id: pin_config_gate
                        type: io.kestra.plugin.core.flow.If
                        condition: "{{ fromJson(item.value).fix_config }}"
                        then:
                          - id: put_config
                            type: io.kestra.plugin.core.http.Request
                            description: Set the subject level compatibility. allowFailed lets a rejected
                              write fall through to the re-read, which is the
                              authority.
                            method: PUT
                            uri: "{{ inputs.schema_registry_url }}/config/{{ fromJson(item.value).subject
                              }}"
                            contentType: application/vnd.schemaregistry.v1+json
                            body: "{{ {'compatibility': inputs.baseline_compatibility} | toJson }}"
                            options:
                              allowFailed: true

                      - id: pin_mode_gate
                        type: io.kestra.plugin.core.flow.If
                        condition: "{{ fromJson(item.value).fix_mode }}"
                        then:
                          - id: put_mode
                            type: io.kestra.plugin.core.http.Request
                            description: Set the subject level mode. allowFailed lets a rejected write fall
                              through to the re-read.
                            method: PUT
                            uri: "{{ inputs.schema_registry_url }}/mode/{{ fromJson(item.value).subject }}"
                            contentType: application/vnd.schemaregistry.v1+json
                            body: "{{ {'mode': inputs.baseline_mode} | toJson }}"
                            options:
                              allowFailed: true

                  - id: reset_verify
                    type: io.kestra.plugin.core.kv.Set
                    key: "{{ render(vars.verify_key) }}"
                    kvType: JSON
                    value: "{}"

                  - id: verify
                    type: io.kestra.plugin.core.flow.Loop
                    description: Read both settings again, straight from the registry, for every
                      subject that was targeted.
                    concurrencyLimit: 1
                    values: "{{ fromJson(outputs.assess.values.result).subjects | toJson }}"
                    tasks:
                      - id: recheck_config
                        type: io.kestra.plugin.core.http.Request
                        method: GET
                        uri: "{{ inputs.schema_registry_url }}/config/{{ item.value }}"
                        options:
                          allowFailed: true

                      - id: recheck_mode
                        type: io.kestra.plugin.core.http.Request
                        method: GET
                        uri: "{{ inputs.schema_registry_url }}/mode/{{ item.value }}"
                        options:
                          allowFailed: true

                      - id: recheck_merge
                        type: io.kestra.plugin.core.output.OutputValues
                        values:
                          merged: >-
                            {{ {'cur': (kv(render(vars.verify_key),
                            errorOnMissing=false) ?? '{}'), 'subject':
                            item.value, 'ccode': outputs.recheck_config.code,
                            'cbody': outputs.recheck_config.body, 'mcode':
                            outputs.recheck_mode.code, 'mbody':
                            outputs.recheck_mode.body, 'gc':
                            outputs.globals.values.config_level, 'gm':
                            outputs.globals.values.mode} | jq('(.cur | if
                            type=="string" then (if .=="" then {} else fromjson
                            end) else . end) as $c | {subject: .subject,
                            config_override: (.ccode == 200), config_effective:
                            (if .ccode == 200 then (.cbody | fromjson |
                            .compatibilityLevel) else .gc end), mode_override:
                            (.mcode == 200), mode_effective: (if .mcode == 200
                            then (.mbody | fromjson | .mode) else .gm end)} as
                            $row | $c + {(.subject): $row}') | first | toJson }}

                      - id: recheck_save
                        type: io.kestra.plugin.core.kv.Set
                        key: "{{ render(vars.verify_key) }}"
                        kvType: JSON
                        value: "{{ outputs.recheck_merge.values.merged }}"

                  - id: verify_assess
                    type: io.kestra.plugin.core.output.OutputValues
                    description: The same rule as assess, applied to the post write reading.
                    values:
                      result: >-
                        {{ {'rows': (kv(render(vars.verify_key),
                        errorOnMissing=false) ?? '{}'), 'bc':
                        inputs.baseline_compatibility, 'bm':
                        inputs.baseline_mode, 'ok':
                        inputs.inherited_is_compliant} | jq('(.rows | if
                        type=="string" then (if .=="" then {} else fromjson end)
                        else . end) as $r | .bc as $bc | .bm as $bm | .ok as $ok
                        | ([ $r[] | select(((.config_effective != $bc) or (($ok
                        | not) and (.config_override | not))) or
                        ((.mode_effective != $bm) or (($ok | not) and
                        (.mode_override | not)))) ] | sort_by(.subject)) as $bad
                        | {checked: ($r | length), bad: ($bad | length),
                        summary: ([ $bad[] | .subject + " [compatibility " +
                        .config_effective + ", mode " + .mode_effective + "]" ]
                        | join("; "))}') | first | toJson }}

                  - id: verify_result
                    type: io.kestra.plugin.core.flow.If
                    condition: "{{ fromJson(outputs.verify_assess.values.result).bad > 0 }}"
                    then:
                      - id: pin_failed
                        type: io.kestra.plugin.core.execution.Fail
                        errorMessage: >-
                          The write ran, but a re-read still shows {{
                          fromJson(outputs.verify_assess.values.result).bad }}
                          of {{
                          fromJson(outputs.verify_assess.values.result).checked
                          }} targeted subjects off baseline: {{
                          fromJson(outputs.verify_assess.values.result).summary
                          }}. The registry may have rejected the write. A client
                          may have changed the setting back during the pause.
                          The baseline mode may need a forced transition.
                    else:
                      - id: pin_ok
                        type: io.kestra.plugin.core.log.Log
                        message: >-
                          Pinned {{ fromJson(outputs.assess.values.result).count
                          }} subjects back to compatibility {{
                          inputs.baseline_compatibility }} and mode {{
                          inputs.baseline_mode }} ({{
                          fromJson(outputs.assess.values.result).subjects |
                          jq('join(", ")') | first }}). A re-read of all {{
                          fromJson(outputs.verify_assess.values.result).checked
                          }} confirms every one is on baseline.
                else:
                  - id: not_approved
                    type: io.kestra.plugin.core.log.Log
                    message: "Pause resumed without approval or killed. No compatibility level or
                      mode was changed."
        else:
          - id: all_clean
            type: io.kestra.plugin.core.log.Log
            message: >-
              All {{ fromJson(outputs.assess.values.result).audited }} audited
              subjects on {{ inputs.schema_registry_url }} resolve to
              compatibility {{ inputs.baseline_compatibility }} and mode {{
              inputs.baseline_mode }}. The safety net is on everywhere. Nothing
              to fix.
    else:
      - id: nothing_in_scope
        type: io.kestra.plugin.core.log.Log
        message: >-
          No audited subjects on {{ inputs.schema_registry_url }}. Either no
          registered subject backs a live topic under the TopicNameStrategy or
          every match was exempt. Nothing read, nothing changed.

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: >-
      {{ {'text': 'Kafka schema registry compatibility and mode drift guard
      failed on ' ~ inputs.schema_registry_url ~ ' (execution ' ~ execution.id ~
      '). A subject may still be off the governed baseline. Check the logs.'} |
      toJson }}

triggers:
  - id: daily_scan
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "0 6 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

A Confluent Schema Registry decides whether a new schema version may be registered by checking the subject's compatibility level, then whether a producer may write at all by checking the subject's mode. Both are per subject settings with a global default. Both are changed by one HTTP call. Neither is reported anywhere once changed.

Set a subject to NONE for a one off migration then forget to put it back. Every breaking change after that is accepted in silence. The registry answers 200 with a new schema id, the producer ships, the consumers break on the next deploy. Leave a subject in READONLY or IMPORT after a restore. Every write is then refused with HTTP 422, which reads like a client bug rather than a registry setting. The registry serves both requests normally in both cases, so nothing raises it.

This blueprint reads both settings for every subject that backs a live Kafka topic, compares each against a baseline the operator declares as an input, pins the drifted ones back behind an approval gate, then re-reads both settings straight from the registry and fails the run if any targeted subject is still off baseline.

This blueprint was created by zkasuran.

Findings

Measured against Confluent Schema Registry 7.9.0 while building this.

With a subject at NONE the registry reports is_compatible: true for a schema that adds a mandatory field with no default, then registers it as a new version. With the same subject at BACKWARD the identical change is refused with HTTP 409 and READER_FIELD_MISSING_DEFAULT_VALUE. The level is the whole safety net.
GET /config/<subject> answers HTTP 404 with error_code 40408 when the subject carries no override, GET /mode/<subject> answers 404 with 40409. A 404 there means the subject inherits the global value. It is not a failure, which is why both reads allow a failed status and the code is part of the reading.
PUT /config/<subject> takes the key compatibility while GET returns compatibilityLevel. Reading back what you just wrote needs the other name.
Moving a subject out of READONLY or IMPORT back to READWRITE needs no force. Moving a subject that already holds schemas into IMPORT is refused with HTTP 422 and Cannot import since found existing subjects, so IMPORT is not a baseline a live registry can hold. A run that asks for it fails on the re-read rather than reporting a fix.
Changing the global level silently drifts every subject that inherits it. The remediation writes a subject level override instead of touching the global value, so the blast radius stays on the subjects that were audited.

Safety

Read-only by default. With the shipped inputs it lists subjects, reads both settings per subject, reports. It writes nothing.
A write needs three things at once: auto_remediate true, dry_run false, a human resuming the Pause with approved=true. An unanswered resume reads as not approved.
Only the setting that drifted is written. A subject off baseline on mode alone never gains a compatibility override it did not have before.
Subjects named in exempt_subjects are dropped before the audit, so a subject deliberately held at NONE for a migration is never read or written.
The global settings are never written. A drifted global shows up as drift on every inheriting subject.
Only subjects backing a live topic are audited. A subject whose topic is gone is left alone.
The proof is a fresh GET of both endpoints per targeted subject after the writes. A rejected write fails the run instead of being reported as fixed.

How it works

list_subjects (io.kestra.plugin.core.http.Request) reads GET /subjects, the subjects the registry serves right now.
list_topics (io.kestra.plugin.kafka.TopicList, read-only) returns the live topics on the cluster. That set is what scopes the audit.
read_global_config and read_global_mode read GET /config plus GET /mode. globals pulls out the two inherited values so every subject is judged against the same pair.
scope keeps the registered subjects that follow the <topic>-value or <topic>-key naming for a live non internal topic. The subjects input replaces that derivation when it is set. exempt_subjects is dropped either way.
in_scope (io.kestra.plugin.core.flow.If) skips straight to a log line when nothing is in scope.
audit (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) reads GET /config/<subject> plus GET /mode/<subject> for each subject, with a failed status allowed so a 404 is read as inherited. Each iteration folds its row into one KV accumulator, which is reset at the start of every run.
assess applies the baseline once to the whole reading. A subject drifts when its effective value differs from the baseline. It also drifts when inherited_is_compliant is false and it carries no override. Compatibility and mode are judged separately, so one subject can drift on either or both.
route branches on whether anything drifted.
- Nothing drifted: all_clean logs that the net is on everywhere. No alert.
- Something drifted: log_findings logs the per subject reading, including the global values so the cause is visible, then alert posts it to Slack.
  - remediate_gate runs only when auto_remediate is true and dry_run is false.
    - approve (io.kestra.plugin.core.flow.Pause) waits for a resume carrying approved.
    - When approved, remediate_pin (Loop, concurrencyLimit: 1) issues PUT /config/<subject> for a compatibility drift plus PUT /mode/<subject> for a mode drift, one subject at a time.
    - verify then re-reads both endpoints for every targeted subject into a second accumulator, verify_assess applies the same rule to that reading, verify_result fails the run through io.kestra.plugin.core.execution.Fail if any targeted subject is still off baseline.
    - When not approved, not_approved logs that nothing changed.
errors posts to Slack on any failure, including an unreachable registry plus a write the registry refused.

What you get

A per subject reading of compatibility and mode, with inherited marked as inherited, against a baseline you declare.
Detection of the two silent failure modes: a safety net switched off, a subject that refuses writes.
A fix that is off by default, writes only the setting that drifted, touches only the subjects it reported.
Proof the fix landed: a fresh read of both endpoints that fails the run when a write was refused or reverted.

Who it's for

Platform teams who own a shared registry where one curl can turn off compatibility checking for a whole subject.
Anyone who has restored a registry from a backup then found producers failing with HTTP 422.
Teams that run compatibility as a governance rule rather than a per subject default, who need the rule checked on a schedule.

Why orchestrate this with Kestra

Reading two settings per subject is two HTTP calls. Doing it for every subject that backs a live topic, every day, against a declared baseline, is a pipeline. Kestra adds the schedule, the live topic set from the broker that scopes the audit, the Pause that turns a governance write into an approved change, the re-read that proves the write landed, the Slack alert, a history of every run you can point an auditor at.

Tested end to end

On Kestra 2.0.5 OSS against Apache Kafka 3.9.1 (KRaft, single node) with Confluent Schema Registry 7.9.0 and a local HTTP sink standing in for Slack. Topics orders, payments, shipments, with four subjects: orders-value (compatibility override), orders-key (no override at all), payments-value (mode override), shipments-value (an override equal to the baseline). Every state was confirmed with GET /config/<subject> plus GET /mode/<subject> straight against the registry before and after each run.

Run	Inputs	Result
1	defaults, `orders-value` at `NONE`, `payments-value` in `READONLY`	4 audited, 2 off baseline (1 compatibility, 1 mode). Nothing written
2	`auto_remediate: true`, `dry_run: true`	Same report, no `Pause`, registry unchanged
3	`auto_remediate: true`, `dry_run: false`, resumed with `approved=false`	Paused, declined, nothing written
4	`auto_remediate: true`, `dry_run: false`, approved	`orders-value` `NONE` to `BACKWARD`, `payments-value` `READONLY` to `READWRITE`, re-read confirmed both. `orders-key` plus `shipments-value` untouched
5	defaults on the repaired registry	4 audited, 0 off baseline, `all_clean`, no alert
6	`inherited_is_compliant: false`	4 off baseline (2 compatibility, 3 mode), since every subject misses at least one override
7	`orders-value` back at `NONE`, `exempt_subjects: ["orders-value"]`	3 audited, 0 off baseline. The exempt subject was never read
8	`baseline_mode: IMPORT`, approved	Every `PUT /mode` refused with HTTP 422. The re-read showed 4 of 4 still off baseline, the run FAILED, Slack alerted. The compatibility write in the same run did land
9	registry URL pointed at a closed port	`list_subjects` failed, run FAILED, `errors` alerted
10	`subjects: ["no-such-subject"]`	Nothing in scope, nothing read, nothing written
11	global level set to `NONE`, defaults	2 off baseline, both reported as inherited `NONE`. The two subjects with explicit `BACKWARD` overrides stayed clean
12	same, approved	Both pinned with a subject level `BACKWARD` override. The global level was left at `NONE`

The detector was proven able to fail rather than only to pass: identical inputs gave 2 off baseline on the degraded registry (run 1) then 0 on the repaired one (run 5). The same registry gave 0 with the drifted subject exempt (run 7). The safety net itself was measured, not assumed: at NONE the registry accepted a schema adding a mandatory field with no default and bumped the subject to version 2, then after run 4 pinned it back to BACKWARD the same class of change was refused with HTTP 409.

Prerequisites

A Kafka cluster plus a Confluent Schema Registry reachable from the Kestra worker. The registry stores its state in a Kafka topic, so it cannot run without a broker. That is why this blueprint is not a demo.
A registry principal allowed to read /config plus /mode. Remediation also needs write on both. A broker principal allowed to list topics.
A registry that allows mode changes. PUT /mode/<subject> depends on the registry's mode.mutability setting, which the test registry for this blueprint was started with enabled.
Subjects named with the default TopicNameStrategy for the topic scoped audit. Any other naming can be audited by listing the subjects in the subjects input.

Secrets

SLACK_WEBHOOK_URL: used by alert and the errors block. In Kestra OSS, provide it as the base64 encoded environment variable SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`schema_registry_url`	`http://localhost:8081`	Registry base URL, no trailing slash
`bootstrap_servers`	`localhost:9092`	Brokers, read the live topic set
`baseline_compatibility`	`BACKWARD`	The level every audited subject must resolve to
`baseline_mode`	`READWRITE`	The mode every audited subject must resolve to
`inherited_is_compliant`	`true`	False requires every audited subject to carry its own override
`subjects`	`[]`	Audit these subjects instead of deriving them from live topics
`exempt_subjects`	`[]`	Subjects allowed to sit off baseline
`auto_remediate`	`false`	Allow the write
`dry_run`	`true`	Must be `false` for a write to apply

Quick start

Add the SLACK_WEBHOOK_URL secret.
Import the flow, set schema_registry_url plus bootstrap_servers.
Run it with the defaults. Read which subjects are off your baseline and which inherit it.
Set baseline_compatibility plus baseline_mode to the rule your team actually governs by. List any deliberate exception in exempt_subjects.
Run it with auto_remediate: true plus dry_run: false, then resume the paused execution with approved=true.
Enable the daily_scan trigger, which always runs in safe mode.

How to extend

Set inherited_is_compliant to false to require an explicit override on every governed subject, so a later change to the global level cannot quietly weaken them.
Point the flow at a second registry by changing schema_registry_url, so one Kestra instance governs several clusters on one schedule.
Record each reading to a table before the gate, giving a drift history you can chart.
Add a Jira or ServiceNow task on the drift branch, so a NONE left behind by a migration opens a ticket as well as a Slack message.
Audit a non default subject naming strategy by filling the subjects input from your own inventory instead of letting live topics scope it.
Gate a CI deploy on this flow: fail the pipeline when any subject is off baseline, rather than fixing it.

Kafka Schema Registry Compatibility and Mode Drift Guard

Findings

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Kafka Schema Registry Compatibility and Mode Drift Guard

Findings

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Kafka Orphaned Topic Reclaim Guard

Kafka Consumer Lag Sentinel

Kafka Schema Registry Subject Version Pruning Guard

Kafka DLQ Quarantine, Classification, and Safe Replay Gate

Kafka Idle Consumer Partition Guard

Real-Time Kafka Record Processing with Per-Record Executions

More Infrastructure Blueprints

Nomad Allocation Reschedule Storm Guard

Weekly npm Outdated Major-Bump Digest

Odoo Inventory Stock Threshold Gate

Odoo Scheduled Action Overdue Sentinel

Nutanix safe patching with snapshot rollback

NPM Package Dependency Typosquatting Sentinel

New to Kestra?