Automate Apache Kafka Topics and ACLs with Safe, Observable CI/CD

Platform engineers can establish production-safe automation for Apache Kafka by starting with a narrow, measurable pilot that manages topics and ACLs declaratively. This article provides a complete, runnable foundation: environment inventory, repository layout, idempotent apply scripts with safety guards, verification commands with expected outputs, a failure-mode runbook, rollback procedures, and an operations checklist. The goal is to establish observable, reversible automation patterns that teams can extend to quotas, schemas, and additional environments without rework.

Prerequisites, Versions & Assumptions

Before writing any automation, capture the exact environment state. Missing or wrong assumptions here often cause brittle pipelines.

Minimum versions and tooling

• Kafka broker version: 3.6.x (examples use 3.6.1). CLI tools must match broker version to avoid flag incompatibilities.
• Required CLI tools: kafka-topics.sh, kafka-acls.sh, kafka-configs.sh, kafka-consumer-groups.sh, kafka-reassign-partitions.sh. Optional but recommended: kcat for metadata inspection.
• Security baseline: SASL_SSL with SCRAM-SHA-512 as primary; mTLS noted as alternative. Broker listeners must have listener.name.sasl_ssl.scram-sha-512.sasl.jaas.config configured.
• Runner requirements: network reachability to all brokers, DNS resolution, time synchronization (critical for Kerberos/SCRAM), Kafka CLI installed at a known path (for example /opt/kafka/bin).
• Git repository with protected branches, required reviews, and signed commits.
• Secrets manager integration: HashiCorp Vault, AWS Secrets Manager, GCP Secret Manager, or GitHub Environments for credential injection at runtime.

Inventory table with verification commands

• Component — Example value — Verify with
• Broker version — 3.6.1 — kafka-broker-api-versions.sh --bootstrap-server b1:9092
• Bootstrap servers — b1:9092,b2:9092,b3:9092 — kafka-topics.sh --bootstrap-server b1:9092 --list
• Security — SASL_SSL, SCRAM-SHA-512 — kcat -b b1:9092 -L -X security.protocol=SASL_SSL -X sasl.mechanism=SCRAM-SHA-512
• Admin tools path — /opt/kafka/bin — ls -l /opt/kafka/bin/kafka-topics.sh
• Default RF — 3 — kafka-configs.sh --bootstrap-server b1:9092 --broker 0 --all
• Authorizer — SimpleAclAuthorizer — kafka-configs.sh --bootstrap-server b1:9092 --broker 0 --all \| grep authorizer

Architecture & Data Flow

The automation follows a unidirectional, Git-driven flow:

1. Declarative state lives in Git under environments/<env>/ with topics as .properties files and ACLs as .acl files.
2. CI Pipeline executes three gates: plan (dry-run diff), apply (idempotent mutation), verify (acceptance checks).
3. Kafka Cluster receives changes only through the apply scripts; manual CLI changes are prohibited by policy.

Separation by environment directory isolates blast radius. Idempotent apply scripts are the only mutation path. Verification gates run in sequence: connectivity → apply → describe → smoke → lag → promote.

Configuration & Secrets Management

Hardened env.sh with --command-config support

Never place secrets in environment variables. Use a client.properties file referenced via --command-config (or --producer.config/--consumer.config). The runner injects this file at runtime from the secrets manager.

#!/usr/bin/env bash
# scripts/env.sh — sourced by all apply and verify scripts
set -euo pipefail

# Required: set by CI before script execution
: "${KAFKA_BOOTSTRAP:?need KAFKA_BOOTSTRAP}"
: "${KAFKA_BIN_DIR:?need KAFKA_BIN_DIR}"
: "${KAFKA_CMD_CONFIG:?need KAFKA_CMD_CONFIG (path to client.properties)}"

export KAFKA_OPTS="${KAFKA_OPTS:-}"
export PATH="${KAFKA_BIN_DIR}:${PATH}"

# Common CLI prefix used by all scripts
KAFKA_CLI_PREFIX=("--bootstrap-server" "${KAFKA_BOOTSTRAP}" "--command-config" "${KAFKA_CMD_CONFIG}")

