Kafka

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic my-topic
# Topic: my-topic  PartitionCount: 6  ReplicationFactor: 3  Configs: min.insync.replicas=2,retention.ms=604800000
#   Partition: 0  Leader: 1  Replicas: 1,2,3  Isr: 1,2,3  Elr:  LastKnownElr:
#   Partition: 1  Leader: 2  Replicas: 2,3,1  Isr: 2,3    Elr:  LastKnownElr:     <- broker 1 is behind on this one

# config/server.properties, combined node 1 of 3
process.roles=broker,controller
node.id=1
controller.listener.names=CONTROLLER
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://kafka-1.example.com:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.quorum.bootstrap.servers=kafka-1.example.com:9093,kafka-2.example.com:9093,kafka-3.example.com:9093   # dynamic quorum, Kafka 3.9+
log.dirs=/var/lib/kafka/data
num.partitions=6
default.replication.factor=3
min.insync.replicas=2
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3

KAFKA_CLUSTER_ID=$(kafka-storage.sh random-uuid)
kafka-storage.sh format --standalone -t "$KAFKA_CLUSTER_ID" -c config/server.properties        # single node: this node is the only voter
kafka-storage.sh format -t "$KAFKA_CLUSTER_ID" -c config/server.properties \
  --initial-controllers "1@kafka-1.example.com:9093:$(kafka-storage.sh random-uuid),2@kafka-2.example.com:9093:$(kafka-storage.sh random-uuid),3@kafka-3.example.com:9093:$(kafka-storage.sh random-uuid)"   # first node of a 3-controller quorum; the trailing UUID is each controller's directory.id
kafka-server-start.sh -daemon config/server.properties
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status        # LeaderId, HighWatermark, CurrentVoters, CurrentObservers
kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --replication   # per-node LogEndOffset and Lag; a voter far behind cannot vote

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic \
  --partitions 6 --replication-factor 3 \
  --config retention.ms=604800000 --config min.insync.replicas=2        # 7 days; acks=all needs 2 in-sync copies
kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --if-not-exists ...   # idempotent in scripts
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topics-with-overrides                # only topics with non-default configs
kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my-topic --partitions 12          # increase only; keyed records re-hash to new partitions
kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic my-topic                         # needs delete.topic.enable=true (the default); removes every replica

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --describe
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter \
  --add-config retention.ms=259200000,max.message.bytes=2097152                   # takes effect without restart
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --delete-config retention.ms   # back to broker default
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 1 --describe --all   # every effective broker setting and where it came from
kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-default --alter --add-config log.retention.ms=172800000   # dynamic default for all brokers

# Keyed records from stdin: "order-1:{...}" becomes key order-1, value {...}
kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-topic \
  --property parse.key=true --property key.separator=: \
  --producer-property acks=all --producer-property compression.type=zstd
printf 'order-1:{"id":1}\norder-2:{"id":2}\n' | kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-topic --property parse.key=true --property key.separator=:

# Read the newest records as they arrive (default), or everything from the start
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning \
  --property print.key=true --property print.timestamp=true --property print.partition=true --property print.offset=true
# One partition from a specific offset, stop after 10 records
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --partition 3 --offset 1200 --max-messages 10
# Join a group so offsets are committed; reruns resume where they stopped
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --group my-debug-group --timeout-ms 10000   # exit when idle for 10 s
# Read a compacted topic and show tombstones
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-compacted --from-beginning --property print.key=true --property null.literal='<null>'

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list --state Stable          # filter: Stable, Empty, Dead, PreparingRebalance, CompletingRebalance
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group
# GROUP     TOPIC     PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG   CONSUMER-ID                 HOST         CLIENT-ID
# my-group  my-topic  0          48210           48213           3     my-app-7f2c...-0d9a         /10.0.4.12   my-app
# my-group  my-topic  1          51002           59870           8868  -                           -            -          <- unassigned: no live member owns it
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members --verbose   # who owns what
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --state              # coordinator broker, protocol, state, member count

G="kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --reset-offsets"
$G --topic my-topic --to-earliest --dry-run                          # reprocess everything retained
$G --topic my-topic --to-latest --execute                            # skip the backlog; the skipped records are not consumed
$G --topic my-topic --to-datetime 2026-09-24T02:00:00.000 --execute  # first offset with timestamp >= that time (broker local time, no zone suffix)
$G --topic my-topic --by-duration PT2H --execute                     # two hours before now
$G --topic my-topic:3 --shift-by -500 --execute                      # replay the last 500 on partition 3
$G --topic my-topic:3 --to-offset 48000 --execute
$G --all-topics --to-current --dry-run --export > offsets.csv        # save current positions as topic,partition,offset
$G --all-topics --from-file offsets.csv --execute                    # restore them
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete-offsets --group my-group --topic old-topic   # forget one topic; next start uses auto.offset.reset

