# Kafka

> Operate Apache Kafka: topics, partitions and consumer groups, the shell tools, offset resets, lag, retention, compaction, producer and consumer tuning, and what to check when it misbehaves.

Canonical: https://www.wiki.jodisand.me/kafka/
Reviewed: 2026-09-24
Related: [Docker](https://www.wiki.jodisand.me/docker/index.md), [Kubernetes](https://www.wiki.jodisand.me/kubernetes/index.md), [Prometheus](https://www.wiki.jodisand.me/prometheus/index.md), [Elasticsearch](https://www.wiki.jodisand.me/elasticsearch/index.md)


## Cheatsheet

| Task | Command |
| --- | --- |
| List topics | `kafka-topics.sh --bootstrap-server localhost:9092 --list` |
| Create a topic | `kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --partitions 6 --replication-factor 3` |
| Describe a topic (leaders, ISR) | `kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic my-topic` |
| Under-replicated partitions, cluster-wide | `kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions` |
| Add partitions (irreversible) | `kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my-topic --partitions 12` |
| Topic configuration overrides | `kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --describe` |
| Change retention | `kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --add-config retention.ms=86400000` |
| Produce from stdin | `kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-topic` |
| Consume from the start | `kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning` |
| Consume with keys and timestamps | `kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --property print.key=true --property print.timestamp=true` |
| List consumer groups | `kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list` |
| Lag per partition | `kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group` |
| Lag for every group | `kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --all-groups` |
| Preview an offset reset | `kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --topic my-topic --reset-offsets --to-earliest --dry-run` |
| Latest offset of each partition | `kafka-get-offsets.sh --bootstrap-server localhost:9092 --topic my-topic --time -1` |
| Log size on disk per partition | `kafka-log-dirs.sh --bootstrap-server localhost:9092 --describe --topic-list my-topic` |
| KRaft controller quorum | `kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status` |
| Preferred leader election | `kafka-leader-election.sh --bootstrap-server localhost:9092 --election-type PREFERRED --all-topic-partitions` |
| Delete a topic (destroys data) | `kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic my-topic` |
| Delete a consumer group | `kafka-consumer-groups.sh --bootstrap-server localhost:9092 --delete --group my-group` |
| Throughput test | `kafka-producer-perf-test.sh --topic my-topic --num-records 100000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=localhost:9092` |

Commands assume Kafka 4.x, which runs KRaft only; ZooKeeper support was removed in 4.0 and `--zookeeper` flags no longer exist. Tools live in `bin/` of the distribution (`/opt/kafka/bin` in the official image) and every one takes `--bootstrap-server`; add `--command-config client.properties` for TLS or SASL. Reference: the [Apache Kafka documentation](https://kafka.apache.org/documentation/).

## Topics, partitions, offsets and groups

A topic is a name. A partition is the unit of storage and parallelism: an append-only log of records on disk, replicated to `replication-factor` brokers, one of which is the leader that serves all reads and writes. An offset is a record's position in its partition, a monotonically increasing 64-bit integer that never reuses a value even after deletion. Ordering is guaranteed within a partition only. A record with a key always lands in the same partition (murmur2 hash of the key modulo partition count), so ordering per key holds until the partition count changes.

A consumer group is a set of consumers sharing a `group.id`. Each partition is assigned to exactly one consumer in the group, so a group can never use more consumers than the topic has partitions; the extras sit idle. The group stores its committed offset per partition in the internal `__consumer_offsets` topic. Lag is `log-end-offset - committed-offset`, summed across partitions. Two groups reading the same topic are independent; Kafka does not delete records on consumption, only on retention.

The in-sync replica set (ISR) is the subset of a partition's replicas that are caught up with the leader within `replica.lag.time.max.ms` (default 30 s). A write with `acks=all` is acknowledged when every ISR member has it, and `min.insync.replicas` sets how small the ISR may shrink before those writes are refused. A partition with fewer replicas in sync than configured is under-replicated; a partition whose leader is gone and no ISR member can take over is offline.

```sh
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
```

`Replicas` is the assignment, `Isr` is who is actually caught up. A leader outside `Isr` is impossible; a leader of `-1` or `none` means the partition is offline.

## KRaft mode

KRaft replaces ZooKeeper with a Raft quorum of controller nodes that store cluster metadata in the internal `__cluster_metadata` log. Every broker replicates that log and serves metadata from a local cache. A node runs as `broker`, `controller` or both (`process.roles=broker,controller`, fine for development and for small clusters of three combined nodes; keep controllers separate on larger clusters so a broker under load cannot stall metadata).

```ini
# 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
```

Storage must be formatted with a cluster ID before the first start, and every node in a cluster must use the same ID. Formatting an already-formatted directory is refused unless `--ignore-formatted` is passed; passing it on a directory with data does nothing useful and mixing cluster IDs makes a broker refuse to join (`INCONSISTENT_CLUSTER_ID`).

```sh
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
```

A quorum of `n` controllers tolerates `(n-1)/2` failures, so run three or five, never an even number. `advertised.listeners` is what clients receive in metadata responses and then connect to; it is the single most common misconfiguration, and shows up as clients reaching the bootstrap server and then timing out on the address the broker advertised.

## Topics and configuration

```sh
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
```

Partition count cannot be reduced. Adding partitions changes which partition a key maps to, so consumers relying on per-key ordering see out-of-order records across the boundary; size the topic up front (partitions equal to the largest number of consumers you expect, rounded up) rather than growing it under load. Replication factor is changed with `kafka-reassign-partitions.sh`, not `--alter`.

Topic configuration is layered: a per-topic override set with `kafka-configs.sh` wins over the broker default (`log.retention.ms`, `log.segment.bytes` and so on in `server.properties`). Names differ between the two layers; the [topic configs](https://kafka.apache.org/documentation/#topicconfigs) table lists both.

```sh
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
```

## Console producer and consumer

```sh
# 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>'
```

Without `--group` the console consumer generates a random `console-consumer-NNNNN` group and does not commit offsets, so it never interferes with real consumers; with `--group my-group` it joins that group and takes partitions away from the real members. Use a distinct debug group name.

`kcat` (formerly kafkacat) is the faster alternative for ad-hoc work: `kcat -b localhost:9092 -L` lists metadata, `kcat -b localhost:9092 -t my-topic -C -o -10 -e` prints the last 10 records per partition and exits, `kcat -b localhost:9092 -t my-topic -P -K:` produces keyed records.

## Consumer groups, lag and offset resets

```sh
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
```

A `CONSUMER-ID` of `-` with a non-zero lag means nobody is consuming that partition: the group is `Empty` (everything is `-`) or has fewer members than partitions in a way the assignor left uncovered. `CURRENT-OFFSET` of `-` means the group has never committed for that partition.

Committed offsets for an `Empty` group expire after `offsets.retention.minutes` (default 7 days). A consumer that restarts after that finds no offset and falls back to `auto.offset.reset`, which is `latest` by default, so a week-long outage silently skips everything that arrived in between. Set `auto.offset.reset=earliest` on consumers where reprocessing is cheaper than loss.

Offset resets rewrite the committed position and only work while the group has no active members; the tool refuses otherwise. Every reset needs `--dry-run` (print the new offsets) or `--execute` (apply them). Scope with `--topic my-topic`, `--topic my-topic:0,1` for specific partitions, or `--all-topics`.

```sh
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
```

> [!WARNING] Resets move the pointer, not the data
> `--to-latest` abandons unconsumed records and `--to-earliest` re-delivers records the application has already processed. Confirm the consumer is idempotent, or filter by timestamp downstream, before executing either. Export the current offsets first so the reset can be undone.

For continuous monitoring, [kafka_exporter](https://github.com/danielqsj/kafka_exporter) or the broker JMX exporter feeds [Prometheus](https://www.wiki.jodisand.me/prometheus/); alert on lag that grows for longer than the consumer's expected catch-up time, not on any lag at all.

```promql
# 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
```

## Retention and compaction

Each partition is a sequence of segment files. A segment rolls when it reaches `segment.bytes` (1 GiB) or `segment.ms` (7 days), and retention only ever deletes whole closed segments, never the active one. That is why a low-volume topic with `retention.ms=3600000` still holds a week of data: the active segment has not rolled. Lower `segment.ms` (or `segment.bytes`) alongside `retention.ms` on such topics.

| Config | Default | Effect |
| --- | --- | --- |
| `cleanup.policy` | `delete` | `delete` by age or size, `compact` keep the latest value per key, `compact,delete` both |
| `retention.ms` | 604800000 (7 d) | Delete segments whose newest record is older; `-1` keeps forever |
| `retention.bytes` | -1 | Per-partition size cap, so multiply by partition count for the topic total |
| `segment.bytes` | 1073741824 | Roll the active segment at this size |
| `segment.ms` | 604800000 | Roll the active segment after this time even if small |
| `min.cleanable.dirty.ratio` | 0.5 | Compaction starts when this fraction of the log is uncompacted |
| `min.compaction.lag.ms` | 0 | Records younger than this are never compacted, guaranteeing readers see every update for that long |
| `delete.retention.ms` | 86400000 (24 h) | How long a tombstone (null value) stays visible after compaction before removal |
| `max.message.bytes` | 1048588 | Largest record batch the broker accepts for this topic |

Compaction keeps at least the last record for each key and removes older ones in closed segments; the head of the log is untouched, so a consumer reading from the start sees the compacted tail followed by every recent update. A record with a null value is a tombstone: it deletes the key once compaction runs and is itself dropped after `delete.retention.ms`, so a consumer that starts more than a day after the delete never learns the key existed. Compacted topics need keys on every record; a null key makes the producer fail for `compact` topics.

```sh
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
```

Retention runs on the `log.retention.check.interval.ms` (5 min) schedule, so a change takes a few minutes to free space. Reducing retention is the fastest way to recover a full disk, and it is destructive in the same way a reset to latest is.

## Producer settings

| Setting | Recommended | Why |
| --- | --- | --- |
| `acks` | `all` | Leader waits for every in-sync replica; with `min.insync.replicas=2` a single broker loss cannot lose acknowledged writes. `1` (leader only) and `0` trade durability for latency |
| `enable.idempotence` | `true` (default since 3.0) | Broker de-duplicates retried batches by producer ID and sequence, so retries never duplicate or reorder within a partition. Forces `acks=all`, `retries>0` and `max.in.flight.requests.per.connection<=5` |
| `delivery.timeout.ms` | 120000 | Total budget for send plus retries; a batch that cannot be delivered in this time fails the callback. Replaces tuning `retries` directly |
| `linger.ms` / `batch.size` | 5–50 ms / 64–256 KiB | Wait to fill a batch; throughput rises sharply, latency by `linger.ms` at most |
| `compression.type` | `zstd` or `lz4` | Compresses per batch; set on the producer so the broker stores compressed bytes as-is (`compression.type=producer` on the topic) |
| `max.request.size` | 1048576 | Largest single request; raise together with the topic's `max.message.bytes` and the broker's `replica.fetch.max.bytes` |
| `transactional.id` | set for exactly-once | Enables `beginTransaction`/`commitTransaction` across partitions; consumers see them only with `isolation.level=read_committed` |

Idempotence covers one producer session and one partition. Exactly-once across a consume-transform-produce loop needs transactions: the producer commits the consumer's offsets inside the same transaction (`sendOffsetsToTransaction`) so a crash either replays or commits both.

`NotEnoughReplicasException` on send means the ISR has shrunk below `min.insync.replicas`: a broker is down or lagging, and `acks=all` writes are refused until it recovers. That is the intended trade; lowering `min.insync.replicas` to keep writes flowing is choosing availability over durability.

## Consumer rebalancing

Members of a group send heartbeats to the group coordinator (the broker that leads the `__consumer_offsets` partition for that group). A member that misses `session.timeout.ms` (45 s default) or does not call `poll()` within `max.poll.interval.ms` (5 min) is removed and the group rebalances: partitions are reassigned and, under the classic protocol with the default eager assignor, every member stops consuming until the new assignment is agreed.

| Setting | Default | Note |
| --- | --- | --- |
| `session.timeout.ms` | 45000 | Missed heartbeats for this long drop the member; heartbeats run on a background thread |
| `heartbeat.interval.ms` | 3000 | Keep at a third of the session timeout |
| `max.poll.interval.ms` | 300000 | Longest gap between `poll()` calls; processing slower than this triggers a rebalance and `CommitFailedException` |
| `max.poll.records` | 500 | Records per `poll()`; lower it if processing a batch approaches `max.poll.interval.ms` |
| `partition.assignment.strategy` | `RangeAssignor, CooperativeStickyAssignor` | Set `CooperativeStickyAssignor` alone to only revoke partitions that actually move (incremental cooperative rebalancing) |
| `group.instance.id` | unset | Static membership: a restarted member with the same ID reclaims its partitions without a rebalance, as long as it returns within `session.timeout.ms` |
| `group.protocol` | `classic` | `consumer` opts in to the KIP-848 protocol (GA in 4.0): assignment is computed on the broker, rebalances are incremental and per-member, and the client-side assignor settings above no longer apply |

A rebalance storm is a group that never settles: a member joins, assignment starts, another member times out, assignment restarts. Causes in order of frequency: processing exceeding `max.poll.interval.ms`, deployments that restart members one after another without static membership, and a consumer count that keeps changing because of an autoscaler reacting to the lag the rebalances themselves cause.

```sh
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
```

## Schema registry

Kafka stores bytes; a schema registry gives producers and consumers an agreed shape. Confluent Schema Registry (Community licence) and Apicurio Registry both expose a REST API on port 8081 and both use the same wire format: a magic byte `0`, a 4-byte big-endian schema ID, then the Avro, Protobuf or JSON Schema payload. A consumer that reads such a record with a plain string deserialiser sees a leading NUL and garbage; that is the schema header, not corruption.

Schemas are versioned under a subject, by default `<topic>-key` and `<topic>-value`. A compatibility level per subject decides which changes the registry accepts: `BACKWARD` (default; a consumer on the new schema can read old data, so you may delete fields and add fields with defaults), `FORWARD`, `FULL`, their `_TRANSITIVE` variants that check every previous version, or `NONE`.

```sh
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"
```

The registry keeps its state in the `_schemas` compacted topic on the Kafka cluster it points at. Deleting that topic deletes every schema, and a soft-deleted subject (`DELETE /subjects/x`) still occupies its IDs until a hard delete (`?permanent=true`).

## Running locally with a container

The official `apache/kafka` image starts a single combined broker-and-controller node with no configuration:

```sh
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
```

The default advertises `localhost:9092`, so it works from the host and from `podman exec` and from nothing else. For other containers or another machine, set the listeners explicitly; the image maps every `KAFKA_*` environment variable to the matching `server.properties` key (upper case, dots as underscores):

```yaml
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: {}
```

See [Docker Compose](https://www.wiki.jodisand.me/docker-compose/) for the stack conventions. A single-node cluster cannot satisfy `replication-factor 3` or `min.insync.replicas=2`; create topics with `--replication-factor 1` locally and keep the production values in the deployment manifests, not in application code.

## Oneliners

```sh
# 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
```

## Scripts

Report every group whose lag exceeds a threshold, for a cron job or a Nagios-style check.

```sh
#!/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"
```

Audit topic durability: flag anything with a replication factor below 3 or without `min.insync.replicas=2`.

```sh
#!/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
```

Apply a retention change to every topic matching a pattern, with a dry run by default.

```sh
#!/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
```

Save every group's committed offsets before a risky change, so a reset can be undone with `--from-file`.

```sh
#!/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
```

## Troubleshooting

| Symptom | Cause | Fix |
| --- | --- | --- |
| Lag grows steadily on every partition | Consumers slower than producers | `--describe --group` to confirm all partitions are owned; add consumers up to the partition count, raise `max.poll.records` only if processing is per-batch, profile the consumer |
| Lag grows on one partition | Hot key, or one consumer stuck | `--describe --group --members --verbose` shows the owner; check that instance's logs and GC; repartition the key if one key dominates |
| Lag high, `CONSUMER-ID` is `-` | No live member, or offsets from a group that is `Empty` | Start the consumer; if the group has been empty longer than `offsets.retention.minutes`, expect `auto.offset.reset` to apply |
| Under-replicated partitions | A broker is down, slow disk or network, or replica fetchers behind after a restart | `--describe --under-replicated-partitions`, then that broker's `server.log`; a count that stays flat after a restart means a follower cannot catch up |
| Group stuck in `PreparingRebalance` | A member joined or left repeatedly; processing exceeds `max.poll.interval.ms` | `--describe --group --state`; client logs for `Member ... sending LeaveGroup` or `poll() timed out`; lower `max.poll.records`, use static membership, enable cooperative or KIP-848 protocol |
| `RecordTooLargeException` on produce | Record exceeds producer `max.request.size` | Raise it on the producer, then `max.message.bytes` on the topic and `replica.fetch.max.bytes` on brokers; consumers need `fetch.max.bytes` and `max.partition.fetch.bytes` to match |
| `MESSAGE_TOO_LARGE` from the broker | Batch exceeds topic `max.message.bytes` | Same chain; compressed size is what counts on the broker, uncompressed on the producer |
| `NotEnoughReplicasException` | ISR below `min.insync.replicas` with `acks=all` | Restore the missing broker; do not lower `min.insync.replicas` to work around it |
| `Timed out waiting for a node assignment` or connects then hangs | `advertised.listeners` returns an address the client cannot reach | `kafka-broker-api-versions.sh --bootstrap-server host:9092` from the client's network; fix `advertised.listeners` for that listener |
| `CommitFailedException` | Member was removed before the commit; usually `max.poll.interval.ms` exceeded | Lower `max.poll.records`, raise `max.poll.interval.ms`, move slow work off the poll thread |
| `OffsetOutOfRangeException` | Committed offset points at deleted data (retention outran the consumer) | `auto.offset.reset` decides; reset explicitly with `--to-earliest` after checking retention |
| `UNKNOWN_TOPIC_OR_PARTITION` | Topic missing and `auto.create.topics.enable=false` (recommended), or a metadata cache lag right after creation | Create it explicitly; retries after a few seconds succeed for the cache case |
| `INCONSISTENT_CLUSTER_ID` on start | `meta.properties` in `log.dirs` belongs to another cluster | Wrong data directory or a reused volume; do not reformat a directory that holds production data |
| Broker disk full, `No space left on device` | Retention too generous, or a segment that never rolls | `kafka-log-dirs.sh --describe`; lower `retention.ms` on the biggest topics, wait for the cleanup interval; add `segment.ms` on quiet topics |
| Controller quorum has no leader | Fewer than a majority of voters reachable | `kafka-metadata-quorum.sh describe --replication`; restore voters, never format a voter to force a new quorum |

## Further reading

- [Kafka documentation](https://kafka.apache.org/documentation/)
- [Topic-level configs](https://kafka.apache.org/documentation/#topicconfigs)
- [Producer configs](https://kafka.apache.org/documentation/#producerconfigs) and [consumer configs](https://kafka.apache.org/documentation/#consumerconfigs)
- [KRaft operations](https://kafka.apache.org/documentation/#kraft)
- [Docker image](https://kafka.apache.org/documentation/#docker)
- [Confluent Schema Registry API](https://docs.confluent.io/platform/current/schema-registry/develop/api.html)