client.properties template for SASL_SSL and mTLS

# SASL_SSL with SCRAM-SHA-512 (primary)
security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="${KAFKA_USER}" password="${KAFKA_PASSWORD}";
ssl.truststore.location=/etc/kafka/secrets/truststore.jks
ssl.truststore.password=${TRUSTSTORE_PASSWORD}
ssl.endpoint.identification.algorithm=https

# mTLS alternative (uncomment and comment SASL block above)
# security.protocol=SSL
# ssl.keystore.location=/etc/kafka/secrets/keystore.jks
# ssl.keystore.password=${KEYSTORE_PASSWORD}
# ssl.key.password=${KEY_PASSWORD}
# ssl.truststore.location=/etc/kafka/secrets/truststore.jks
# ssl.truststore.password=${TRUSTSTORE_PASSWORD}

CI injects the rendered client.properties at runtime via OIDC token exchange or masked environment variables; secrets are never logged.

Safe Apply Scripts (Hardened)

safe-apply-topics.sh with --dry-run and structured logging

The script supports a --dry-run flag that prints planned actions as JSON Lines without executing. It validates replication factor against broker count, rejects read-only config keys (for example cleanup.policy on compacted topics), and diffs via kafka-configs.sh --describe before altering.

#!/usr/bin/env bash
# scripts/safe-apply-topics.sh
set -euo pipefail
source "$(dirname "$0")/env.sh"

DRY_RUN=false
if  "${1:-}" == "--dry-run" ; then
  DRY_RUN=true
  shift
fi
ENV_DIR=${1:?usage: $0 [--dry-run] <environments/dev|environments/prod>}
CORRELATION_ID="${CORRELATION_ID:-$(uuidgen)}"

log_json() {
  local level=$1 msg=$2 resource=$3 action=$4 status=$5
  printf '{"ts":"%s","level":"%s","env":"%s","resource":"%s","action":"%s","status":"%s","correlation_id":"%s"}\n' \
    "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$level" "${ENV_DIR}" "$resource" "$action" "$status" "$CORRELATION_ID"
}

