New to Kestra?
Use blueprints to kickstart your first workflows.
Gate MySQL Debezium CDC Loads on Schema Changes with Approval
Hold Debezium MySQL CDC batches when DDL changes a watched table. Approve, apply the DDL to the target, verify the schema, then upsert the rows.
Categories
Data
id: debezium-mysql-ddl-change-gate
namespace: company.data
description: |
Replicate watched MySQL tables to a downstream MySQL database with Debezium. Each run reads the
new binlog events and the new DDL that Debezium records in its schema history. Row changes load
with an idempotent upsert and delete. When an ALTER, DROP or RENAME touches a watched table, the
flow stops before the load and asks for a decision. Nothing loads with a changed schema until a
person approves, or until the change is additive and auto_apply_additive is true. A captured
batch stays pending in KV until it loads, so a held or failed batch is retried, never skipped.
concurrency:
limit: 1
behavior: CANCEL
inputs:
- id: source_host
type: STRING
displayName: Source MySQL host
defaults: mysql-primary
- id: source_port
type: STRING
displayName: Source MySQL port
defaults: "3306"
- id: source_server_id
type: STRING
displayName: Debezium server id
defaults: "184054"
description: Must not be the server_id of any replica or other binlog client of
the source.
- id: target_jdbc_url
type: STRING
displayName: Target JDBC URL
defaults: "jdbc:mysql://mysql-analytics:3306/shop_replica?connectionTimeZone=UT\
C&forceConnectionTimeZoneToSession=true"
description: The database in the URL holds one table with the same name and
primary key for each watched table. Keep the session time zone at UTC.
- id: auto_apply_additive
type: BOOL
displayName: Apply additive DDL without approval
defaults: false
description: When true, an ALTER that only adds nullable or defaulted columns is
applied to the target without a pause. Every other change waits for a
person.
- id: approval_timeout
type: DURATION
displayName: Approval timeout
defaults: PT4H
description: When nobody decides in time, the execution fails and the batch
stays pending. The next run asks again.
- id: max_records
type: INT
displayName: Maximum change events per run
defaults: 10000
variables:
state_name: ddl-gate
# The Capture task stores offsets and schema history under this KV key. The check_state_key task fails if the key changes.
state_key: "{{ flow.id }}_states_ddl-gate_debezium-state.dat"
pending_key: "{{ flow.id }}_pending_batch"
decision_key: "{{ flow.id }}_decision_open"
checkpoint_key: "{{ flow.id }}_last_load"
source_jdbc_url: "jdbc:mysql://{{ inputs.source_host }}:{{ inputs.source_port
}}/information_schema?connectTimeout=10000"
tasks:
- id: decision_open
type: io.kestra.plugin.core.flow.If
description: A paused execution does not count for the concurrency limit. This
check stops a second run while a decision on the pending batch is open.
condition: "{{ kv(render(vars.decision_key), errorOnMissing=false) is not null }}"
then:
- id: decision_open_log
type: io.kestra.plugin.core.log.Log
level: WARN
message: "Execution {{ kv(render(vars.decision_key)) }} waits for a decision on
the pending batch. This run does nothing."
- id: decision_open_exit
type: io.kestra.plugin.core.execution.Exit
state: WARNING
- id: take_batch
type: io.kestra.plugin.core.flow.If
description: Capture new events only when no earlier batch is pending. The
capture moves the connector offset, so a pending batch must load first.
condition: "{{ kv(render(vars.pending_key), errorOnMissing=false) is null }}"
then:
- id: copy_history
type: io.kestra.plugin.core.kv.Set
description: Copy the connector state before the capture. The detect_ddl task
compares it with the state after the capture.
key: "{{ flow.id }}_history_before_{{ execution.id }}"
kvType: JSON
ttl: P90D
value: "{{ (kv(render(vars.state_key), errorOnMissing=false) ?? {}) | toJson }}"
- id: capture
type: io.kestra.plugin.debezium.mysql.Capture
description: Read new row events of the watched tables from the binlog. The
first run takes an initial snapshot. Debezium also appends every DDL
it parses to its schema history.
hostname: "{{ inputs.source_host }}"
port: "{{ inputs.source_port }}"
username: "{{ secret('MYSQL_CDC_USERNAME') }}"
password: "{{ secret('MYSQL_CDC_PASSWORD') }}"
serverId: "{{ inputs.source_server_id }}"
stateName: "{{ vars.state_name }}"
snapshotMode: INITIAL
ignoreDdl: false
format: INLINE
splitTable: TABLE
key: DROP
metadata: DROP
deleted: ADD_FIELD
deletedFieldName: __deleted
# The watched tables. The plugin reads this list as written, so put literal regular expressions here.
includedTables:
- 'shop\.orders'
- 'shop\.customers'
properties:
# Keep DDL of the watched tables only in the schema history, so every new history entry concerns a watched table.
schema.history.internal.store.only.captured.tables.ddl: "true"
maxRecords: "{{ inputs.max_records }}"
maxDuration: PT1M
maxWait: PT30S
timeout: PT30M
- id: save_pending
type: io.kestra.plugin.core.kv.Set
description: Record the batch right after the capture moved the offset. It stays
pending until the load succeeds, so no captured event is lost.
key: "{{ render(vars.pending_key) }}"
kvType: JSON
value: |
{{ {'execution': execution.id, 'captured_at': now(), 'size': outputs.capture.size, 'uris': outputs.capture.uris, 'history_copy_key': flow.id ~ '_history_before_' ~ execution.id} | toJson }}
- id: check_state_key
type: io.kestra.plugin.core.execution.Fail
description: The history copy and the DDL detection depend on the KV key that
the plugin uses. Stop if a plugin upgrade changes it. The batch stays
pending.
condition: "{{ outputs.capture.stateOffsetKey != render(vars.state_key) }}"
errorMessage: "The Debezium state key is {{ outputs.capture.stateOffsetKey }},
expected {{ render(vars.state_key) }}. Update vars.state_key before
the next run."
else:
- id: retry_pending
type: io.kestra.plugin.core.log.Log
level: WARN
message: "Batch of execution {{ kv(render(vars.pending_key)).execution }} is
still pending. This run retries it and reads no new events."
- id: batch
type: io.kestra.plugin.core.kv.Get
description: The batch to gate and load, new or retried.
key: "{{ render(vars.pending_key) }}"
errorOnMissing: true
- id: detect_ddl
type: io.kestra.plugin.scripts.python.Script
description: Compare the schema history before and after the capture. Keep only
the new, non-snapshot DDL that touches a watched table, and classify each
change as additive or breaking.
containerImage: python:3.12-slim
timeout: PT10M
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
inputFiles:
before.json: "{{ kv(outputs.batch.value.history_copy_key) | toJson }}"
# No fallback here: a missing connector state must fail the run, not hide a schema change.
after.json: "{{ kv(render(vars.state_key)) | toJson }}"
script: |
import base64, json, re
def load_history(path):
with open(path, encoding="utf-8") as f:
state = json.load(f) or {}
raw = state.get("history") or ""
text = base64.b64decode(raw).decode("utf-8") if raw else ""
return [json.loads(line) for line in text.splitlines() if line.strip()]
def table_id(raw):
return raw.replace('"', "").replace("`", "").lower()
def columns(table):
return {c["name"]: c for c in (table or {}).get("columns", [])}
def shape(col):
return [col.get(k) for k in ("typeName", "length", "scale", "optional", "autoIncremented", "generated")]
before = load_history("before.json")
after = load_history("after.json")
if not after:
raise SystemExit("The connector state has no schema history. Do not load rows without a DDL check.")
if after[:len(before)] != before:
raise SystemExit("The schema history after the capture does not extend the previous history. Inspect the connector state.")
# The history holds only captured tables, so every table it names is a watched table.
watched = {table_id(c["id"]) for e in after for c in (e.get("tableChanges") or [])}
known = {}
for entry in before:
for change in entry.get("tableChanges") or []:
known[table_id(change["id"])] = None if change.get("type") == "DROP" else change.get("table")
events, statements, ignored, snapshot_ddl = [], [], 0, 0
for entry in after[len(before):]:
ddl = entry.get("ddl") or ""
db = (entry.get("databaseName") or "").lower()
pos = entry.get("position") or {}
changes = entry.get("tableChanges") or []
hits = [c for c in changes if table_id(c["id"]) in watched]
if not changes:
# A statement without parsed table changes still blocks when it names a watched table.
hits = [{"id": t, "type": "UNPARSED"} for t in sorted(watched)
if re.search(r"\b" + re.escape(t.split(".")[1]) + r"\b", ddl, re.I) and (not db or t.startswith(db + "."))]
if pos.get("snapshot"):
snapshot_ddl += 1
elif not hits:
ignored += 1
else:
target_sql = re.sub(r"`?" + re.escape(db) + r"`?\.(?=`?\w)", "", ddl, flags=re.I) if db else ddl
statements.append(target_sql)
for change in hits:
tid = table_id(change["id"])
old_cols, new_cols = columns(known.get(tid)), columns(change.get("table"))
added = sorted(set(new_cols) - set(old_cols))
dropped = sorted(set(old_cols) - set(new_cols))
changed = sorted(c for c in set(old_cols) & set(new_cols) if shape(old_cols[c]) != shape(new_cols[c]))
same_key = (known.get(tid) or {}).get("primaryKeyColumnNames") == (change.get("table") or {}).get("primaryKeyColumnNames")
additive = (change.get("type") == "ALTER" and tid in known and not dropped and not changed and same_key
and all(new_cols[c].get("optional") or new_cols[c].get("hasDefaultValue") for c in added))
events.append({"table": tid, "type": change.get("type"), "additive": additive,
"added": added, "dropped": dropped, "changed": changed, "ddl": ddl,
"binlog": "{}:{}".format(pos.get("file"), pos.get("pos")), "gtids": pos.get("gtids")})
for change in changes:
known[table_id(change["id"])] = None if change.get("type") == "DROP" else change.get("table")
lines = ["{} {} ({}) at {}: {} [added={} dropped={} changed={}]".format(
e["table"], e["type"], "additive" if e["additive"] else "breaking", e["binlog"], e["ddl"],
",".join(e["added"]) or "-", ",".join(e["dropped"]) or "-", ",".join(e["changed"]) or "-") for e in events]
dropped_columns = {}
for e in events:
dropped_columns.setdefault(e["table"], []).extend(e["dropped"])
result = {"ddl_count": len(events), "breaking": any(not e["additive"] for e in events), "events": events,
"statements": statements, "tables": sorted({e["table"] for e in events}), "dropped_columns": dropped_columns,
"ignored_ddl": ignored, "snapshot_ddl": snapshot_ddl, "summary": "\n".join(lines)}
print("::" + json.dumps({"outputs": result}) + "::")
- id: run_summary
type: io.kestra.plugin.core.log.Log
message: >-
Batch of execution {{ outputs.batch.value.execution }}: {{
outputs.batch.value.size }} change event(s) in {{ outputs.batch.value.uris
| keys | join(', ') | default('no table') }}. Watched DDL: {{
outputs.detect_ddl.vars.ddl_count }} (breaking={{
outputs.detect_ddl.vars.breaking }}). Other history entries: {{
outputs.detect_ddl.vars.ignored_ddl }}. Snapshot DDL: {{
outputs.detect_ddl.vars.snapshot_ddl }}. {{
outputs.detect_ddl.vars.summary }}
- id: ddl_gate
type: io.kestra.plugin.core.flow.If
description: Hold the batch when a schema change on a watched table is in it.
condition: "{{ outputs.detect_ddl.vars.ddl_count > 0 }}"
then:
- id: notify_ddl
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{{ {'text': 'Schema change on watched MySQL table(s) ' ~ (outputs.detect_ddl.vars.tables | join(', ')) ~ '. ' ~ outputs.batch.value.size ~ ' row event(s) are held. ' ~ ((inputs.auto_apply_additive and not outputs.detect_ddl.vars.breaking) ? 'The change is additive and auto_apply_additive is true, so it is applied now.' : 'Resume execution ' ~ execution.id ~ ' with APPLY_DDL_AND_LOAD or LOAD_ONLY. To hold the batch, do not resume. At the approval timeout the execution fails, the batch stays pending and the next run asks again.') ~ ' Changes: ' ~ outputs.detect_ddl.vars.summary} | toJson }}
retry:
type: constant
interval: PT20S
maxAttempts: 3
- id: needs_approval
type: io.kestra.plugin.core.flow.If
condition: "{{ not (inputs.auto_apply_additive and not
outputs.detect_ddl.vars.breaking) }}"
then:
- id: claim_decision
type: io.kestra.plugin.core.kv.Set
description: Mark the decision as open. The key expires with the approval
timeout, also when the execution is killed.
key: "{{ render(vars.decision_key) }}"
value: "{{ execution.id }}"
overwrite: false
ttl: "{{ inputs.approval_timeout }}"
- id: approval
type: io.kestra.plugin.core.flow.Pause
description: APPLY_DDL_AND_LOAD runs the captured DDL on the target, checks the
schema and loads. LOAD_ONLY checks the schema and loads, for a
target that you changed by hand. To hold the batch, do not resume.
At the timeout the execution fails and the batch stays pending for
the next run.
pauseDuration: "{{ inputs.approval_timeout }}"
behavior: FAIL
onResume:
- id: decision
type: SELECT
values:
- APPLY_DDL_AND_LOAD
- LOAD_ONLY
- id: reviewer_note
type: STRING
required: false
- id: route_decision
type: io.kestra.plugin.core.flow.Switch
value: "{{ (inputs.auto_apply_additive and not outputs.detect_ddl.vars.breaking)
? 'APPLY_DDL_AND_LOAD' : outputs.approval.onResume.decision }}"
cases:
APPLY_DDL_AND_LOAD:
- id: apply_ddl
type: io.kestra.plugin.core.flow.Loop
description: Run the captured statements on the target in binlog order, with the
source database name removed.
values: "{{ outputs.detect_ddl.vars.statements }}"
concurrencyLimit: 1
tasks:
- id: run_statement
type: io.kestra.plugin.jdbc.mysql.Query
url: "{{ inputs.target_jdbc_url }}"
username: "{{ secret('TARGET_DB_USERNAME') }}"
password: "{{ secret('TARGET_DB_PASSWORD') }}"
sql: "{{ item.value }}"
fetchType: NONE
LOAD_ONLY:
- id: target_changed_by_hand
type: io.kestra.plugin.core.log.Log
message: "LOAD_ONLY chosen. The schema check must pass before any row loads.
Note: {{ outputs.approval.onResume.reviewer_note ?? '-' }}"
defaults:
- id: unknown_decision
type: io.kestra.plugin.core.execution.Fail
errorMessage: "Unknown decision. Nothing was loaded."
- id: close_decision
type: io.kestra.plugin.core.kv.Delete
key: "{{ render(vars.decision_key) }}"
- id: verify_schema
type: io.kestra.plugin.core.flow.Loop
description: For each changed table, the target columns must now match the
source columns by name, type and nullability.
values: "{{ outputs.detect_ddl.vars.tables }}"
concurrencyLimit: 1
tasks:
- id: source_columns
type: io.kestra.plugin.jdbc.mysql.Query
url: "{{ render(vars.source_jdbc_url) }}"
username: "{{ secret('MYSQL_CDC_USERNAME') }}"
password: "{{ secret('MYSQL_CDC_PASSWORD') }}"
fetchType: FETCH_ONE
sql: >-
SELECT COUNT(*) AS column_count,
COALESCE(GROUP_CONCAT(CONCAT(COLUMN_NAME, ' ', COLUMN_TYPE, IF(IS_NULLABLE = 'YES', ' NULL', ' NOT NULL')) ORDER BY ORDINAL_POSITION SEPARATOR ', '), '') AS signature
FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = '{{
item.value | split('[.]') | first }}' AND TABLE_NAME = '{{
item.value | split('[.]') | last }}'
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: target_columns
type: io.kestra.plugin.jdbc.mysql.Query
url: "{{ inputs.target_jdbc_url }}"
username: "{{ secret('TARGET_DB_USERNAME') }}"
password: "{{ secret('TARGET_DB_PASSWORD') }}"
fetchType: FETCH_ONE
sql: >-
SELECT COUNT(*) AS column_count,
COALESCE(GROUP_CONCAT(CONCAT(COLUMN_NAME, ' ', COLUMN_TYPE, IF(IS_NULLABLE = 'YES', ' NULL', ' NOT NULL')) ORDER BY ORDINAL_POSITION SEPARATOR ', '), '') AS signature
FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE()
AND TABLE_NAME = '{{ item.value | split('[.]') | last }}'
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: schema_mismatch
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.source_columns.row.signature !=
outputs.target_columns.row.signature }}"
errorMessage: "Target table {{ item.value | split('[.]') | last }} does not
match the source after the decision. Source: [{{
outputs.source_columns.row.signature }}]. Target: [{{
outputs.target_columns.row.signature }}]. Nothing was loaded."
- id: load_changes
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.batch.value.uris | length > 0 }}"
then:
- id: load_tables
type: io.kestra.plugin.core.flow.Loop
description: One table at a time. Each table keeps only the last event per
primary key, then upserts and deletes.
values: "{{ outputs.batch.value.uris | keys }}"
concurrencyLimit: 1
tasks:
- id: target_plan
type: io.kestra.plugin.jdbc.mysql.Query
description: The target database builds the upsert and delete statements from
its own column list and primary key. Debezium sends DATETIME and
TIMESTAMP values as ISO-8601 UTC text, so the upsert converts
them.
url: "{{ inputs.target_jdbc_url }}"
username: "{{ secret('TARGET_DB_USERNAME') }}"
password: "{{ secret('TARGET_DB_PASSWORD') }}"
fetchType: FETCH_ONE
sql: |
SELECT COUNT(*) AS column_count,
CONCAT('[', GROUP_CONCAT(JSON_QUOTE(c.COLUMN_NAME) ORDER BY c.ORDINAL_POSITION), ']') AS columns_json,
CONCAT('INSERT INTO `', MAX(c.TABLE_NAME), '` (',
GROUP_CONCAT(CONCAT('`', c.COLUMN_NAME, '`') ORDER BY c.ORDINAL_POSITION SEPARATOR ', '), ') VALUES (',
GROUP_CONCAT(IF(c.IS_NULLABLE = 'NO' AND c.COLUMN_DEFAULT IS NOT NULL AND c.EXTRA NOT LIKE '%DEFAULT_GENERATED%',
CONCAT('IFNULL(', c.param_expr, ', DEFAULT(`', c.COLUMN_NAME, '`))'), c.param_expr) ORDER BY c.ORDINAL_POSITION SEPARATOR ', '),
') AS new ON DUPLICATE KEY UPDATE ',
GROUP_CONCAT(CONCAT('`', c.COLUMN_NAME, '` = new.`', c.COLUMN_NAME, '`') ORDER BY c.ORDINAL_POSITION SEPARATOR ', ')) AS upsert_sql,
(SELECT CONCAT('[', GROUP_CONCAT(JSON_QUOTE(k.COLUMN_NAME) ORDER BY k.ORDINAL_POSITION), ']')
FROM information_schema.KEY_COLUMN_USAGE k
WHERE k.TABLE_SCHEMA = DATABASE() AND k.TABLE_NAME = MAX(c.TABLE_NAME) AND k.CONSTRAINT_NAME = 'PRIMARY') AS key_json,
(SELECT CONCAT('DELETE FROM `', MAX(k.TABLE_NAME), '` WHERE ', GROUP_CONCAT(CONCAT('`', k.COLUMN_NAME, '` = ?') ORDER BY k.ORDINAL_POSITION SEPARATOR ' AND '))
FROM information_schema.KEY_COLUMN_USAGE k
WHERE k.TABLE_SCHEMA = DATABASE() AND k.TABLE_NAME = MAX(c.TABLE_NAME) AND k.CONSTRAINT_NAME = 'PRIMARY') AS delete_sql
FROM (SELECT TABLE_NAME, COLUMN_NAME, ORDINAL_POSITION, IS_NULLABLE, COLUMN_DEFAULT, EXTRA,
IF(DATA_TYPE IN ('datetime', 'timestamp'), 'CAST(REPLACE(REPLACE(?, ''T'', '' ''), ''Z'', '''') AS DATETIME(6))', '?') AS param_expr
FROM information_schema.COLUMNS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '{{ item.value | split('[.]') | last }}' AND EXTRA NOT LIKE '%GENERATED%') c
retry:
type: constant
interval: PT10S
maxAttempts: 3
- id: target_ready
type: io.kestra.plugin.core.execution.Fail
condition: "{{ outputs.target_plan.row.column_count == 0 or
outputs.target_plan.row.key_json is null }}"
errorMessage: "Target table {{ item.value | split('[.]') | last }} is missing or
has no primary key. Nothing was loaded for it."
- id: tag_keys
type: io.kestra.plugin.graalvm.js.FileTransform
description: Add the primary key as one field for the deduplication. Stop when a
row has a column that the target table does not have, unless an
approved DDL in this batch dropped it.
from: "{{ outputs.batch.value.uris[item.value] }}"
script: |
var keyColumns = {{ outputs.target_plan.row.key_json }};
var targetColumns = {{ outputs.target_plan.row.columns_json }};
var droppedColumns = {{ (outputs.detect_ddl.vars.dropped_columns[item.value | lower] ?? []) | toJson }};
for (var name in row) {
if (name.indexOf('__') !== 0 && targetColumns.indexOf(name) < 0 && droppedColumns.indexOf(name) < 0) {
throw new Error('Column ' + name + ' of {{ item.value }} is not in the target table. Run the DDL gate before loading.');
}
}
row['__key'] = JSON.stringify(keyColumns.map(function (c) { return String(row[c]); }));
- id: last_event_per_key
type: io.kestra.plugin.core.storage.DeduplicateItems
from: "{{ outputs.tag_keys.uri }}"
expr: "{{ __key }}"
- id: upsert_rows
type: io.kestra.plugin.core.storage.FilterItems
from: "{{ outputs.last_event_per_key.uri }}"
filterCondition: "{{ __deleted == true }}"
filterType: EXCLUDE
- id: delete_rows
type: io.kestra.plugin.core.storage.FilterItems
from: "{{ outputs.last_event_per_key.uri }}"
filterCondition: "{{ __deleted == true }}"
filterType: INCLUDE
- id: apply_upserts
type: io.kestra.plugin.jdbc.mysql.Batch
from: "{{ outputs.upsert_rows.uri }}"
url: "{{ inputs.target_jdbc_url }}"
username: "{{ secret('TARGET_DB_USERNAME') }}"
password: "{{ secret('TARGET_DB_PASSWORD') }}"
sql: "{{ outputs.target_plan.row.upsert_sql }}"
columns: "{{ fromJson(outputs.target_plan.row.columns_json) }}"
- id: apply_deletes
type: io.kestra.plugin.jdbc.mysql.Batch
from: "{{ outputs.delete_rows.uri }}"
url: "{{ inputs.target_jdbc_url }}"
username: "{{ secret('TARGET_DB_USERNAME') }}"
password: "{{ secret('TARGET_DB_PASSWORD') }}"
sql: "{{ outputs.target_plan.row.delete_sql }}"
columns: "{{ fromJson(outputs.target_plan.row.key_json) }}"
- id: save_checkpoint
type: io.kestra.plugin.core.kv.Set
description: Record what this run loaded, for audits and for the next on-call engineer.
key: "{{ render(vars.checkpoint_key) }}"
kvType: JSON
value: |
{{ {'execution': execution.id, 'captured_by': outputs.batch.value.execution, 'finished_at': now(), 'events': outputs.batch.value.size, 'tables': outputs.batch.value.uris | keys, 'ddl_applied': outputs.detect_ddl.vars.statements, 'decision': (outputs.detect_ddl.vars.ddl_count == 0) ? 'NO_DDL' : ((inputs.auto_apply_additive and not outputs.detect_ddl.vars.breaking) ? 'AUTO_ADDITIVE' : outputs.approval.onResume.decision)} | toJson }}
- id: clear_pending
type: io.kestra.plugin.core.kv.Delete
description: The batch is loaded. The next run captures new events.
key: "{{ render(vars.pending_key) }}"
- id: drop_history_copy
type: io.kestra.plugin.core.kv.Delete
key: "{{ outputs.batch.value.history_copy_key }}"
errors:
- id: close_decision_on_failure
type: io.kestra.plugin.core.flow.If
description: A timed out or failed decision is closed, so the next run asks again.
condition: "{{ kv(render(vars.decision_key), errorOnMissing=false) == execution.id }}"
then:
- id: delete_decision_key
type: io.kestra.plugin.core.kv.Delete
key: "{{ render(vars.decision_key) }}"
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
payload: |
{{ {'text': (kv(render(vars.pending_key), errorOnMissing=false) is not null) ? ('Debezium MySQL DDL gate: execution ' ~ execution.id ~ ' FAILED. The batch of execution ' ~ kv(render(vars.pending_key)).execution ~ ' is not loaded and stays pending. The next run retries it. Read the execution logs.') : ('Debezium MySQL DDL gate: execution ' ~ execution.id ~ ' FAILED before a batch was captured from ' ~ inputs.source_host ~ '. The offset did not move. Read the execution logs.')} | toJson }}
retry:
type: constant
interval: PT20S
maxAttempts: 3
triggers:
- id: every_five_minutes
type: io.kestra.plugin.core.trigger.Schedule
description: A schedule, not the Debezium trigger, because the flow must control
when the offset moves.
cron: "*/5 * * * *"
A MySQL schema change can break a downstream copy in silence. This flow replicates watched tables with Debezium and holds the load when DDL changes one of them.
How it works
- A Schedule runs the flow every five minutes.
- With no pending batch, the Debezium MySQL Capture task reads new binlog events. The flow saves the batch as pending in KV.
- The plugin does not emit DDL events. A Python script compares the Debezium schema history before and after the capture and marks each watched change as additive or breaking.
- With watched DDL, the flow sends a Slack message and pauses. APPLY_DDL_AND_LOAD runs the DDL on the target. LOAD_ONLY is for a target that you changed by hand. The target columns must then match the source.
- Per table, the flow keeps the last event per primary key and loads an upsert and a delete with the JDBC Batch task.
- The pending batch is deleted only after the load. After a failure or an approval timeout, the next run retries it before it reads new events.
What you get
- No load into a table whose schema changed without review.
- No skipped events, and an idempotent load.
- Optional automatic apply of additive columns.
- A KV checkpoint with the decision and the applied DDL.
Who it's for
- Data and platform engineers who run MySQL CDC.
- On-call owners of a downstream MySQL copy.
Why orchestrate this with Kestra
- Pause with onResume inputs gives an audited approval step.
- KV state, retries, concurrency and alerts are declared in YAML.
Prerequisites
- MySQL with binlog_format ROW and binlog_row_image FULL. Keep binary logs longer than the approval timeout.
- A Debezium user with SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT and LOCK TABLES.
- Target tables with the same primary keys. The target user needs DML, ALTER, CREATE, DROP and INDEX.
- A Docker task runner.
Secrets
MYSQL_CDC_USERNAMEandMYSQL_CDC_PASSWORD: Debezium user on the source.TARGET_DB_USERNAMEandTARGET_DB_PASSWORD: user that writes to the target.SLACK_WEBHOOK_URL: webhook for decisions and failures.
Quick start
- Create the secrets and edit includedTables in the capture task.
- Execute the flow once. The first run loads a snapshot.
- Run an ALTER on a watched table, then resume the paused execution with a decision.
- Enable the schedule.
How to extend
- Send the decision to Microsoft Teams.
- Load a second target after the schema check.
Links
Tasks
Orchestrate with Kestra
Orchestrate Slack with KestraCreate a Blueprint
Go to the developer platformMore Data Blueprints
