What are the most common Kafka CLI commands?

Common commands include: kafka-topics.sh (create, list, describe topics), kafka-console-producer.sh (send messages), kafka-console-consumer.sh (read messages), kafka-consumer-groups.sh (manage consumer groups), and kafka-configs.sh (configure topics/brokers).

How do I create a Kafka topic?

Use: kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --partitions 3 --replication-factor 2. This creates a topic with 3 partitions and replication factor of 2 for fault tolerance.

How do I check consumer lag in Kafka?

Use: kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --describe. The LAG column shows how many messages behind the consumer group is. High lag indicates consumers are not keeping up with producers.

What are best practices for Kafka configuration?

Best practices include: set acks=all for producers (reliability), use appropriate partition counts (balance parallelism and overhead), set replication factor of 3 for production, configure retention policies based on use case, monitor consumer lag, and use idempotent producers for exactly-once semantics.

Apache Kafka Cheat Sheet: Commands, Configuration & Best Practices

This Apache Kafka Cheat Sheet provides quick reference for common commands, configurations, and code examples. Bookmark this page for daily Kafka development and operations.

Topic Management Commands

List All Topics

List all topics in the Kafka cluster

kafka-topics.sh --bootstrap-server localhost:9092 --list

Create Topic

Create a topic with 3 partitions and replication factor of 2

kafka-topics.sh --bootstrap-server localhost:9092 --create --topic my-topic --partitions 3 --replication-factor 2

Describe Topic

Show detailed information about a topic (partitions, replicas, leaders)

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic my-topic

Delete Topic

Delete a topic (requires delete.topic.enable=true)

kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic my-topic

Increase Partitions

Increase number of partitions (can only increase, not decrease)

kafka-topics.sh --bootstrap-server localhost:9092 --alter --topic my-topic --partitions 6

Producer Commands

Console Producer

Start a console producer to send messages interactively

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-topic

Producer with Key

Send messages with keys (format: key:value)

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-topic --property "parse.key=true" --property "key.separator=:"

Producer with Acks

Producer with acknowledgment (all = wait for all replicas)

kafka-console-producer.sh --bootstrap-server localhost:9092 --topic my-topic --producer-property acks=all

Consumer Commands

Console Consumer

Consume messages from beginning of topic

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

Consumer Group

Consume as part of a consumer group

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --group my-group

Consumer from Offset

Consume from specific partition and offset

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --partition 0 --offset 100

Limit Messages

Consume only first 10 messages

kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --max-messages 10

Consumer Group Commands

List Consumer Groups

List all consumer groups

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list

Describe Consumer Group

Show consumer group details (members, partitions, offsets, lag)

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --describe

Reset Offset

Reset consumer group offset to earliest (replay all messages)

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --topic my-topic --reset-offsets --to-earliest --execute

Delete Consumer Group

Delete a consumer group

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --delete

Configuration Examples

Producer Configuration

Common producer settings for reliability and performance

acks=all
retries=3
batch.size=16384
linger.ms=10
compression.type=snappy
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

Consumer Configuration

Common consumer settings for reliable message processing

group.id=my-consumer-group
auto.offset.reset=earliest
enable.auto.commit=false
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

Broker Configuration

Key broker settings for performance and retention

num.network.threads=8
num.io.threads=8
socket.send.buffer.bytes=102400
socket.receive.buffer.bytes=102400
log.retention.hours=168
log.segment.bytes=1073741824

Code Examples

Java Producer Example

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

ProducerRecord<String, String> record = new ProducerRecord<>(
    "my-topic", 
    "key", 
    "Hello Kafka!"
);

producer.send(record);
producer.close();

Java Consumer Example

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "my-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("my-topic"));

while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, String> record : records) {
        System.out.println(record.value());
    }
}

Python Producer Example

from kafka import KafkaProducer

producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    value_serializer=lambda v: v.encode('utf-8')
)

producer.send('my-topic', value='Hello Kafka!')
producer.flush()
producer.close()

Python Consumer Example

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'my-topic',
    bootstrap_servers=['localhost:9092'],
    group_id='my-group',
    value_deserializer=lambda m: m.decode('utf-8')
)

for message in consumer:
    print(message.value)

Best Practices

Producer Best Practices

Use acks=all for reliability
Enable idempotence for exactly-once semantics
Use batching for better throughput
Set appropriate retry configuration
Use compression (snappy, gzip, lz4)

Consumer Best Practices

Use consumer groups for parallel processing
Manually commit offsets for reliability
Monitor consumer lag regularly
Handle errors and retries properly
Set appropriate fetch sizes

Topic Design

Choose partition count carefully
Use meaningful topic names
Set appropriate retention policies
Use replication factor of 3+ for production
Consider compaction for keyed topics

Monitoring

Monitor consumer lag
Track broker metrics (CPU, disk, network)
Monitor topic throughput
Set up alerts for critical metrics
Use Kafka Manager or Confluent Control Center

Troubleshooting Commands

Check Consumer Lag

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group my-group --describe

Look at the LAG column. High lag indicates consumers are not keeping up.

Check Topic Size

kafka-log-dirs.sh --bootstrap-server localhost:9092 --topic-list my-topic --describe

Shows disk usage per partition for a topic.

View Broker Configuration

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 0 --describe

Shows configuration for broker with ID 0.

Validate Kafka Message Formats

Use our tools to validate JSON message schemas, generate message formats, and ensure your Kafka applications use correct data structures.

Schema Generator JSON Validator