id: kafka-throttled-partition-reassignment
namespace: company.team
description: |
Move Kafka partitions onto a new set of brokers (to drain a broker before decommissioning it, or to fill a new one)
without starving producers and consumers. The flow plans the smallest move instead of kafka-reassign-partitions'
random --generate, checks the cluster is healthy, saves the current assignment for rollback, asks for approval,
runs the move under a replication throttle, and always removes the throttle afterwards.
triggers:
- id: move_request
type: io.kestra.plugin.core.trigger.Webhook
description: |
Called by a capacity or decommissioning runbook with the topics and the brokers they should end up on:
curl -X POST -H 'Content-Type: application/json' -d '{"topics":"orders,payments","target_brokers":"1,2,4"}' https://kestra.example.com/api/v1/main/executions/webhook/company.team/kafka-throttled-partition-reassignment/<key>
key: "{{ secret('KAFKA_REASSIGN_WEBHOOK_KEY') }}"
concurrency:
limit: 1
behavior: QUEUE
inputs:
- id: bootstrap_servers
type: STRING
defaults: 127.0.0.1:9091,127.0.0.1:9092,127.0.0.1:9093
- id: topics
type: STRING
defaults: orders,payments
description: Comma-separated topics to move. The webhook body's "topics" overrides it.
- id: target_brokers
type: STRING
defaults: 1,2,4
description: The broker ids the topics should live on afterwards. To drain
broker 3, list every other broker. The webhook body's "target_brokers"
overrides it.
- id: throttle_mb_per_sec
type: INT
defaults: 10
description: Replication limit per broker in MiB/s while the move runs, on both
the sending and the receiving side. Leave room for normal replication; a
third of the network is a common start.
- id: max_minutes
type: INT
defaults: 120
description: How long the move may take after approval. The plan's estimate is
shown before you approve.
- id: on_timeout
type: SELECT
values: [ cancel, keep_running ]
defaults: cancel
description: cancel stops the move, puts every partition back on its original
replicas in the original order, and removes the throttle. keep_running
leaves the move and its throttle in place and tells you how to finish it.
- id: kafka_image
type: STRING
defaults: apache/kafka:3.9.1
description: Its CLI tools run every Kafka step. Use the version of your brokers
or newer.
- id: python_image
type: STRING
defaults: python:3.12-slim
variables:
topics: "{{ trigger.body.topics ?? inputs.topics }}"
targets: "{{ trigger.body.target_brokers ?? inputs.target_brokers }}"
rollback_key: kafka_reassignment_rollback
tasks:
- id: inspect
type: io.kestra.plugin.scripts.shell.Commands
description: Reads the cluster with the stock Kafka tools and keeps their raw
output as files for the plan. Changes nothing.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: host
pullPolicy: IF_NOT_PRESENT
user: root
containerImage: "{{ inputs.kafka_image }}"
env:
BOOTSTRAP: "{{ inputs.bootstrap_servers }}"
TOPICS: "{{ render(vars.topics) }}"
TARGETS: "{{ render(vars.targets) }}"
outputFiles:
- "*.txt"
- "*.json"
commands:
- |
k=/opt/kafka/bin; B="--bootstrap-server $BOOTSTRAP"
fail() { echo "MOVE FAILED: $*" >&2; exit 1; }
$k/kafka-broker-api-versions.sh $B > api.txt 2> err.txt || fail "cannot reach $BOOTSTRAP: $(grep -m1 -iE 'exception|error' err.txt)"
grep -oE '\(id: [0-9]+' api.txt | grep -oE '[0-9]+' | sort -n | paste -sd, - > brokers.txt
missing=""; for t in $(echo "$TARGETS" | tr ',' ' '); do tr ',' '\n' < brokers.txt | grep -qx "$t" || missing="$missing $t"; done
[ -z "$missing" ] || fail "target broker(s)$missing are not in the cluster (brokers answering: $(cat brokers.txt)). Nothing was changed"
printf '{"version":1,"topics":[%s]}' "$(echo "$TOPICS" | tr ',' '\n' | sed '/^ *$/d; s/^ *//; s/ *$//; s/.*/{"topic":"&"}/' | paste -sd, -)" > topics.json
$k/kafka-reassign-partitions.sh $B --generate --topics-to-move-json-file topics.json --broker-list "$TARGETS" > generate.txt 2>&1 \
|| fail "kafka-reassign-partitions --generate refused the request: $(grep -m1 -iE 'error|exception' generate.txt)"
sed -n '/^Current partition replica assignment/{n;p;}' generate.txt > current.json
sed -n '/^Proposed partition reassignment configuration/{n;p;}' generate.txt > generated.json
$k/kafka-topics.sh $B --describe --under-replicated-partitions > urp.txt 2>&1
$k/kafka-topics.sh $B --describe --unavailable-partitions > unavailable.txt 2>&1
$k/kafka-reassign-partitions.sh $B --list > ongoing.txt 2>&1
$k/kafka-log-dirs.sh $B --describe --topic-list "$TOPICS" 2>/dev/null | tail -n 1 > sizes.json
$k/kafka-configs.sh $B --describe --entity-type brokers > throttles.txt 2>&1
echo "brokers $(cat brokers.txt); $(grep -c . urp.txt) under-replicated partitions; $(grep -c 'adding:' ongoing.txt) reassignments running"
- id: plan
type: io.kestra.plugin.scripts.python.Script
description: Plans the smallest move, checks the preconditions, and sizes the
data each broker receives.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
pullPolicy: IF_NOT_PRESENT
containerImage: "{{ inputs.python_image }}"
inputFiles:
brokers.txt: "{{ outputs.inspect.outputFiles['brokers.txt'] }}"
current.json: "{{ outputs.inspect.outputFiles['current.json'] }}"
generated.json: "{{ outputs.inspect.outputFiles['generated.json'] }}"
urp.txt: "{{ outputs.inspect.outputFiles['urp.txt'] }}"
unavailable.txt: "{{ outputs.inspect.outputFiles['unavailable.txt'] }}"
ongoing.txt: "{{ outputs.inspect.outputFiles['ongoing.txt'] }}"
sizes.json: "{{ outputs.inspect.outputFiles['sizes.json'] }}"
throttles.txt: "{{ outputs.inspect.outputFiles['throttles.txt'] }}"
env:
TARGETS: "{{ render(vars.targets) }}"
THROTTLE_MB: "{{ inputs.throttle_mb_per_sec }}"
outputFiles:
- plan.json
- rollback.json
- partitions.json
script: |
import json, os, re
from collections import Counter
alive = [int(b) for b in open("brokers.txt").read().strip().split(",") if b]
targets = [int(b) for b in os.environ["TARGETS"].replace(" ", "").split(",") if b]
current = json.load(open("current.json"))["partitions"]
generated = {(p["topic"], p["partition"]): p["replicas"] for p in json.load(open("generated.json"))["partitions"]}
problems, warnings = [], []
missing = [b for b in targets if b not in alive]
if missing:
problems.append(f"target broker(s) {missing} are not in the cluster (alive: {alive})")
max_rf = max(len(p["replicas"]) for p in current)
if len(set(targets) - set(missing)) < max_rf:
problems.append(f"replication factor {max_rf} needs at least {max_rf} target brokers, got {targets}")
urp = [l.split("Partition:")[0].replace("Topic:", "").strip() + "-" + l.split("Partition:")[1].split()[0] for l in open("urp.txt") if "Partition:" in l]
if urp:
problems.append(f"{len(urp)} partition(s) are under-replicated ({', '.join(urp[:5])}): moving data now removes a replica that may be the only in-sync copy. Fix the cluster first")
offline = [l for l in open("unavailable.txt") if "Partition:" in l]
if offline:
problems.append(f"{len(offline)} partition(s) have no leader")
running = [l.split(":")[0] for l in open("ongoing.txt") if "adding:" in l]
if running:
problems.append(f"a reassignment is already running ({', '.join(running[:5])}): Kafka would merge both moves. Let it finish or cancel it first")
throttled, broker = {}, None
for line in open("throttles.txt"):
m = re.match(r"Dynamic configs for broker (\d+)", line)
broker = m.group(1) if m else broker
m = re.search(r"replication\.throttled\.rate=(\d+)", line)
if m and broker:
throttled[broker] = int(m.group(1))
if throttled:
warnings.append("leftover replication throttle from an earlier move whose --verify never ran: " + ", ".join(f"broker {b} at {r / 1048576:.1f} MiB/s" for b, r in sorted(throttled.items())) + ". This move replaces it and removes it at the end")
# Smallest move: a replica stays where it is if its broker is a target; only replicas on other brokers are
# replaced, in place, by the least loaded target broker. --generate instead reshuffles every partition.
load = Counter(b for p in current for b in p["replicas"] if b in targets)
plan, rollback = [], []
for p in current:
new = list(p["replicas"])
for i, b in enumerate(new):
if b not in targets:
pick = min((t for t in targets if t not in new and t not in missing), key=lambda t: (load[t], t), default=None)
if pick is None:
break
new[i] = pick; load[pick] += 1
if new != p["replicas"]:
plan.append({"topic": p["topic"], "partition": p["partition"], "replicas": new})
rollback.append({"topic": p["topic"], "partition": p["partition"], "replicas": p["replicas"]})
sizes = {}
for broker in json.load(open("sizes.json")).get("brokers", []) if os.path.getsize("sizes.json") else []:
for d in broker["logDirs"]:
for part in d["partitions"]:
sizes[part["partition"]] = max(sizes.get(part["partition"], 0), part["size"])
incoming, replica_moves = Counter(), 0
for p, old in zip(plan, rollback):
for b in set(p["replicas"]) - set(old["replicas"]):
incoming[b] += sizes.get(f'{p["topic"]}-{p["partition"]}', 0); replica_moves += 1
gen_moves = sum(len(set(r) - set(p["replicas"])) for p in current for r in [generated.get((p["topic"], p["partition"]), p["replicas"])])
rate = int(os.environ["THROTTLE_MB"]) * 1048576
eta_s = max(incoming.values(), default=0) / rate if rate else 0
json.dump({"version": 1, "partitions": plan}, open("plan.json", "w"))
json.dump({"version": 1, "partitions": rollback}, open("rollback.json", "w"))
json.dump({"partitions": [{"topic": p["topic"], "partition": p["partition"]} for p in plan]}, open("partitions.json", "w"))
out = {"problems": "; ".join(problems), "warnings": "; ".join(warnings), "partitions": len(plan),
"replica_moves": replica_moves, "generate_replica_moves": gen_moves,
"incoming_mb": {str(b): round(v / 1048576, 1) for b, v in sorted(incoming.items())},
"total_mb": round(sum(incoming.values()) / 1048576, 1), "eta_minutes": round(eta_s / 60, 1),
"moves": [f'{p["topic"]}-{p["partition"]}: {o["replicas"]} -> {p["replicas"]}' for p, o in zip(plan, rollback)][:20]}
print("::" + json.dumps({"outputs": out}) + "::")
- id: preflight
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.plan.vars.problems != '' }}"
then:
- id: refuse
type: io.kestra.plugin.core.execution.Fail
errorMessage: "MOVE FAILED: not started, nothing changed: {{
outputs.plan.vars.problems }}"
- id: anything_to_move
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.plan.vars.partitions == 0 }}"
then:
- id: report_nothing
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Kafka reassignment: every partition of {{ render(vars.topics) }}
already lives on brokers {{ render(vars.targets) }}. Nothing to move."
- id: done_already
type: io.kestra.plugin.core.execution.Exit
state: SUCCESS
- id: save_rollback
type: io.kestra.plugin.core.kv.Set
description: The assignment before the move, so it can be restored even if
Kestra is not around to do it.
key: "{{ vars.rollback_key }}"
kvType: JSON
value: "{{ {'execution': execution.id, 'at': execution.startDate,
'reassignment': fromJson(read(outputs.plan.outputFiles['rollback.json']))}
| toJson }}"
- id: request_approval
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: |
Kafka reassignment ready: {{ outputs.plan.vars.replica_moves }} replica(s) of {{ outputs.plan.vars.partitions }} partition(s) of {{ render(vars.topics) }} onto brokers {{ render(vars.targets) }} (the random plan of `--generate` for the same request moved {{ outputs.plan.vars.generate_replica_moves }}).
Data to copy: {{ outputs.plan.vars.total_mb }} MiB ({% for e in outputs.plan.vars.incoming_mb %}broker {{ e.key }}: {{ e.value }} MiB{% if not loop.last %}, {% endif %}{% endfor %}), about {{ outputs.plan.vars.eta_minutes }} min at {{ inputs.throttle_mb_per_sec }} MiB/s per broker. Time limit after approval: {{ inputs.max_minutes }} min, then {{ inputs.on_timeout == 'cancel' ? 'cancel and roll back' : 'leave it running' }}.
{% for m in outputs.plan.vars.moves %}{{ m }}
{% endfor %}{% if outputs.plan.vars.warnings != '' %}
Note: {{ outputs.plan.vars.warnings }}{% endif %}
Resume execution {{ execution.id }} in Kestra to approve. The current assignment is saved in KV `{{ vars.rollback_key }}`.
- id: approve_move
type: io.kestra.plugin.core.flow.Pause
description: Waits up to 4 hours for a decision. No decision means no move.
pauseDuration: PT4H
behavior: RESUME
onResume:
- id: approve
type: BOOL
displayName: Start the reassignment now?
defaults: false
- id: approval_gate
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.approve_move.onResume.approve ?? false }}"
then:
- id: move
type: io.kestra.plugin.scripts.shell.Commands
description: Re-checks that nothing else started a move during the approval,
starts the throttled move, follows it with --verify until it completes
or the time limit is reached, and cancels and rolls back if asked to.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: host
pullPolicy: IF_NOT_PRESENT
user: root
containerImage: "{{ inputs.kafka_image }}"
timeout: "PT{{ inputs.max_minutes + 15 }}M"
inputFiles:
plan.json: "{{ outputs.plan.outputFiles['plan.json'] }}"
rollback.json: "{{ outputs.plan.outputFiles['rollback.json'] }}"
env:
BOOTSTRAP: "{{ inputs.bootstrap_servers }}"
RATE: "{{ inputs.throttle_mb_per_sec * 1048576 }}"
MAX_SECONDS: "{{ inputs.max_minutes * 60 }}"
ON_TIMEOUT: "{{ inputs.on_timeout }}"
commands:
- |
k=/opt/kafka/bin; B="--bootstrap-server $BOOTSTRAP"
fail() { echo "MOVE FAILED: $*" >&2; exit 1; }
emit() { echo; echo "::{\"outputs\":{\"state\":\"$1\",\"seconds\":$(( $(date +%s) - start )),\"done\":$2,\"total\":$total}}::"; }
total=$(grep -o '"partition"' plan.json | wc -l)
$k/kafka-reassign-partitions.sh $B --list > ongoing.txt 2>&1
! grep -q 'adding:' ongoing.txt || fail "another reassignment started while this one waited for approval: $(grep -m1 'adding:' ongoing.txt). Nothing was started"
start=$(date +%s)
$k/kafka-reassign-partitions.sh $B --execute --reassignment-json-file plan.json --throttle "$RATE" > execute.txt 2>&1 \
|| { cat execute.txt; fail "kafka-reassign-partitions --execute failed: $(grep -m1 -iE 'error|exception' execute.txt)"; }
grep -E 'throttle|Successfully' execute.txt || true
# --verify is what removes the throttle once every partition has moved. A move that is never
# verified keeps throttling replication on those brokers forever.
while :; do
$k/kafka-reassign-partitions.sh $B --verify --reassignment-json-file plan.json > verify.txt 2>&1 || true
done=$(grep -c 'is completed' verify.txt || true); waiting=$(grep -c 'still in progress' verify.txt || true)
if [ "$waiting" -eq 0 ] && [ "$done" -eq "$total" ]; then grep -i 'clearing' verify.txt || true; emit done "$done"; exit 0; fi
if [ $(( $(date +%s) - start )) -ge "$MAX_SECONDS" ]; then break; fi
echo "$(date -u +%H:%M:%S) $done/$total partitions moved"
sleep 10
done
if [ "$ON_TIMEOUT" = keep_running ]; then emit running "$done"; exit 0; fi
# --preserve-throttles: clearing them needs every broker to answer, and a dead destination broker is a
# common reason for a move to overrun. The finally block removes them once nothing is moving.
$k/kafka-reassign-partitions.sh $B --cancel --preserve-throttles --reassignment-json-file plan.json > cancel.txt 2>&1 || { cat cancel.txt; fail "--cancel failed after the time limit: $(grep -m1 -iE 'error|exception' cancel.txt)"; }
cat cancel.txt
# --cancel restores each partition's replica set but not always its order, which moves the preferred
# leader. Re-applying the saved assignment fixes the order without copying any data.
$k/kafka-reassign-partitions.sh $B --execute --reassignment-json-file rollback.json > restore.txt 2>&1 || { cat restore.txt; fail "re-applying the original replica order failed"; }
for i in $(seq 1 20); do $k/kafka-reassign-partitions.sh $B --verify --reassignment-json-file rollback.json > verify.txt 2>&1; grep -q 'still in progress' verify.txt || break; sleep 3; done
emit cancelled "$done"
- id: after_move
type: io.kestra.plugin.scripts.shell.Commands
description: Checks that every partition has the planned replicas (or the
original ones after a cancel) and elects the preferred leaders of the
moved partitions.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: host
pullPolicy: IF_NOT_PRESENT
user: root
containerImage: "{{ inputs.kafka_image }}"
inputFiles:
plan.json: "{{ outputs.plan.outputFiles['plan.json'] }}"
rollback.json: "{{ outputs.plan.outputFiles['rollback.json'] }}"
partitions.json: "{{ outputs.plan.outputFiles['partitions.json'] }}"
env:
BOOTSTRAP: "{{ inputs.bootstrap_servers }}"
STATE: "{{ outputs.move.vars.state }}"
TOPICS: "{{ render(vars.topics) }}"
commands:
- |
k=/opt/kafka/bin; B="--bootstrap-server $BOOTSTRAP"
want=$([ "$STATE" = cancelled ] && echo rollback.json || echo plan.json)
$k/kafka-topics.sh $B --describe --topic "^($(echo "$TOPICS" | tr ',' '|'))$" | awk '/Partition:/ {print $2 "-" $4 " " $8}' > actual.txt
wrong=$(grep -oE '"topic":"[^"]+","partition":[0-9]+,"replicas":\[[0-9, ]*\]' $want | sed -E 's/"topic":"([^"]+)","partition":([0-9]+),"replicas":\[([0-9, ]*)\]/\1-\2 \3/; s/ //2g' \
| while read -r tp r; do grep -qx "$tp $r" actual.txt || echo "$tp (want $r, has $(grep "^$tp " actual.txt | cut -d' ' -f2))"; done | paste -sd';' -)
[ "$STATE" = running ] && wrong=""
elected=""
if [ "$STATE" = done ]; then
$k/kafka-leader-election.sh $B --election-type PREFERRED --path-to-json-file partitions.json > elect.txt 2>&1 || true
elected=$(grep -c 'Successfully completed leader election' elect.txt; true)
fi
urp=$($k/kafka-topics.sh $B --describe --under-replicated-partitions | grep -c 'Partition:' || true)
echo "::{\"outputs\":{\"wrong\":\"$wrong\",\"under_replicated\":$urp,\"leader_elections\":\"$elected\"}}::"
- id: report
type: io.kestra.plugin.core.flow.Switch
value: "{{ outputs.move.vars.state }}"
cases:
done:
- id: check_result
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.after_move.vars.wrong == '' }}"
then:
- id: report_done
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Kafka reassignment done: {{ outputs.plan.vars.partitions }}
partition(s) of {{ render(vars.topics) }} now on brokers {{
render(vars.targets) }}, {{ outputs.plan.vars.total_mb }}
MiB in {{ outputs.move.vars.seconds }} s under a {{
inputs.throttle_mb_per_sec }} MiB/s throttle. Throttle
removed by --verify; preferred leaders elected; {{
outputs.after_move.vars.under_replicated }} under-replicated
partition(s) in the cluster."
else:
- id: wrong_assignment
type: io.kestra.plugin.core.execution.Fail
errorMessage: "MOVE FAILED: Kafka reported the move as complete, but these
partitions do not have the planned replicas: {{
outputs.after_move.vars.wrong }}"
cancelled:
- id: report_cancelled
type: io.kestra.plugin.core.execution.Fail
errorMessage: "MOVE FAILED: not finished within {{ inputs.max_minutes }} min ({{
outputs.move.vars.done }} of {{ outputs.move.vars.total }}
partitions had moved). Cancelled: {% if
outputs.after_move.vars.wrong == '' %}every partition is back on
its original replicas in the original order{% else %}these
partitions are not back as they were: {{
outputs.after_move.vars.wrong }}{% endif %}. Raise
throttle_mb_per_sec or max_minutes and run again."
running:
- id: report_running
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Kafka reassignment still running after {{ inputs.max_minutes }}
min ({{ outputs.move.vars.done }} of {{ outputs.move.vars.total
}} partitions moved). It continues under the {{
inputs.throttle_mb_per_sec }} MiB/s throttle, which stays until
someone runs `kafka-reassign-partitions --verify` with the plan
after it finishes. The plan is in execution {{ execution.id }}'s
outputs; the original assignment is in KV `{{ vars.rollback_key
}}`."
else:
- id: report_not_moved
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Kafka reassignment of {{ render(vars.topics) }} not started ({% if
outputs.approve_move.onResume.approve is defined %}declined{% else
%}no decision within 4 h{% endif %}). Nothing was changed."
finally:
- id: after_approved_move
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.approve_move.onResume.approve ?? false }}"
then:
- id: throttle_guard
type: io.kestra.plugin.scripts.shell.Commands
description: Runs after every approved move, also when a task failed or the
worker restarted. Once no reassignment is running, any replication
throttle left on a broker or topic is removed; while one is still
running, the throttle is left in place on purpose.
taskRunner:
type: io.kestra.plugin.scripts.runner.docker.Docker
networkMode: host
pullPolicy: IF_NOT_PRESENT
user: root
containerImage: "{{ inputs.kafka_image }}"
inputFiles:
brokers.txt: "{{ outputs.inspect.outputFiles['brokers.txt'] }}"
env:
BOOTSTRAP: "{{ inputs.bootstrap_servers }}"
TOPICS: "{{ render(vars.topics) }}"
commands:
- |
k=/opt/kafka/bin; B="--bootstrap-server $BOOTSTRAP"
running=$($k/kafka-reassign-partitions.sh $B --list 2>&1 | grep -c 'adding:' || true)
removed=""; failed=""
if [ "$running" -eq 0 ]; then
for b in $($k/kafka-configs.sh $B --describe --entity-type brokers | awk '/configs for broker/ {b = $5} /throttled.rate=/ {print b}' | sort -u); do
if $k/kafka-configs.sh $B --alter --entity-type brokers --entity-name "$b" --delete-config leader.replication.throttled.rate,follower.replication.throttled.rate > /dev/null 2>&1
then removed="$removed broker $b,"; else failed="$failed broker $b,"; fi
done
# A broker that is down is not listed by --describe, and its throttle cannot be removed until it is back.
now=$($k/kafka-broker-api-versions.sh $B 2>/dev/null | grep -oE '\(id: [0-9]+' | grep -oE '[0-9]+' || true)
for b in $(tr ',' ' ' < brokers.txt); do echo "$now" | grep -qx "$b" || failed="$failed broker $b (not answering),"; done
for t in $(echo "$TOPICS" | tr ',' ' '); do
if $k/kafka-configs.sh $B --describe --entity-type topics --entity-name "$t" | grep -q 'throttled.replicas='; then
$k/kafka-configs.sh $B --alter --entity-type topics --entity-name "$t" --delete-config leader.replication.throttled.replicas,follower.replication.throttled.replicas > /dev/null && removed="$removed topic $t,"
fi
done
fi
removed=${removed%,}; failed=${failed%,}; echo "::{\"outputs\":{\"running\":$running,\"removed\":\"${removed# }\",\"failed\":\"${failed# }\"}}::"
- id: guard_report
type: io.kestra.plugin.core.flow.If
condition: "{{ outputs.throttle_guard.vars.running > 0 or
outputs.throttle_guard.vars.removed != '' or
outputs.throttle_guard.vars.failed != '' }}"
then:
- id: report_guard
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Kafka reassignment, after execution {{ execution.id }}: {% if
outputs.throttle_guard.vars.running > 0 %}{{
outputs.throttle_guard.vars.running }} partition move(s) are still
running, so the replication throttle was left in place. Run
`kafka-reassign-partitions --verify` with the plan once they
finish; that removes it.{% else %}{% if
outputs.throttle_guard.vars.removed != '' %}replication throttle
removed on {{ outputs.throttle_guard.vars.removed }}; left in
place, it would keep limiting replication there.{% endif %}{% if
outputs.throttle_guard.vars.failed != '' %}{% if
outputs.throttle_guard.vars.removed != '' %} {% endif %}Could not
check or remove the throttle on {{
outputs.throttle_guard.vars.failed }}: once it is back, run
`kafka-configs --alter --entity-type brokers --entity-name <id>
--delete-config
leader.replication.throttled.rate,follower.replication.throttled.\
rate`.{% endif %}{% endif %}"
errors:
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: |
Kafka reassignment FAILED in execution {{ execution.id }}{% if errorLogs() | length > 0 %} at {{ errorLogs() | jq('([.[] | select(.message | test("MOVE FAILED"))] + .)[0] | "`" + .taskId + "`: " + (.message | sub("^.*MOVE FAILED: "; "") | split("\n")[0])') | first }}{% endif %}
{% if outputs.move is defined %}The original assignment is in KV `{{ vars.rollback_key }}`: `kafka-reassign-partitions --execute --reassignment-json-file <that JSON>` restores it.{% else %}No partition was moved.{% endif %}
outputs:
- id: plan
type: JSON
value: "{{ outputs.plan.vars ?? {} }}"
- id: result
type: JSON
value: "{{ {'move': outputs.move.vars ?? {}, 'after': outputs.after_move.vars ??
{}} }}"
