Apache Kafka Cheat Sheet

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 3 --replication-factor 1
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic orders
# KRaft mode needs no ZooKeeper
# replication-factor cannot exceed the broker count

orders / partition 0 / offset 2
# offsets start at 0 and grow inside one partition
# order is guaranteed in that partition only, not across the topic

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --from-beginning
# without --from-beginning the console consumer starts at the latest offset

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders
# bootstrap.servers only needs enough brokers to learn the leaders
# the producer then appends to the leader of the chosen partition

acks=all
enable.idempotence=true
retries=2147483647
delivery.timeout.ms=120000
compression.type=zstd
linger.ms=5
# delivery.timeout.ms is the cap on retries
# linger.ms waits so a batch can fill

acks=0    do not wait; the write can vanish
acks=1    the leader has it; followers may not
acks=all  wait for the in-sync replicas
# consumers pull; brokers do not push

orders has 3 partitions
billing:    A gets 0 and 1, B gets 2
analytics:  C gets 0, 1, and 2
# one partition goes to one consumer in a group
# extra members past the partition count sit idle
# a second group has its own offsets

bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic orders --group billing --from-beginning
# a second process with --group billing joins and triggers a rebalance

enable.auto.commit=false
max.poll.interval.ms=300000
# 1. poll a batch
# 2. process with an idempotency key, such as order_id + event_type
# 3. write the result
# 4. commitSync only after step 3
# a poison record goes to orders.dlq, then commit the original offset

bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group billing
# lag is how far the group sits behind the end of each partition
# committed offsets are stored in __consumer_offsets

bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic orders --property parse.key=true --property key.separator=:
customer-42:{"order_id":"o1","total":19.5}
# the same key keeps landing on the same partition

events for one customer must stay in order  ->  customer_id
independent orders you want spread out      ->  order_id
country on a global app                      ->  one hot partition
# a null key does not keep entity order
# more partitions do not fix a low-cardinality key

# a new partition count remaps keys for records written after the change
# old records stay on the partition where they were appended
# a compacted topic needs a stable business key

bin/kafka-topics.sh --bootstrap-server localhost:9092 --create --topic orders --partitions 6 --replication-factor 3 --config min.insync.replicas=2
# the leader serves the partition; followers copy it
# ISR means replicas caught up enough to take over
unclean.leader.election.enable=false

acks=all
min.insync.replicas=2
# the write is acknowledged only when enough in-sync replicas have it
# if the ISR shrinks below 2, the produce fails instead of pretending
# replication factor 3 is durability, not 3x throughput

# at-most-once:  commit, then process  -> a crash can lose the record
# at-least-once: process, then commit -> a crash can deliver it twice
# exactly-once inside Kafka: idempotent producer plus a transaction
# a database outside Kafka still needs its own idempotent write

enable.idempotence=true
transactional.id=checkout-1
isolation.level=read_committed
# isolation.level is a consumer setting so aborted transactions stay hidden
# transactional.id stays stable for that producer instance

cleanup.policy=delete
retention.ms=604800000
retention.bytes=-1
# 7 days; -1 means no size cap
# a slow consumer does not stop cleanup
# replay works only for records retention still holds

cleanup.policy=compact
min.cleanable.dirty.ratio=0.5
# compaction is asynchronous and needs a key
# cleanup.policy=compact,delete also drops very old segments

# key set, value null, on a compacted topic, deletes that key
customer-7
delete.retention.ms=86400000
# the tombstone stays long enough for consumers to observe the delete

{
  "type": "record",
  "name": "Order",
  "namespace": "app.orders",
  "fields": [
    {
      "name": "order_id",
      "type": "string"
    },
    {
      "name": "total_usd",
      "type": "double"
    },
    {
      "name": "coupon_code",
      "type": [
        "null",
        "string"
      ],
      "default": null
    }
  ]
}
# Kafka stores bytes; JSON is easy, Avro and Protobuf are schema-driven

# version 1: order_id, total_usd
# version 2: add coupon_code optional, with a default
# BACKWARD  new consumers can read old records
# FORWARD   old consumers can read new records
# renaming total_usd breaks readers; an optional field usually does not

# TopicNameStrategy stores the value schema as orders-value
# set compatibility before a second team depends on the topic
# Schema Registry keeps the version outside the record body

# source: a file tail, or Debezium CDC, into a topic
# sink: a topic out to a file, a database, or object storage
# distributed workers store config, offsets, and status in internal topics
# keep transforms small; business rules belong in the consumer or in Streams

{
  "name": "orders-file-sink",
  "config": {
    "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
    "tasks.max": "1",
    "topics": "orders",
    "file": "/tmp/orders.sink.txt",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

curl -s -X POST localhost:8083/connectors -H 'Content-Type: application/json' -d @orders-sink.json
curl -s localhost:8083/connectors/orders-file-sink/status
# the connector can exist while one task has failed
# FileStreamSourceConnector uses topic; the sink uses topics

StreamsBuilder builder = new StreamsBuilder();
KStream<String, String> orders = builder.stream("orders");
orders.filter((storeId, json) -> json != null && !json.startsWith("test"))
    .groupByKey()
    .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(1)))
    .count(Materialized.as("store-order-counts"))
    .toStream()
    .to("store-order-counts");
// key the input by store_id; this app is a client, not a cluster

KStream<String, String> orders = builder.stream("orders");
KTable<String, String> customers = builder.table("customers");
orders.join(customers, (order, customer) -> order + "|" + customer)
    .to("orders-enriched");
// both sides use the same key, here customer_id
// a stream is the history; a table is the latest value per key

application.id=store-order-counts
bootstrap.servers=localhost:9092
processing.guarantee=at_least_once
# count() keeps a local store backed by a changelog topic
# exactly_once_v2 is the stronger guarantee, and it still stops at Kafka

listeners=SASL_SSL://:9093
security.inter.broker.protocol=SASL_SSL
ssl.keystore.location=/var/kafka/kafka.keystore.jks
ssl.keystore.password=<secret>
ssl.truststore.location=/var/kafka/kafka.truststore.jks
ssl.truststore.password=<secret>
sasl.enabled.mechanisms=SCRAM-SHA-512
sasl.mechanism.inter.broker.protocol=SCRAM-SHA-512
authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer
allow.everyone.if.no.acl.found=false
super.users=User:admin
# KRaft authorizer; encrypt clients and broker-to-broker traffic

security.protocol=SASL_SSL
sasl.mechanism=SCRAM-SHA-512
sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="billing" password="<secret>";
ssl.truststore.location=/etc/kafka/client.truststore.jks
# one principal per application, and rotate it like a database password

bin/kafka-acls.sh --bootstrap-server localhost:9093 --command-config admin.properties --add --allow-principal User:checkout --operation Write --operation Describe --topic orders
bin/kafka-acls.sh --bootstrap-server localhost:9093 --command-config admin.properties --add --allow-principal User:billing --operation Read --operation Describe --topic orders
bin/kafka-acls.sh --bootstrap-server localhost:9093 --command-config admin.properties --add --allow-principal User:billing --operation Read --group billing
# no wildcard principal; grant a transactional id only to the producer that uses it