Kafka Consumer Group Offset Checkpoint and Rewind with Kestra

id: kafka-consumer-group-offset-checkpoint-rewind
namespace: company.team

description: |
  Checkpoint a Kafka consumer group's committed offsets to the KV store on a schedule, and
  rewind the group to the latest checkpoint taken before a given time, to replay what a bad
  consumer release processed. The rewind only runs on a group with no active members, shows
  the per-partition plan, waits for approval, alters the offsets with ConsumerGroupAlterOffsets,
  and describes the group again to verify every partition landed on the checkpoint.

inputs:
  - id: mode
    type: SELECT
    displayName: Mode
    values:
      - CHECKPOINT
      - REWIND
    defaults: CHECKPOINT

  - id: bootstrap_servers
    type: STRING
    displayName: Kafka bootstrap servers
    defaults: "localhost:9092"

  - id: group_id
    type: STRING
    displayName: Consumer group
    defaults: "shipment-sync"

  - id: rewind_before
    type: DATETIME
    displayName: Rewind to the last checkpoint before
    description: Used in REWIND mode. Usually the time the bad release went out.
    required: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: In REWIND mode, only log the plan. Must be false for the offsets to change.
    defaults: true

  - id: keep_checkpoints
    type: INT
    displayName: Checkpoints kept
    defaults: 168

variables:
  kv_key: "kafka_offsets_{{ inputs.group_id | slugify }}"