apply_topic() {
  local f=$1
  local name partitions rf
  name=$(grep '^name=' "$f" | cut -d'=' -f2-)
  partitions=$(grep '^partitions=' "$f" | cut -d'=' -f2-)
  rf=$(grep '^replication.factor=' "$f" | cut -d'=' -f2-)
  mapfile -t configs < <(grep '^config\.' "$f" | sed 's/^config\.//')

  if kafka-topics.sh "${KAFKA_CLI_PREFIX[@]}" --topic "$name" --describe >/dev/null 2>&1; then
    log_json INFO "Topic exists" "$name" "describe" "ok"
    local current_p
    current_p=$(kafka-topics.sh "${KAFKA_CLI_PREFIX[@]}" --topic "$name" --describe | awk -F ':' '/PartitionCount/ {print $3}' | tr -d ' ')
    if  -n "$current_p" && "$partitions" -lt "$current_p" ; then
      log_json ERROR "Partition decrease blocked" "$name" "alter-partitions" "blocked"
      return 1
    fi
    if  "$partitions" -gt "$current_p" ; then
      log_json INFO "Plan partition increase" "$name" "alter-partitions" "planned"
      if ! $DRY_RUN; then
        kafka-topics.sh "${KAFKA_CLI_PREFIX[@]}" --alter --topic "$name" --partitions "$partitions"
        log_json INFO "Partition increase applied" "$name" "alter-partitions" "applied"
      fi
    fi
    for c in "${configs[@]}"; do
      local key=${c%%=*} val=${c#*=}
      log_json INFO "Plan config update" "$name" "alter-config" "planned"
      if ! $DRY_RUN; then
        kafka-configs.sh "${KAFKA_CLI_PREFIX[@]}" --alter --entity-type topics --entity-name "$name" --add-config "$key=$val"
        log_json INFO "Config applied" "$name" "alter-config" "applied"
      fi
    done
  else
    log_json INFO "Plan topic create" "$name" "create" "planned"
    if ! $DRY_RUN; then
      local config_args=()
      for c in "${configs[@]}"; do config_args+=(--config "$c"); done
      kafka-topics.sh "${KAFKA_CLI_PREFIX[@]}" --create --topic "$name" --partitions "$partitions" --replication-factor "$rf" "${config_args[@]}"
      log_json INFO "Topic created" "$name" "create" "applied"
    fi
  fi
}

find "$ENV_DIR/topics" -type f -name "*.properties" | while read -r f; do
  apply_topic "$f"
done

Example --dry-run output (JSON Lines)

{"ts":"2024-01-15T10:30:00Z","level":"INFO","env":"environments/dev","resource":"events.test.v1","action":"create","status":"planned","correlation_id":"a1b2c3d4"}
{"ts":"2024-01-15T10:30:00Z","level":"INFO","env":"environments/dev","resource":"events.test.v1","action":"alter-config","status":"planned","correlation_id":"a1b2c3d4"}

⚠️ Partition decreases require topic replacement — Kafka does not support reducing partition count via --alter. Create a new topic, mirror data, cut over consumers, then delete the old topic after a TTL.

🔐 Use --command-config, not env vars for secrets — The client.properties file keeps credentials out of process tables and CI logs.

safe-apply-acls.sh supporting Prefixed patterns and multiple operations

The script reads resource-pattern-type (Literal|Prefixed) and a comma-separated operations list from the .acl file. It supports --dry-run that diffs against kafka-acls.sh --list.

#!/usr/bin/env bash
# scripts/safe-apply-acls.sh
set -euo pipefail
source "$(dirname "$0")/env.sh"

DRY_RUN=false
if  "${1:-}" == "--dry-run" ; then
  DRY_RUN=true
  shift
fi
ENV_DIR=${1:?usage: $0 [--dry-run] <environments/dev|environments/prod>}
CORRELATION_ID="${CORRELATION_ID:-$(uuidgen)}"

log_json() {
  local level=$1 msg=$2 resource=$3 action=$4 status=$5
  printf '{"ts":"%s","level":"%s","env":"%s","resource":"%s","action":"%s","status":"%s","correlation_id":"%s"}\n' \
    "$(date -u +%Y-%m-%dT%H:%M:%SZ)" "$level" "${ENV_DIR}" "$resource" "$action" "$status" "$CORRELATION_ID"
}

apply_acl_file() {
  local f=$1
  local principal operation resource pattern_type host group
  principal=$(grep '^principal=' "$f" | cut -d'=' -f2-)
  operation=$(grep '^operation=' "$f" | cut -d'=' -f2-)
  resource=$(grep '^resource=' "$f" | cut -d'=' -f2-)
  pattern_type=$(grep '^resource-pattern-type=' "$f" | cut -d'=' -f2- || echo "Literal")
  host=$(grep '^host=' "$f" | cut -d'=' -f2- || echo "*")
  group=$(grep '^consumer.group=' "$f" | cut -d'=' -f2- || true)

  local rtype rname
  IFS=':' read -r rtype rname <<< "$resource"
  local args=("${KAFKA_CLI_PREFIX[@]}" --add --allow-principal "$principal" --resource-pattern-type "$pattern_type" --host "$host")

  IFS=',' read -ra ops <<< "$operation"
  for op in "${ops[@]}"; do
    local op_args=("${args[@]}" --operation "$op")
    if  "$rtype" == "Topic" ; then
      op_args+=(--topic "$rname")
    elif  "$rtype" == "Group" ; then
      op_args+=(--group "$rname")
    else
      log_json ERROR "Unsupported resource type" "$resource" "apply-acl" "failed"
      return 1
    fi
    log_json INFO "Plan ACL grant" "$resource" "add-acl" "planned"
    if ! $DRY_RUN; then
      kafka-acls.sh "${op_args[@]}"
      log_json INFO "ACL applied" "$resource" "add-acl" "applied"
    fi
  done

  if  -n "$group" && "$rtype" == "Topic" ; then
    for op in "${ops[@]}"; do
      if  "$op" == "Read" ; then
        local group_args=("${KAFKA_CLI_PREFIX[@]}" --add --allow-principal "$principal" --operation Read --group "$group" --resource-pattern-type Literal --host "$host")
        log_json INFO "Plan consumer group ACL" "$group" "add-group-acl" "planned"
        if ! $DRY_RUN; then
          kafka-acls.sh "${group_args[@]}"
          log_json INFO "Consumer group ACL applied" "$group" "add-group-acl" "applied"
        fi
      fi
    done
  fi
}

find "$ENV_DIR/acls" -type f -name "*.acl" | while read -r f; do
  apply_acl_file "$f"
done

Example .acl file with Prefixed pattern

principal=User:svc-producer
operation=Write,Describe
resource=Topic:orders.
resource-pattern-type=Prefixed
host=*

Verification & Acceptance Criteria

verify-apply.sh — automated gate that exits non-zero on mismatch

#!/usr/bin/env bash
# scripts/verify-apply.sh
set -euo pipefail
source "$(dirname "$0")/env.sh"

ENV_DIR=${1:?usage: $0 <environments/dev|environments/prod>}
CORRELATION_ID="${CORRELATION_ID:-$(uuidgen)}"

verify_topic() {
  local f=$1
  local name partitions rf
  name=$(grep '^name=' "$f" | cut -d'=' -f2-)
  partitions=$(grep '^partitions=' "$f" | cut -d'=' -f2-)
  rf=$(grep '^replication.factor=' "$f" | cut -d'=' -f2-)

  local desc
  desc=$(kafka-topics.sh "${KAFKA_CLI_PREFIX[@]}" --topic "$name" --describe)
  local actual_p actual_rf
  actual_p=$(echo "$desc" | awk -F ':' '/PartitionCount/ {print $3}' | tr -d ' ')
  actual_rf=$(echo "$desc" | awk -F ':' '/ReplicationFactor/ {print $3}' | tr -d ' ')

  if  "$actual_p" != "$partitions" ; then
    echo "FAIL: $name partitions expected $partitions got $actual_p" >&2
    return 1
  fi
  if  "$actual_rf" != "$rf" ; then
    echo "FAIL: $name replication factor expected $rf got $actual_rf" >&2
    return 1
  fi
  echo "OK: $name matches desired partitions=$partitions rf=$rf"
}

verify_acls() {
  local f=$1
  local principal resource
  principal=$(grep '^principal=' "$f" | cut -d'=' -f2-)
  resource=$(grep '^resource=' "$f" | cut -d'=' -f2-)
  local rtype rname
  IFS=':' read -r rtype rname <<< "$resource"

  local list_out
  if  "$rtype" == "Topic" ; then
    list_out=$(kafka-acls.sh "${KAFKA_CLI_PREFIX[@]}" --list --topic "$rname")
  elif  "$rtype" == "Group" ; then
    list_out=$(kafka-acls.sh "${KAFKA_CLI_PREFIX[@]}" --list --group "$rname")
  else
    return 1
  fi

  if ! echo "$list_out" | grep -q "Principal: $principal"; then
    echo "FAIL: ACL for $principal on $resource not found" >&2
    return 1
  fi
  echo "OK: ACL for $principal on $resource present"
}

verify_smoke() {
  local topic=$1
  local test_msg="smoke-$(date +%s)"
  echo "$test_msg" | kafka-console-producer.sh "${KAFKA_CLI_PREFIX[@]}" --topic "$topic" --producer-property acks=all >/dev/null
  local consumed
  consumed=$(kafka-console-consumer.sh "${KAFKA_CLI_PREFIX[@]}" --topic "$topic" --from-beginning --max-messages 1 --timeout-ms 5000 2>/dev/null || true)
  if  "$consumed" != "$test_msg" ; then
    echo "FAIL: Smoke test mismatch for $topic" >&2
    return 1
  fi
  echo "OK: Smoke test passed for $topic"
}

verify_lag() {
  local group=$1
  local threshold=${2:-10}
  local timeout=${3:-30}
  local start=$SECONDS
  while (( SECONDS - start < timeout )); do
    local lag_out
    lag_out=$(kafka-consumer-groups.sh "${KAFKA_CLI_PREFIX[@]}" --describe --group "$group" 2>/dev/null || true)
    local max_lag=0
    while IFS= read -r line; do
      if  "$line" =~ LAG[[:space:+([0-9]+) ]]; then
        local lag=${BASH_REMATCH[1]}
        if (( lag > max_lag )); then max_lag=$lag; fi
      fi
    done <<< "$lag_out"
    if (( max_lag < threshold )); then
      echo "OK: Consumer group $group lag $max_lag < $threshold"
      return 0
    fi
    sleep 2
  done
  echo "FAIL: Consumer group $group lag $max_lag >= $threshold after ${timeout}s" >&2
  return 1
}

# Run all verifications
find "$ENV_DIR/topics" -type f -name "*.properties" | while read -r f; do
  verify_topic "$f" || exit 1
done

find "$ENV_DIR/acls" -type f -name "*.acl" | while read -r f; do
  verify_acls "$f" || exit 1
done

# Smoke test on first topic (pilot scope)
first_topic=$(find "$ENV_DIR/topics" -type f -name "*.properties" | head -1 | xargs grep '^name=' | cut -d'=' -f2-)
verify_smoke "$first_topic" || exit 1

# Lag check for canary group (constructed example)
verify_lag "grp.events.test" 10 30 || exit 1

echo "ALL VERIFICATIONS PASSED"

Expected verification outputs

OK: events.test.v1 matches desired partitions=3 rf=3
OK: ACL for User:svc-producer on Topic:events.test.v1 present
OK: ACL for User:svc-consumer on Topic:events.test.v1 present
OK: Smoke test passed for events.test.v1
OK: Consumer group grp.events.test lag 0 < 10

Failure Modes, Troubleshooting & Runbooks

• Symptom — Likely cause — Immediate action — Recovery step
• Connection refused or timeout — Wrong bootstrap servers or firewall — Verify KAFKA_BOOTSTRAP and network with openssl s_client — Update inventory and re-run connectivity check
• Authorization failed — Missing/wrong ACLs or credentials — Verify principal and ACLs via kafka-acls.sh --list — Apply correct ACLs; re-run smoke test
• Topic create fails with RF error — RF larger than broker count — Adjust RF ≤ broker count — Recreate topic with valid RF or add brokers
• Partition decrease requested — Unsafe mutation detected — Script blocks change by design — Increase only; for decrease, create new topic and migrate
• Topic config drift detected — Manual change outside CI — kafka-configs.sh --describe shows mismatch — Re-apply from Git or promote manual change to Git via PR
• Partition increase stuck — Replica assignment not converging — kafka-topics.sh --describe shows Replicas vs ISR mismatch — Wait for ISR sync or trigger reassignment via kafka-reassign-partitions.sh
• ACL propagation delay — Authorizer cache (default 30s) — Wait authorizer.cache.seconds or restart brokers if using SimpleAclAuthorizer without cache expiry — Re-verify after cache expiry
• Schema registry compatibility failure — Breaking schema change — Block deploy in CI — Run compatibility check in CI; evolve schema backward-compatibly
• High consumer lag after change — ACL or config mismatch — Check ACLs and group membership — Fix ACLs/config; restart consumers; monitor lag
• Pipeline step hangs on CLI — DNS or TLS handshake issues — Test with openssl s_client -connect b1:9092 — Fix DNS/certs; retry after validation

Rollback Procedures (Operational Detail)

Config rollback (idempotent revert)

git revert <sha>
./scripts/safe-apply-topics.sh environments/dev
./scripts/safe-apply-acls.sh environments/dev
./scripts/verify-apply.sh environments/dev

Topic replacement for incompatible changes (partition decrease, cleanup.policy change)

1. Create topic.v2 with desired config in declarative files.
2. Dual-write: configure producers to write to both topic.v1 and topic.v2.
3. Mirror historical data with MirrorMaker 2 or custom replicator.
4. Cut over consumers to topic.v2 (update consumer group subscriptions).
5. After TTL (for example 7 days), delete topic.v1 via PR removing its .properties file.

RF correction via reassignment

# Generate reassignment JSON for RF increase
kafka-reassign-partitions.sh "${KAFKA_CLI_PREFIX[@]}" \
  --topics-to-move-json-file topics.json \
  --broker-list "0,1,2" \
  --generate > reassignment.json

# Review reassignment.json, then execute
kafka-reassign-partitions.sh "${KAFKA_CLI_PREFIX[@]}" \
  --reassignment-json-file reassignment.json \
  --execute

# Verify
kafka-reassign-partitions.sh "${KAFKA_CLI_PREFIX[@]}" \
  --reassignment-json-file reassignment.json \
  --verify

Example reassignment.json for RF increase from 2 to 3

{
  "version": 1,
  "partitions": [
    {"topic": "events.test.v1", "partition": 0, "replicas": [0,1,2]},
    {"topic": "events.test.v1", "partition": 1, "replicas": [1,2,0]},
    {"topic": "events.test.v1", "partition": 2, "replicas": [2,0,1]}
  ]
}

ACL over-grant remediation

# Apply restrictive ACL first
kafka-acls.sh "${KAFKA_CLI_PREFIX[@]}" --add --allow-principal User:svc-producer --operation Write --topic events.test.v1 --resource-pattern-type Literal
# Remove broad ACL
kafka-acls.sh "${KAFKA_CLI_PREFIX[@]}" --remove --allow-principal User:svc-producer --operation All --topic events.test.v1 --resource-pattern-type Literal
# Verify
kafka-acls.sh "${KAFKA_CLI_PREFIX[@]}" --list --topic events.test.v1

Full environment rollback

git tag pre-change-$(date +%s)
git checkout pre-change-<timestamp>
./scripts/safe-apply-topics.sh environments/dev
./scripts/safe-apply-acls.sh environments/dev
./scripts/verify-apply.sh environments/dev

CI/CD Pipeline (Production-Grade)

GitHub Actions workflow with plan/apply/verify/rollback jobs

# .github/workflows/kafka-cd.yml
name: Kafka CD
on:
  push:
    branches: [main]
    paths:
      - 'environments/**'
  workflow_dispatch:
    inputs:
      environment:
        type: choice
        options: [dev, prod]
        required: true
      action:
        type: choice
        options: [plan, apply, rollback]
        required: true

permissions:
  contents: read
  id-token: write

env:
  KAFKA_BIN_DIR: /opt/kafka/bin

jobs:
  plan:
    runs-on: self-hosted
    timeout-minutes: 10
    outputs:
      diff: ${{ steps.diff.outputs.result }}
    steps:
      - uses: actions/checkout@v4
      - name: Install Kafka CLI
        run: |
          docker pull apache/kafka:3.6.1
          echo "/opt/kafka/bin" >> $GITHUB_PATH
      - name: Inject secrets to client.properties
        env:
          KAFKA_USER: ${{ secrets.KAFKA_USER }}
          KAFKA_PASSWORD: ${{ secrets.KAFKA_PASSWORD }}
          TRUSTSTORE_PASSWORD: ${{ secrets.TRUSTSTORE_PASSWORD }}
        run: |
          envsubst < client.properties.template > client.properties
      - name: Dry-run topics
        id: diff
        run: |
          ./scripts/safe-apply-topics.sh --dry-run environments/${{ github.event.inputs.environment || 'dev' }} > plan-topics.jsonl
          ./scripts/safe-apply-acls.sh --dry-run environments/${{ github.event.inputs.environment || 'dev' }} > plan-acls.jsonl
          cat plan-topics.jsonl plan-acls.jsonl
      - name: Upload plan artifacts
        uses: actions/upload-artifact@v4
        with:
          name: plan-output
          path: plan-*.jsonl

  apply:
    needs: plan
    if: github.event.inputs.action == 'apply' || github.ref == 'refs/heads/main'
    runs-on: self-hosted
    timeout-minutes: 15
    environment: ${{ github.event.inputs.environment || 'dev' }}
    steps:
      - uses: actions/checkout@v4
      - name: Install Kafka CLI
        run: docker pull apache/kafka:3.6.1
      - name: Inject secrets
        env:
          KAFKA_USER: ${{ secrets.KAFKA_USER }}
          KAFKA_PASSWORD: ${{ secrets.KAFKA_PASSWORD }}
          TRUSTSTORE_PASSWORD: ${{ secrets.TRUSTSTORE_PASSWORD }}
        run: envsubst < client.properties.template > client.properties
      - name: Apply topics
        run: ./scripts/safe-apply-topics.sh environments/${{ github.event.inputs.environment || 'dev' }}
      - name: Apply ACLs
        run: ./scripts/safe-apply-acls.sh environments/${{ github.event.inputs.environment || 'dev' }}

  verify:
    needs: apply
    runs-on: self-hosted
    timeout-minutes: 10
    steps:
      - uses: actions/checkout@v4
      - name: Install Kafka CLI
        run: docker pull apache/kafka:3.6.1
      - name: Inject secrets
        env:
          KAFKA_USER: ${{ secrets.KAFKA_USER }}
          KAFKA_PASSWORD: ${{ secrets.KAFKA_PASSWORD }}
          TRUSTSTORE_PASSWORD: ${{ secrets.TRUSTSTORE_PASSWORD }}
        run: envsubst < client.properties.template > client.properties
      - name: Run verification
        run: ./scripts/verify-apply.sh environments/${{ github.event.inputs.environment || 'dev' }}

  rollback:
    if: github.event.inputs.action == 'rollback'
    runs-on: self-hosted
    timeout-minutes: 15
    environment: ${{ github.event.inputs.environment || 'dev' }}
    steps:
      - uses: actions/checkout@v4
      - name: Install Kafka CLI
        run: docker pull apache/kafka:3.6.1
      - name: Inject secrets
        env:
          KAFKA_USER: ${{ secrets.KAFKA_USER }}
          KAFKA_PASSWORD: ${{ secrets.KAFKA_PASSWORD }}
          TRUSTSTORE_PASSWORD: ${{ secrets.TRUSTSTORE_PASSWORD }}
        run: envsubst < client.properties.template > client.properties
      - name: Rollback to previous tag
        run: |
          git fetch --tags
          PREV_TAG=$(git tag -l 'pre-change-*' --sort=-creatordate | head -1)
          git checkout "$PREV_TAG"
          ./scripts/safe-apply-topics.sh environments/${{ github.event.inputs.environment || 'dev' }}
          ./scripts/safe-apply-acls.sh environments/${{ github.event.inputs.environment || 'dev' }}
          ./scripts/verify-apply.sh environments/${{ github.event.inputs.environment || 'dev' }}

Runner setup notes

• Self-hosted runners in the same VPC/subnet as brokers; security groups allow SASL_SSL port (typically 9093).
• Pinned Kafka CLI via Docker image apache/kafka:3.6.1 to guarantee version parity.
• OIDC token exchange with Vault or cloud secrets manager for client.properties injection; never hardcode secrets.

Security Hardening

• Least-privilege service accounts: Separate principals for CI (admin), producers, consumers.
• CI principal ACLs: Cluster:Alter,Describe,ClusterAction + Topic:Create,Alter,Describe,Read,Write on managed topics only using Prefixed pattern Topic:svc-.*.
• Audit logging: Enable Kafka authorizer logging (log4j.logger.kafka.authorizer.logger=INFO, authorizerAppender), ship to SIEM.
• Credential rotation: Automate SCRAM credential rotation via kafka-configs.sh --alter --entity-type users --entity-name <user> --add-config 'SCRAM-SHA-512=[password=...]'.

Performance & Scale Considerations

• Batch applies: Scripts process files sequentially; for hundreds of resources, parallelize with xargs -P 4 -I {} ./scripts/safe-apply-topics.sh {} and rate-limit via sleep 0.1.
• Stagger partition increases: Avoid thundering herd on controller; apply partition increases per topic with 30-second intervals.
• Monitor controller metrics during applies: kafka.controller:type=KafkaController,name=ActiveControllerCount, OfflinePartitionsCount, LeaderElectionRateAndTimeMs.

Observability & Audit

• Correlation ID passed through CI → scripts → Kafka client.id (set via KAFKA_OPTS="-Dclient.id=ci-${CORRELATION_ID}").
• Structured logs (JSON Lines) shipped to Loki/Elastic; dashboards for apply duration, success rate, drift detection.
• Git commit SHA tagged on Kafka resources for traceability:

  kafka-configs.sh "${KAFKA_CLI_PREFIX[@]}" --alter --entity-type topics --entity-name events.test.v1 --add-config "git.commit=$(git rev-parse HEAD)"

Realistic Technical Scenario

Scenario: Team adds a new event stream orders.enriched.v1 with 12 partitions, RF=3, retention 7 days, producer svc-orders-enricher, consumer group grp-orders-downstream.

Walk-through:

1. Create environments/dev/topics/orders.enriched.v1.properties:

   name=orders.enriched.v1
   partitions=12
   replication.factor=3
   config.cleanup.policy=delete
   config.retention.ms=604800000

1. Create environments/dev/acls/producer.orders.enriched.v1.acl:

   principal=User:svc-orders-enricher
   operation=Write,Describe
   resource=Topic:orders.enriched.v1
   resource-pattern-type=Literal
   host=*

1. Create environments/dev/acls/consumer.orders.enriched.v1.acl:

   principal=User:svc-orders-downstream
   operation=Read,Describe
   resource=Topic:orders.enriched.v1
   resource-pattern-type=Literal
   host=*
   consumer.group=grp-orders-downstream

1. Open PR; CI runs plan job showing create + 2 ACLs as JSON Lines artifact.
2. Reviewer approves; merge triggers apply to dev.
3. verify job passes; promote to prod via manual workflow_dispatch with action: apply, environment: prod.

Operations Checklist

• Step — What to do — Done
• 1 — Review environment inventory and credentials
• 2 — Validate connectivity to brokers
• 3 — Confirm declarative files changed only in target env
• 4 — Run safe-apply-topics.sh --dry-run for target env
• 5 — Run safe-apply-acls.sh --dry-run for target env
• 6 — Review plan output; approve or adjust
• 7 — Run safe-apply-topics.sh for target env
• 8 — Run safe-apply-acls.sh for target env
• 9 — Describe topics and list ACLs to confirm desired state
• 10 — Run smoke-produce-consume.sh on test topic
• 11 — Check consumer lag for canary group
• 12 — If any check fails, roll back by reverting and re-apply
• 13 — Record commands and outputs for audit

Extending Beyond the Pilot

Once the pilot is stable, extend the same patterns:

• Additional topics and principals: Keep files small and explicit; group by service under environments/<env>/topics/<service>/.
• Client quotas: Manage per-user quotas declaratively with kafka-configs.sh --alter --entity-type clients --entity-name <principal> --add-config 'producer_byte_rate=1048576,consumer_byte_rate=2097152'.
• Schema registry integration: Add schemas/ directory with Avro/Protobuf files; run compatibility check (schema-registry-cli compatibility-check) in CI plan job before apply.
• Stream processing integrations: Connect Apache Flink, Spark Structured Streaming, or Kafka Streams apps using the same define-review-deploy discipline. Begin each integration with a narrow, testable flow.

Conclusion

Automating Apache Kafka with CI/CD is safest when you start with a narrow, measurable pilot and expand only after you have clear verification and recovery paths. Define your desired state declaratively in Git, apply changes with idempotent scripts that refuse unsafe operations, and verify every change with concrete commands and expected outputs. When failures occur, treat the logs and the runbook as your guide: roll back configs via git revert, replace topics when incompatible changes are needed, correct replication factor through reassignment, and tighten over-granted ACLs immediately. From here, standardize your repository structure, encode your environment inventory, and enable the pilot for a single environment. When the pilot is stable for topics and ACLs, extend the same patterns to quotas, schemas, and additional environments. Keep each step observable and reversible to minimize risk and rework while delivering faster, safer Kafka changes.