Kafka 4.3 · commands from the labs

Cheat sheet

The commands and settings you will reach for most. Each card names the module that explains it. Use your browser's print function for a paper copy.

Set-up used on this page. The helper k runs a Kafka tool inside the first lab broker, and B is the address of the cluster from inside the containers.

k() { docker exec -i kafka-1 /opt/kafka/bin/"$@"; }
B=kafka-1:19092

Kafka 4.2 and later renamed some tool flags. Use --reader-property for the console producer, --formatter-property for the console consumer, and --command-property for client settings. The performance tool takes several settings after one --command-property, separated by spaces.

Topics

modules 01, 09

k kafka-topics.sh --bootstrap-server $B --list
k kafka-topics.sh --bootstrap-server $B --describe --topic orders
k kafka-topics.sh --bootstrap-server $B --create --topic orders \
  --partitions 3 --replication-factor 3
k kafka-topics.sh --bootstrap-server $B --create --topic t \
  --replica-assignment 3:1:2          # exact brokers; no --partitions
k kafka-topics.sh --bootstrap-server $B --alter --topic orders --partitions 6
k kafka-topics.sh --bootstrap-server $B --delete --topic orders

The partition count can only go up, and raising it moves keys.

Topic and broker settings

modules 01, 09

k kafka-configs.sh --bootstrap-server $B --describe \
  --entity-type topics --entity-name orders --all
k kafka-configs.sh --bootstrap-server $B --alter \
  --entity-type topics --entity-name orders --add-config retention.ms=86400000
k kafka-configs.sh --bootstrap-server $B --alter \
  --entity-type topics --entity-name orders --delete-config retention.ms
k kafka-configs.sh --bootstrap-server $B --describe \
  --entity-type brokers --entity-name 1 --all
k kafka-configs.sh --bootstrap-server $B --alter \
  --entity-type brokers --entity-default --add-config log.cleaner.threads=2

Produce and consume by hand

modules 01, 02

k kafka-console-producer.sh --bootstrap-server $B --topic orders \
  --reader-property parse.key=true --reader-property key.separator=:

k kafka-console-consumer.sh --bootstrap-server $B --topic orders \
  --from-beginning --formatter-property print.key=true \
  --formatter-property print.partition=true \
  --formatter-property print.offset=true

k kafka-console-consumer.sh --bootstrap-server $B --topic payments \
  --from-beginning --isolation-level read_committed

Offsets and files

module 01

k kafka-get-offsets.sh --bootstrap-server $B --topic orders
k kafka-get-offsets.sh --bootstrap-server $B --topic orders --time earliest

docker exec kafka-1 ls /tmp/kraft-combined-logs/orders-0
k kafka-dump-log.sh --print-data-log \
  --files /tmp/kraft-combined-logs/orders-0/00000000000000000000.log

k kafka-log-dirs.sh --bootstrap-server $B --describe --broker-list 1

Load tests

module 02

k kafka-producer-perf-test.sh --topic perf --num-records 300000 \
  --record-size 200 --throughput -1 \
  --command-property bootstrap.servers=$B batch.size=131072 linger.ms=20

# in transactions, one commit each second
k kafka-producer-perf-test.sh --topic payments --num-records 30 \
  --record-size 20 --throughput 10 --transactional-id pay-1 \
  --transaction-duration-ms 1000 --command-property bootstrap.servers=$B

Consumer groups and lag

module 03

k kafka-consumer-groups.sh --bootstrap-server $B --list
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g --members
k kafka-consumer-groups.sh --bootstrap-server $B --describe --group g --state
k kafka-consumer-groups.sh --bootstrap-server $B --describe --all-groups

# move a group; it must have no members. Look first with --dry-run.
k kafka-consumer-groups.sh --bootstrap-server $B --group g --topic events \
  --reset-offsets --to-earliest --dry-run
#   other targets: --to-latest  --to-datetime  --to-offset  --shift-by -100
#   then replace --dry-run with --execute

Health in three commands

modules 04, 09

k kafka-topics.sh --bootstrap-server $B --describe --unavailable-partitions
k kafka-topics.sh --bootstrap-server $B --describe --under-min-isr-partitions
k kafka-topics.sh --bootstrap-server $B --describe --under-replicated-partitions

No output means healthy. Run all three before and after every restart.

Leaders and replicas

module 04

k kafka-leader-election.sh --bootstrap-server $B \
  --election-type preferred --topic t --partition 0
k kafka-leader-election.sh --bootstrap-server $B \
  --election-type unclean --topic t --partition 0      # loses data

k kafka-reassign-partitions.sh --bootstrap-server $B \
  --reassignment-json-file /tmp/move.json --execute    # add --throttle
k kafka-reassign-partitions.sh --bootstrap-server $B \
  --reassignment-json-file /tmp/move.json --verify

k kafka-delete-records.sh --bootstrap-server $B --offset-json-file /tmp/del.json

Controllers and metadata

module 05

k kafka-metadata-quorum.sh --bootstrap-server $B describe --status
k kafka-metadata-quorum.sh --bootstrap-server $B describe --replication
k kafka-cluster.sh cluster-id --bootstrap-server $B
k kafka-features.sh --bootstrap-server $B describe
k kafka-broker-api-versions.sh --bootstrap-server $B | grep '^kafka-'