# Lag per group, summed over partitions
sum by (consumergroup, topic) (kafka_consumergroup_lag)
# Lag growing for 15 minutes: consumers cannot keep up or are stuck
sum by (consumergroup) (kafka_consumergroup_lag) > 1000 and deriv(sum by (consumergroup) (kafka_consumergroup_lag)[15m:1m]) > 0

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name user-state --alter \
  --add-config cleanup.policy=compact,min.compaction.lag.ms=600000,segment.ms=3600000
kafka-log-dirs.sh --bootstrap-server localhost:9092 --describe --topic-list my-topic | tail -n +3 | jq '.brokers[].logDirs[].partitions[] | {partition, size}'   # bytes per partition per broker

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --state     # PreparingRebalance or CompletingRebalance for more than a few seconds is the symptom
grep -E 'Rebalance|revoked|assigned|LeaveGroup' /var/log/my-app/app.log | tail -50               # client side: who left and why

R=http://schema-registry.example.com:8081
curl -fsS "$R/subjects" | jq .                                              # every subject
curl -fsS "$R/subjects/my-topic-value/versions" | jq .                      # version numbers
curl -fsS "$R/subjects/my-topic-value/versions/latest" | jq -r .schema | jq . # the schema itself
curl -fsS "$R/schemas/ids/42" | jq -r .schema                               # resolve an ID seen in a record header
curl -fsS "$R/config/my-topic-value" || curl -fsS "$R/config"               # subject level, then global compatibility
# Register a new version; returns {"id": N}. Fails with 409 if incompatible with the subject's level
curl -fsS -X POST -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  --data "$(jq -n --arg s "$(cat order.avsc)" '{schemaType: "AVRO", schema: $s}')" "$R/subjects/my-topic-value/versions"
# Test compatibility without registering
curl -fsS -X POST -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  --data "$(jq -n --arg s "$(cat order.avsc)" '{schema: $s}')" "$R/compatibility/subjects/my-topic-value/versions/latest" | jq .is_compatible
curl -fsS -X PUT -H 'Content-Type: application/vnd.schemaregistry.v1+json' --data '{"compatibility":"FULL_TRANSITIVE"}' "$R/config/my-topic-value"

podman run -d --name kafka -p 9092:9092 apache/kafka:4.1.0
podman exec kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --partitions 3 --replication-factor 1

services:
  kafka:
    image: apache/kafka:4.1.0
    ports: ["9092:9092"]
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:9092,CONTROLLER://:9093,INTERNAL://:19092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://localhost:9092,INTERNAL://kafka:19092   # host clients use localhost, compose peers use kafka:19092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,INTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@localhost:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1          # defaults of 3 make the internal topics uncreatable on one broker
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_LOG_DIRS: /var/lib/kafka/data
    volumes:
      - kafka-data:/var/lib/kafka/data:Z
volumes:
  kafka-data: {}

# Total lag per group, sorted
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --all-groups 2>/dev/null | awk 'NR>1 && $6 ~ /^[0-9]+$/ {lag[$1]+=$6} END {for (g in lag) print lag[g], g}' | sort -rn

# Records per partition (latest minus earliest offset)
paste <(kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic my-topic --time -2) <(kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic my-topic --time -1) | awk -F'[:\t]' '{print $2, $6-$3}'

# Partitions with no leader (unavailable)
kafka-topics.sh --bootstrap-server localhost:9092 --describe --unavailable-partitions

# Partitions whose leader is not the preferred (first) replica
kafka-topics.sh --bootstrap-server localhost:9092 --describe | awk '/Partition:/ {split($8,r,","); if ($6 != r[1]) print}'

# Topics with replication factor below 3
kafka-topics.sh --bootstrap-server localhost:9092 --describe | awk '/ReplicationFactor:/ && $8 < 3 {print $2, "rf=" $8}'

# Every topic's retention override
for t in $(kafka-topics.sh --bootstrap-server localhost:9092 --list); do printf '%s\t' "$t"; kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name "$t" --describe | grep -o 'retention.ms=[0-9-]*' || echo default; done

# Disk used per broker, all topics
kafka-log-dirs.sh --bootstrap-server localhost:9092 --describe | tail -n +3 | jq -r '.brokers[] | "\(.broker) \([.logDirs[].partitions[].size] | add / 1e9 | floor) GB"'

# Tail a topic and pretty-print JSON values
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic | jq -c .

# Last record of each partition
kcat -b localhost:9092 -t my-topic -C -o -1 -e -f '%p:%o %k %s\n'