tasks:
  - id: describe
    type: io.kestra.plugin.kafka.ConsumerGroupDescribe
    description: Read-only. Committed offset, end offset, lag and active members per
      partition.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"
    groupIds:
      - "{{ inputs.group_id }}"

  - id: group
    type: io.kestra.plugin.core.output.OutputValues
    values:
      state: "{{ outputs.describe.groups | first | jq('.state') | first }}"
      members: "{{ outputs.describe.groups | first | jq('.members | length') | first }}"
      offsets: "{{ outputs.describe.groups | first | jq('[.offsets[] | {topic,
        partition, offset: .currentOffset, end: .endOffset}] | sort_by(.topic,
        .partition)') | first | toJson }}"

  - id: by_mode
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ inputs.mode }}"
    cases:
      CHECKPOINT:
        - id: has_offsets
          type: io.kestra.plugin.core.flow.If
          condition: "{{ fromJson(outputs.group.values.offsets) | length > 0 }}"
          then:
            - id: save_checkpoint
              type: io.kestra.plugin.core.kv.Set
              description: Append this checkpoint and keep the newest keep_checkpoints
                entries.
              key: "{{ render(vars.kv_key) }}"
              kvType: JSON
              value: >-
                {{ {'old': (kv(render(vars.kv_key), errorOnMissing=false) ??
                []), 'cp': {'at': execution.startDate, 'offsets':
                fromJson(outputs.group.values.offsets)}}
                   | jq('(.old + [.cp]) | sort_by(.at) | .[-' ~ inputs.keep_checkpoints ~ ':]') | first | toJson }}
            - id: log_checkpoint
              type: io.kestra.plugin.core.log.Log
              message: "Checkpoint of {{ inputs.group_id }} at {{ execution.startDate }}: {{
                fromJson(outputs.group.values.offsets) |
                jq('map(\"\\(.topic)-\\(.partition)@\\(.offset)\") | join(\",
                \")') | first }}."
          else:
            - id: no_offsets
              type: io.kestra.plugin.core.execution.Fail
              errorMessage: "Consumer group {{ inputs.group_id }} has no committed offsets:
                nothing to checkpoint. Check the group name."

      REWIND:
        - id: need_time
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.rewind_before is null }}"
          then:
            - id: missing_time
              type: io.kestra.plugin.core.execution.Fail
              errorMessage: "REWIND needs rewind_before."

        - id: plan
          type: io.kestra.plugin.core.output.OutputValues
          description: The newest checkpoint strictly before rewind_before, and per
            partition the current and target offsets.
          values:
            checkpoint: >-
              {{ (kv(render(vars.kv_key), errorOnMissing=false) ?? []) |
              jq('[.[] | select(.at < "' ~ (inputs.rewind_before |
              date("yyyy-MM-dd'T'HH:mm:ss.SSSSSS'Z'", timeZone="UTC")) ~ '")] |
              last // {}') | first | toJson }}

        - id: plan_items
          type: io.kestra.plugin.core.output.OutputValues
          values:
            items: >-
              {{ {'cp': fromJson(outputs.plan.values.checkpoint), 'now':
              fromJson(outputs.group.values.offsets)} | jq('[(.now |
              map({("\(.topic)-\(.partition)"): .}) | add // {}) as $n
                 | (.cp.offsets // [])[] | {topic, partition, current: ($n["\(.topic)-\(.partition)"].offset // null), target: .offset} | . + {replay: (if .current == null then null else .current - .target end)}]') | first | toJson }}

        - id: log_plan
          type: io.kestra.plugin.core.log.Log
          message: >-
            Rewind plan for {{ inputs.group_id }} (state {{
            outputs.group.values.state }}, {{ outputs.group.values.members }}
            members): checkpoint {{ fromJson(outputs.plan.values.checkpoint).at
            ?? 'none' }}. {{ fromJson(outputs.plan_items.values.items) |
            jq('map("\(.topic)-\(.partition): \(.current) -> \(.target)
            (\(.replay) to replay)") | join(", ")') | first }}.

        - id: checks
          type: io.kestra.plugin.core.flow.If
          description: Refuse instead of guessing.
          condition: >-
            {{ fromJson(outputs.plan.values.checkpoint).at is not defined
               or outputs.group.values.members | number > 0
               or fromJson(outputs.plan_items.values.items) | jq('[.[] | select(.replay == null or .replay < 0)] | length') | first > 0 }}
          then:
            - id: refused
              type: io.kestra.plugin.core.execution.Fail
              errorMessage: >-
                Rewind refused: {{ fromJson(outputs.plan.values.checkpoint).at
                is not defined ? 'no checkpoint before ' ~ inputs.rewind_before
                ~ '. ' : '' }}{{ outputs.group.values.members | number > 0 ?
                'the group has ' ~ outputs.group.values.members ~ ' active
                members, stop the consumers first. ' : '' }}{{
                fromJson(outputs.plan_items.values.items) | jq('[.[] |
                select(.replay == null or .replay < 0)] | if length > 0 then
                "the checkpoint is ahead of, or unknown to, the group on " +
                (map("\(.topic)-\(.partition)") | join(", ")) + "." else ""
                end') | first }}

        - id: apply_gate
          type: io.kestra.plugin.core.flow.If
          condition: "{{ not inputs.dry_run }}"
          then:
            - id: approve
              type: io.kestra.plugin.core.flow.Pause
              description: Resume the execution to apply the plan logged by log_plan. Kill it
                to cancel.

            - id: alter
              type: io.kestra.plugin.kafka.ConsumerGroupAlterOffsets
              properties:
                bootstrap.servers: "{{ inputs.bootstrap_servers }}"
              groupId: "{{ inputs.group_id }}"
              offsets: "{{ fromJson(outputs.plan_items.values.items) | jq('map({topic,
                partition, offset: .target})') | first }}"

            - id: verify
              type: io.kestra.plugin.kafka.ConsumerGroupDescribe
              properties:
                bootstrap.servers: "{{ inputs.bootstrap_servers }}"
              groupIds:
                - "{{ inputs.group_id }}"

            - id: verify_result
              type: io.kestra.plugin.core.flow.If
              condition: >-
                {{ {'want': fromJson(outputs.plan_items.values.items), 'got':
                (outputs.verify.groups | first).offsets} | jq('(.got |
                map({("\(.topic)-\(.partition)"): .currentOffset}) | add) as $g
                | [.want[] | select($g["\(.topic)-\(.partition)"] != .target)] |
                length') | first > 0 }}
              then:
                - id: rewind_failed
                  type: io.kestra.plugin.core.execution.Fail
                  errorMessage: "ConsumerGroupAlterOffsets ran, but the re-describe shows offsets
                    that differ from the checkpoint."
              else:
                - id: rewound
                  type: io.kestra.plugin.core.log.Log
                  message: >-
                    {{ inputs.group_id }} rewound to the checkpoint of {{
                    fromJson(outputs.plan.values.checkpoint).at }}: {{
                    fromJson(outputs.plan_items.values.items) | jq('map(.replay)
                    | add') | first }} messages will be replayed when the
                    consumers restart.
          else:
            - id: dry_run_only
              type: io.kestra.plugin.core.log.Log
              message: "Dry run: offsets unchanged. Run again with dry_run false to apply this
                plan after approval."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kafka offset ' ~ (inputs.mode | lower) ~ ' for group ' ~ inputs.group_id ~ ' failed or was refused (execution ' ~ execution.id ~ '). Offsets were not changed unless the log says rewound.'} | toJson }}

triggers:
  - id: hourly_checkpoint
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled. Takes a checkpoint every hour; a rewind is always
      started by hand.
    cron: "0 * * * *"
    disabled: true
    inputs:
      mode: CHECKPOINT

A consumer release goes out with a bug. For two hours it reads messages, commits their offsets and writes garbage downstream. Once the fix is deployed, those messages need processing again. That means moving the group's committed offsets back to where they were before the release. Doing it by hand under pressure with kafka-consumer-groups.sh --reset-offsets is easy to get wrong:

picking the wrong group or the wrong time;
resetting while the consumers are still running;
not checking that the reset actually landed.

This blueprint takes an hourly checkpoint of a group's committed offsets in the KV store. On demand, it rewinds the group to the newest checkpoint taken before a given time. The rewind shows the per-partition plan, refuses unsafe cases, waits for approval and alters the offsets. It then describes the group again to prove every partition is on the checkpoint.

This blueprint was created by zkasuran.

How it works

describe (io.kestra.plugin.kafka.ConsumerGroupDescribe, read-only) returns the group state, active members, and committed and end offsets per partition.
CHECKPOINT mode:
- save_checkpoint (io.kestra.plugin.core.kv.Set) appends the offsets with a timestamp to the KV key kafka_offsets_<group>, and keeps the newest keep_checkpoints entries.
- A group with no committed offsets fails the run instead of saving an empty checkpoint.
REWIND mode:
- plan picks the newest checkpoint strictly before rewind_before. For each partition it computes the current offset, the target and the messages that will be replayed. log_plan prints the plan.
- checks refuses when there is no checkpoint before that time, or when the group still has active members (Kafka cannot move a live group's offsets, and the consumers would overwrite them). It also refuses when a partition's checkpoint is ahead of the current offset or unknown to the group.
- With dry_run: true (the default) nothing else happens.
- With dry_run: false:
  - approve (io.kestra.plugin.core.flow.Pause) waits for someone to resume the execution.
  - alter (io.kestra.plugin.kafka.ConsumerGroupAlterOffsets) sets the checkpoint offsets.
  - verify describes the group again, and verify_result fails if any partition differs from the checkpoint.
errors alerts Slack when a run fails or is refused.

What you get

An offset history per consumer group, taken on a schedule and kept in the KV store, so there is always a known-good position to go back to.
A rewind plan per partition (current offset, target, messages to replay) before anything changes.
Refusals instead of guesses: no checkpoint before the time, a group with live members, or a checkpoint ahead of the group.
An approval pause, then a re-describe that proves every partition landed on the checkpoint.

Who it's for

Platform and data engineering teams running Kafka consumers whose releases sometimes have to be replayed.
On-call engineers who need to undo a bad consumer release without typing --reset-offsets by hand during an incident.
Teams evaluating the Kafka plugin who want a worked example of ConsumerGroupDescribe and ConsumerGroupAlterOffsets with gates around them.

Why orchestrate this with Kestra

kafka-consumer-groups.sh --reset-offsets changes offsets, but it does not remember where the group was an hour ago, check that the consumers are stopped, wait for a second person, or prove the result. Kestra adds the schedule that builds the checkpoint history, the KV store that keeps it, the Pause that turns a rewind into an approved change, and an execution history of every rewind with its plan.

Tested end to end

On Kestra 2.0.5 OSS against Apache Kafka 3.9.1 (KRaft), with a local HTTP sink standing in for Slack. The demo topic shipments has 3 partitions, and the group shipment-sync had read all 93 messages.

Run	Inputs	Result
1	`CHECKPOINT`	Checkpoint saved: partitions 0, 1 and 2 at offset 31
2	(12 new messages consumed) `CHECKPOINT`	Second checkpoint: partition 1 at 43
3	`REWIND`, before the second checkpoint, dry run	Plan: partition 1 from 43 to 31, 12 to replay. Offsets unchanged
4	`REWIND`, before any checkpoint	Refused, Slack alert
5	`REWIND` with a console consumer still running	Refused: 1 active member, Slack alert
6	`REWIND`, `dry_run: false`	Paused for approval. After resume, offsets altered and verified. `kafka-consumer-groups.sh` shows partition 1 at 31 with lag 12, and restarting the consumer replays exactly the 12 messages

Prerequisites

A Kafka cluster reachable from the Kestra worker, and a principal allowed to describe the group and alter its offsets.
Stop the group's consumers before a rewind. The flow refuses to run while members are active.
Local testing: docker run -d --name kafka -p 9092:9092 apache/kafka:3.9.1, then consume a topic with kafka-console-consumer.sh --group <name> to create a group.

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook used by the errors block. In Kestra OSS, provide it as the base64-encoded environment variable SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`mode`	`CHECKPOINT`	`CHECKPOINT` or `REWIND`
`bootstrap_servers`	`localhost:9092`	Brokers
`group_id`	`shipment-sync`	Consumer group
`rewind_before`	none	Rewind target time
`dry_run`	`true`	Must be `false` for offsets to change
`keep_checkpoints`	`168`	One week of hourly checkpoints

The hourly_checkpoint trigger is shipped disabled. A rewind is always started by hand.

Quick start

Add the SLACK_WEBHOOK_URL secret.
Import the flow and set bootstrap_servers and group_id.
Run it in CHECKPOINT mode, then enable the hourly_checkpoint trigger.
When a release has to be replayed, stop its consumers and run REWIND with rewind_before set to the release time, first with dry_run: true to read the plan.
Run it again with dry_run: false, resume the paused execution to approve, and restart the consumers.

How to extend

Checkpoint several groups by making group_id a list and looping over it.
Stop and start the consumers from the flow, for example by scaling a Kubernetes deployment to zero before alter and back after verify.
Add a rewind_to_offsets mode that takes explicit offsets, for a replay that does not match a checkpoint.
Send the plan to the approver in Slack with a link to the paused execution.

Kafka Consumer Group Offset Checkpoint and Rewind

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 Consumer Group Offset Checkpoint and Rewind

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 DLQ Quarantine, Classification, and Safe Replay Gate

Kafka ACL Overly Broad Grant Audit

Self-heal Kafka Connect connectors with a restart budget and escalation

Watch a Kafka Consumer Group's Backlog and Alert Before It Becomes an Outage

Real-Time Kafka Record Processing with Per-Record Executions

Kafka Consumer Lag Sentinel

More Infrastructure Blueprints

Block a release when k6 shows it is slower than the last good one

Kubernetes AI-powered incident monitoring, analysis, and alerting

Clusterify

Self-heal Kafka Connect connectors with a restart budget and escalation

Kafka Client Quota Guard

Kafka ACL Overly Broad Grant Audit

New to Kestra?