k kafka-dump-log.sh --cluster-metadata-decoder --files \
  /tmp/kraft-combined-logs/__cluster_metadata-0/00000000000000000000.log

Transactions

module 06

k kafka-transactions.sh --bootstrap-server $B list
k kafka-transactions.sh --bootstrap-server $B describe --transactional-id pay-1
k kafka-transactions.sh --bootstrap-server $B find-hanging --broker-id 1
k kafka-transactions.sh --bootstrap-server $B describe-producers \
  --topic payments --partition 0

The first two were run in the lab. The last two exist in the tool's help; I did not run them.

Kafka Streams

module 07

# read counts written as 8-byte numbers
k kafka-console-consumer.sh --bootstrap-server $B --topic streams-wordcount-output \
  --from-beginning --formatter-property print.key=true \
  --formatter-property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer

# start over: the group must be empty
k kafka-streams-application-reset.sh --bootstrap-server $B \
  --application-id streams-wordcount --input-topics streams-plaintext-input --dry-run

On Alpine images, install libstdc++ first, or the state store fails to load.

Kafka Connect, over HTTP on port 8083

module 08

GET    /connector-plugins
GET    /connectors
POST   /connectors                      {"name": …, "config": {…}}
GET    /connectors/NAME/status          check every task's state
GET    /connectors/NAME/offsets
POST   /connectors/NAME/tasks/0/restart
DELETE /connectors/NAME

The first four and the create call were run in the lab. Restart and delete are the standard calls; I did not run them.

Schema registry, on port 8086 in the lab

module 08

R=http://localhost:8086
H='Content-Type: application/vnd.schemaregistry.v1+json'
curl -s $R/config
curl -s $R/subjects/orders-value/versions
curl -s -X POST -H "$H" -d "$SCHEMA" $R/subjects/orders-value/versions
curl -s -X POST -H "$H" -d "$SCHEMA" \
  "$R/compatibility/subjects/orders-value/versions/latest?verbose=true"
curl -s -X PUT -H "$H" -d '{"compatibility":"FULL"}' $R/config/orders-value

Quotas, share groups, second cluster

modules 09, 10

k kafka-configs.sh --bootstrap-server $B --alter \
  --add-config 'producer_byte_rate=5242880' \
  --entity-type clients --entity-name slow-client

k kafka-console-share-consumer.sh --bootstrap-server $B --topic jobs --group workers
k kafka-share-groups.sh --bootstrap-server $B --describe --group workers --members

docker exec -d kafka-1 /opt/kafka/bin/connect-mirror-maker.sh /tmp/mm2.properties

More commands, with real outputs, are in the labs: security, monitoring, change data capture and tiered storage. For practice: failure drills and mock interview. For experience: case studies and projects.

Settings to start from

These are the choices the modules argue for. Change them when you have a measured reason.

Where	Setting	Value	Why
Topic	replication factor	3	Survives one broker failure with protection left.
Topic	`min.insync.replicas`	2	An acknowledged record is on two brokers. The default of 1 is not safe.
Topic	`unclean.leader.election.enable`	`false` (default)	No silent loss of committed records.
Topic, low volume	`segment.ms`	near the retention time	Otherwise data outlives retention by up to 7 days.
Producer	`acks`, `enable.idempotence`	`all`, `true` (defaults)	No loss on failover, no duplicates from retries.
Producer, for throughput	`batch.size`, `linger.ms`, `compression.type`	131072, 20, `lz4`	The lab measured 54 times more throughput from batching alone.
Producer	`client.id`	the application's name	Logs, metrics and quotas all use it.
Consumer	`enable.auto.commit`	`false` when loss matters	Commit only what is finished.
Consumer	`auto.offset.reset`	chosen on purpose	The default `latest` makes a new group skip existing data.
Consumer	`group.protocol`	`consumer` for new Java applications	Only affected partitions stop in a rebalance.
Consumer of transactional topics	`isolation.level`	`read_committed`	The default shows aborted records.
Broker	`default.replication.factor`	3	The default of 1 creates unprotected topics.
Broker	`auto.create.topics.enable`	`false`	A typing error should not create a topic.
Broker	`broker.rack`	the zone	Replicas are spread over zones.
Cluster	controllers	3, on their own servers	Survives one failure; not disturbed by broker load.
Kafka Streams	`state.dir`	a durable disk	The default temporary directory can be cleared.
Kafka Connect sink	`errors.tolerance` with a dead-letter topic	`all`, plus context headers	One bad record should not stop the pipeline, and should not vanish.

Numbers worth remembering

What	Default
Retention, and time-based segment roll	7 days each
Segment size	1 GiB
Retention check	every 5 minutes
Largest record batch	about 1 MiB
Producer batch size and linger	16384 bytes, 5 ms
Producer delivery timeout	2 minutes
Consumer session timeout, poll interval, records in a poll	45 seconds, 5 minutes, 500
Follower out of sync after	30 seconds
Broker fenced after	9 seconds without a heartbeat
Transaction timeout, and the most a client may ask	1 minute, 15 minutes
Idle group's offsets kept for	7 days
Internal topics `__consumer_offsets`, `__transaction_state`	50 partitions each
Share group record lock, delivery limit	30 seconds, 5