# Timestamp of the oldest retained record on partition 0
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --partition 0 --offset earliest --max-messages 1 --property print.timestamp=true --property print.value=false

# Which broker coordinates a group
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --state | awk 'NR==2 {print $2}'

# Members of a group and their partition counts
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --members

# Groups with no members (offsets will expire after offsets.retention.minutes)
kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list --state Empty

# Inspect a segment file on a broker (record batches, keys, values)
kafka-dump-log.sh --files /var/lib/kafka/data/my-topic-0/00000000000000000000.log --print-data-log | head -40

# Producer throughput test with realistic settings
kafka-producer-perf-test.sh --topic perf-test --num-records 500000 --record-size 512 --throughput -1 --producer-props bootstrap.servers=localhost:9092 acks=all linger.ms=10 compression.type=zstd

# Consumer throughput test
kafka-consumer-perf-test.sh --bootstrap-server localhost:9092 --topic perf-test --messages 500000 --group perf-test-group

# Move every partition off broker 3 before decommissioning it (generate, then execute the JSON it prints)
kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --topics-to-move-json-file topics.json --broker-list 1,2,4 --generate

# Progress of a running reassignment
kafka-reassign-partitions.sh --bootstrap-server localhost:9092 --reassignment-json-file reassign.json --verify

# Broker API versions: confirms connectivity and the broker's version through any listener
kafka-broker-api-versions.sh --bootstrap-server kafka-1.example.com:9092 | head -3

# ACLs for a principal
kafka-acls.sh --bootstrap-server localhost:9092 --list --principal User:my-app --command-config admin.properties

#!/usr/bin/env bash
# usage: kafka-lag-check.sh <bootstrap> <threshold>; exit 2 when any group exceeds it
set -euo pipefail
bootstrap=${1:?bootstrap server required}
threshold=${2:-10000}
rc=0
while read -r lag group; do
  [[ $lag =~ ^[0-9]+$ ]] || continue
  if (( lag > threshold )); then
    printf 'CRITICAL %s lag=%s\n' "$group" "$lag"
    rc=2
  fi
done < <(
  timeout 60s kafka-consumer-groups.sh --bootstrap-server "$bootstrap" --describe --all-groups 2>/dev/null \
    | awk 'NR>1 && $6 ~ /^[0-9]+$/ {lag[$1]+=$6} END {for (g in lag) print lag[g], g}'
)
(( rc == 0 )) && echo "OK all groups under $threshold"
exit "$rc"

#!/usr/bin/env bash
set -euo pipefail
bootstrap=${1:?bootstrap server required}
kafka-topics.sh --bootstrap-server "$bootstrap" --describe 2>/dev/null \
  | awk '/ReplicationFactor:/ && $2 !~ /^__/ {print $2, $8, $0}' \
  | while read -r topic rf rest; do
      isr=$(grep -o 'min.insync.replicas=[0-9]*' <<<"$rest" | cut -d= -f2 || true)
      if (( rf < 3 )); then
        printf '%s\trf=%s\tbelow 3\n' "$topic" "$rf"
      elif [[ ${isr:-1} -lt 2 ]]; then
        printf '%s\trf=%s\tmin.insync.replicas=%s\n' "$topic" "$rf" "${isr:-default}"
      fi
    done

#!/usr/bin/env bash
# usage: kafka-bulk-retention.sh <bootstrap> <regex> <retention.ms> [apply]
set -euo pipefail
bootstrap=${1:?} pattern=${2:?} retention=${3:?} mode=${4:-dry-run}
mapfile -t topics < <(kafka-topics.sh --bootstrap-server "$bootstrap" --list | grep -E "$pattern" | grep -v '^__')
(( ${#topics[@]} )) || { echo "no topics match $pattern" >&2; exit 1; }
for t in "${topics[@]}"; do
  if [[ $mode == apply ]]; then
    kafka-configs.sh --bootstrap-server "$bootstrap" --entity-type topics --entity-name "$t" --alter --add-config "retention.ms=$retention"
  else
    printf 'would set retention.ms=%s on %s\n' "$retention" "$t"
  fi
done

#!/usr/bin/env bash
set -euo pipefail
bootstrap=${1:?} outdir=${2:-offsets-$(date +%Y%m%dT%H%M%S)}
mkdir -p "$outdir"
for g in $(kafka-consumer-groups.sh --bootstrap-server "$bootstrap" --list); do
  kafka-consumer-groups.sh --bootstrap-server "$bootstrap" --group "$g" --all-topics --reset-offsets --to-current --dry-run --export > "$outdir/$g.csv" 2>/dev/null \
    && printf 'saved %s (%s partitions)\n' "$g" "$(wc -l < "$outdir/$g.csv")"
done