@@ -123,6 +123,7 @@ packages/core/platform/.build-revision
 examples/backups/*/.bucket-info.env
 examples/backups/*/.sentinel.env
 examples/backups/*/.backup-name.env
+examples/backups/*/.source-dump.txt
 packages/core/platform/images/migrations/etcd-operator-crds/
 
 # CI scratch: promote-rc.yaml checks the dispatch ref here so release tooling
 
@@ -0,0 +1,46 @@
+#!/bin/bash
+# Step 02: Map the Kafka application kind to the Job strategy from step 01. The
+# strategy parameters (which topics to back up, an optional replication-factor
+# override for the topics restore recreates) travel through the BackupClass and
+# are exposed to the strategy template as `.Parameters`. They are also
+# snapshotted onto the resulting Backup's driverMetadata, so a later restore
+# re-renders with the same values.
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+source "$SCRIPT_DIR/00-helpers.sh"
+
+print_header "Step 02: Create BackupClass '${BACKUPCLASS_NAME}'"
+log_command "kubectl apply -f - (BackupClass: $BACKUPCLASS_NAME -> $STRATEGY_NAME)"
+
+kubectl apply -f - <<EOF
+apiVersion: backups.cozystack.io/v1alpha1
+kind: BackupClass
+metadata:
+  name: ${BACKUPCLASS_NAME}
+spec:
+  strategies:
+    - application:
+        apiGroup: apps.cozystack.io
+        kind: Kafka
+      strategyRef:
+        apiGroup: strategy.backups.cozystack.io
+        kind: Job
+        name: ${STRATEGY_NAME}
+      parameters:
+        # Empty "topics" backs up every non-internal topic - the wider net, not
+        # the safer one: it can pull in a compacted or transactional topic,
+        # which this logical strategy fails closed on (see the drain guard in
+        # 01-create-strategy.sh and README "Scope and limitations"). Set a
+        # comma/space-separated list to restrict the backup to topics known to
+        # be plain.
+        topics: "${TOPIC}"
+        # Restore recreates each topic with the replication factor the backup
+        # captured from the source. Set replicationFactor to override it - the
+        # way to restore into a cluster with fewer brokers than the source.
+        # Like every parameter it is snapshotted at backup time, so it has to
+        # be in place before the BackupJob whose Backup will be restored.
+EOF
+
+log_success "BackupClass '${BACKUPCLASS_NAME}' created."
+echo -e "\n${GREEN}${BOLD}Next:${NC} ./03-create-bucket.sh"
@@ -0,0 +1,119 @@
+#!/bin/bash
+# Step 03: Provision an in-cluster Bucket and cache its S3 coordinates. The
+# generic Job strategy has no chart support to emit a backup Secret, so the
+# tenant materialises the "<app>-backup-s3" Secret itself. This step only
+# caches the coordinates; create_s3_secret (00-helpers.sh) turns them into
+# per-app Secrets in steps 04 and 07.
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+source "$SCRIPT_DIR/00-helpers.sh"
+
+print_header "Step 03: Provision Bucket '${BUCKET_NAME}' in '${NAMESPACE}'"
+
+log_command "kubectl apply -f - (Bucket: $BUCKET_NAME)"
+# spec.users.backup is required for the Bucket app to provision a
+# BucketAccess (and credentials Secret) the backup tooling can read.
+kubectl apply -f - <<EOF
+apiVersion: apps.cozystack.io/v1alpha1
+kind: Bucket
+metadata:
+  name: ${BUCKET_NAME}
+  namespace: ${NAMESPACE}
+spec:
+  users:
+    backup:
+      readonly: false
+EOF
+
+wait_hr_ready "bucket-${BUCKET_NAME}"
+kubectl -n "$NAMESPACE" wait bucketclaims.objectstorage.k8s.io "bucket-${BUCKET_NAME}" --for=jsonpath='{.status.bucketReady}'=true --timeout=300s
+# Cozystack's bucket app provisions a BucketAccess named "<bucket-name>-backup"
+# (the "-backup" suffix is the BucketAccessClass name); the BucketInfo Secret
+# carries the same name.
+kubectl -n "$NAMESPACE" wait bucketaccesses.objectstorage.k8s.io "bucket-${BUCKET_NAME}-backup" --for=jsonpath='{.status.accessGranted}'=true --timeout=300s
+
+# E2E only: accessGranted describes the COSI object, not whether the S3 server
+# has reloaded the new IAM identity, so a freshly granted key can still answer
+# the strategy Pod's SigV4 PUT with 403 - which reds the run as a backup defect
+# rather than the grant-propagation race it is. The shared preflight proves the
+# key works before anything depends on it; the other backup round-trips gate it
+# the same way.
+if [[ "${COZY_E2E_BACKUP_PREFLIGHT:-0}" == "1" ]]; then
+    # shellcheck source=/dev/null
+    source "$SCRIPT_DIR/../../../hack/e2e-chainsaw/_lib/backup-access-preflight.sh"
+    cozy_backup_access_preflight "$NAMESPACE" "bucket-${BUCKET_NAME}-backup" 90
+fi
+
+log_substep "Reading bucket coordinates from BucketInfo Secret..."
+TMP=$(mktemp)
+trap 'rm -f "$TMP"' EXIT
+kubectl -n "$NAMESPACE" get secret "bucket-${BUCKET_NAME}-backup" -o jsonpath='{.data.BucketInfo}' | base64 -d > "$TMP"
+
+S3_ACCESS_KEY=$(jq -r '.spec.secretS3.accessKeyID' "$TMP")
+S3_SECRET_KEY=$(jq -r '.spec.secretS3.accessSecretKey' "$TMP")
+# S3_ENDPOINT can be overridden via the environment: BucketInfo advertises the
+# external ingress URL, which in CI is an unroutable placeholder; the e2e
+# harness points this at the in-cluster seaweedfs Service instead.
+S3_ENDPOINT="${S3_ENDPOINT:-$(jq -r '.spec.secretS3.endpoint' "$TMP")}"
+S3_REGION=$(jq -r 'if (.spec.secretS3.region // "") == "" then "us-east-1" else .spec.secretS3.region end' "$TMP")
+S3_BUCKET=$(jq -r '.spec.bucketName' "$TMP")
+
+# `jq -r` returns the literal string "null" for missing JSON paths; fail fast
+# here so a missing field surfaces at extraction time instead of as a confusing
+# curl/SigV4 error inside the backup Pod.
+for v in S3_ACCESS_KEY S3_SECRET_KEY S3_ENDPOINT S3_REGION S3_BUCKET; do
+    [[ -n "${!v}" && "${!v}" != "null" ]] || { log_error "BucketInfo missing required field: ${v}"; exit 1; }
+done
+
+# The endpoint's CA, so the strategy Pod verifies TLS instead of skipping it.
+# The default name tracks the seaweedfs chart's fullnameOverride (seaweedfs ->
+# seaweedfs-ca-cert), but a downstream fullname change would rename it, so fall
+# back to discovering the cert-manager CA Certificate (the seaweedfs-labelled
+# one with spec.isCA=true) and read its secretName. S3_CA_SECRET="" skips the
+# copy for a publicly-trusted endpoint (see 00-helpers.sh).
+S3_CA_B64=""
+if [[ -n "$S3_CA_SECRET" ]]; then
+    # Only an absent Secret earns a discovery pass. The read crosses into
+    # another namespace, which a tenant persona is not granted, and a denial
+    # must say so rather than surface as "not found" and point at a name
+    # that cannot help.
+    if ! err=$(kubectl -n "$S3_CA_NAMESPACE" get secret "$S3_CA_SECRET" 2>&1 >/dev/null); then
+        case "$err" in
+            *NotFound*|*"not found"*) ;;
+            *)
+                log_error "Cannot read S3 CA secret ${S3_CA_NAMESPACE}/${S3_CA_SECRET}: ${err}"
+                log_error "Run this step with a persona that can read Secrets in ${S3_CA_NAMESPACE}, or set S3_CA_SECRET=\"\" for a publicly-trusted endpoint."
+                exit 1 ;;
+        esac
+        log_warning "S3 CA secret ${S3_CA_NAMESPACE}/${S3_CA_SECRET} not found; discovering the seaweedfs CA Certificate..."
+        discovered=$(kubectl -n "$S3_CA_NAMESPACE" get certificates.cert-manager.io \
+            -l app.kubernetes.io/name=seaweedfs \
+            -o jsonpath='{range .items[*]}{.spec.isCA}{" "}{.spec.secretName}{"\n"}{end}' \
+            | awk '$1=="true"{print $2; exit}' || true)
+        [[ -n "$discovered" ]] || { log_error "No seaweedfs CA Certificate found in ${S3_CA_NAMESPACE} (or the read was denied, see above); set S3_CA_SECRET explicitly (or empty for a publicly-trusted endpoint)."; exit 1; }
+        log_success "Discovered seaweedfs CA secret ${S3_CA_NAMESPACE}/${discovered}"
+        S3_CA_SECRET="$discovered"
+    fi
+    log_substep "Caching S3 CA ${S3_CA_NAMESPACE}/${S3_CA_SECRET}[${S3_CA_KEY}]..."
+    S3_CA_B64=$(kubectl -n "$S3_CA_NAMESPACE" get secret "$S3_CA_SECRET" -o jsonpath="{.data.${S3_CA_KEY//./\\.}}")
+    [[ -n "$S3_CA_B64" ]] || { log_error "S3 CA secret ${S3_CA_NAMESPACE}/${S3_CA_SECRET} has no ${S3_CA_KEY}"; exit 1; }
+fi
+
+# Persist for steps 04 / 07. The cache stores raw S3 credentials, so apply
+# restrictive perms before writing the body - umask alone could leave the file
+# group/world-readable. The CA stays base64 so the file remains one export per
+# line.
+umask 077
+cat > "$SCRIPT_DIR/.bucket-info.env" <<ENV
+export S3_ACCESS_KEY=${S3_ACCESS_KEY}
+export S3_SECRET_KEY=${S3_SECRET_KEY}
+export S3_ENDPOINT=${S3_ENDPOINT}
+export S3_REGION=${S3_REGION}
+export S3_BUCKET=${S3_BUCKET}
+export S3_CA_B64=${S3_CA_B64}
+ENV
+chmod 600 "$SCRIPT_DIR/.bucket-info.env"
+
+log_success "Bucket '${BUCKET_NAME}' ready; coordinates cached in $(basename "$SCRIPT_DIR")/.bucket-info.env."
+echo -e "\n${GREEN}${BOLD}Next:${NC} ./04-create-kafka.sh"
@@ -0,0 +1,64 @@
+#!/bin/bash
+# Step 04: Provision a Kafka instance with KAFKA_REPLICAS brokers, create the
+# S3 Secret the Job strategy consumes, then seed a topic with sentinel messages
+# used to verify the backup/restore round-trip. The topic is created via the
+# CLI (not the chart's `topics:`) so there is no KafkaTopic CR - the in-place
+# restore in step 06 can then delete and let the driver recreate the topic
+# deterministically, with no Topic Operator reconciliation racing the restore.
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+source "$SCRIPT_DIR/00-helpers.sh"
+
+print_header "Step 04: Provision Kafka '${KAFKA_NAME}'"
+
+kubectl apply -f - <<EOF
+apiVersion: apps.cozystack.io/v1alpha1
+kind: Kafka
+metadata:
+  name: ${KAFKA_NAME}
+  namespace: ${NAMESPACE}
+spec:
+  external: false
+  kafka:
+    replicas: ${KAFKA_REPLICAS}
+    size: 2Gi
+    resourcesPreset: "c1.small"
+  topics: []
+EOF
+
+wait_hr_ready "kafka-${KAFKA_NAME}"
+
+log_substep "Waiting for the Strimzi Kafka cluster to be Ready (broker and controller pools up)..."
+kubectl -n "$NAMESPACE" wait kafka.kafka.strimzi.io "kafka-${KAFKA_NAME}" \
+    --for=condition=Ready --timeout=600s
+
+log_substep "Creating S3 credentials Secret '${KAFKA_NAME}-backup-s3'..."
+create_s3_secret "$KAFKA_NAME"
+
+log_substep "Creating topic '${TOPIC}' (${PARTITIONS} partitions, ${TOPIC_REPLICAS} replicas) and publishing ${MESSAGE_COUNT} sentinel messages..."
+seed_topic "$KAFKA_NAME"
+
+count=$(topic_message_count "$KAFKA_NAME")
+[[ "$count" == "$MESSAGE_COUNT" ]] || { log_error "expected ${MESSAGE_COUNT} messages in '${TOPIC}', got '${count}'"; exit 1; }
+rf=$(topic_replication_factor "$KAFKA_NAME")
+[[ "$rf" == "$TOPIC_REPLICAS" ]] || { log_error "expected '${TOPIC}' to have ${TOPIC_REPLICAS} replicas, got '${rf}'"; exit 1; }
+log_success "Topic '${TOPIC}' holds ${count} record(s) at replication factor ${rf}."
+
+# Seed the decoy. It is never named in the BackupClass, so a correct run must
+# leave it untouched; a --topic call that matched it as a regex instead would
+# either pull its partitions into the backup or delete it in step 06.
+log_substep "Creating decoy topic '${DECOY_TOPIC}' (${DECOY_COUNT} records) to pin literal topic matching..."
+seed_decoy_topic "$KAFKA_NAME"
+decoy=$(decoy_message_count "$KAFKA_NAME")
+[[ "$decoy" == "$DECOY_COUNT" ]] || { log_error "expected ${DECOY_COUNT} messages in '${DECOY_TOPIC}', got '${decoy}'"; exit 1; }
+log_success "Decoy topic '${DECOY_TOPIC}' holds ${decoy} record(s)."
+
+# Snapshot the source content so the restore steps can diff against it and
+# prove the records - not just their count - round-tripped. cleanup.sh removes
+# this file.
+topic_dump "$KAFKA_NAME" > "$SCRIPT_DIR/.source-dump.txt"
+dumped=$(wc -l < "$SCRIPT_DIR/.source-dump.txt" | tr -d '[:space:]')
+[[ "$dumped" == "$MESSAGE_COUNT" ]] || { log_error "source dump has ${dumped} records, expected ${MESSAGE_COUNT}"; exit 1; }
+
+echo -e "\n${GREEN}${BOLD}Next:${NC} ./05-create-backupjob.sh"
@@ -0,0 +1,43 @@
+#!/bin/bash
+# Step 05: Create a BackupJob and wait for it to succeed. The controller
+# resolves the BackupClass -> Job strategy, renders the Pod template against
+# the Kafka app, and runs it as a batch/v1.Job (owned by the BackupJob). The
+# Pod freezes each partition's end offset, drains [begin,end) and uploads a
+# tarball to S3.
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+source "$SCRIPT_DIR/00-helpers.sh"
+
+print_header "Step 05: Create BackupJob '${BACKUPJOB_NAME}'"
+
+kubectl apply -f - <<EOF
+apiVersion: backups.cozystack.io/v1alpha1
+kind: BackupJob
+metadata:
+  name: ${BACKUPJOB_NAME}
+  namespace: ${NAMESPACE}
+spec:
+  applicationRef:
+    apiGroup: apps.cozystack.io
+    kind: Kafka
+    name: ${KAFKA_NAME}
+  backupClassName: ${BACKUPCLASS_NAME}
+EOF
+
+log_substep "Waiting for BackupJob to Succeed (this drains the topic to S3)..."
+# Skip an explicit "wait Running" gate: a fast-completing strategy Pod can
+# transition Pending -> Succeeded between two API observations, in which case
+# `kubectl wait --for=jsonpath='{.status.phase}'=Running` would time out even
+# though the BackupJob actually finished. wait_for_field already polls until
+# the terminal phase, which covers both the slow and the fast path.
+wait_for_field backupjob "$BACKUPJOB_NAME" '{.status.phase}' Succeeded "$NAMESPACE" 600 Failed
+
+backup_ref=$(kubectl -n "$NAMESPACE" get backupjob "$BACKUPJOB_NAME" -o jsonpath='{.status.backupRef.name}')
+[[ -n "$backup_ref" ]] || { log_error "BackupJob succeeded but BackupRef is empty"; exit 1; }
+# The restore steps reference the Backup by this resolved name rather than by
+# assuming it equals the BackupJob's. cleanup.sh removes the file.
+printf 'export BACKUP_NAME=%q\n' "$backup_ref" > "$SCRIPT_DIR/.backup-name.env"
+log_success "Backup '${backup_ref}' is Ready."
+
+echo -e "\n${GREEN}${BOLD}Next:${NC} ./06-restore-in-place.sh"