Config

Set a setting (log rotation)

./bin/kafka-configs.sh --zookeeper zookeeper1:2181/kafka \
--alter --entity-type topics \
--entity-name test \
--add-config segment.ms=60000

Remove a setting (log rotation)

./bin/kafka-configs.sh --zookeeper zookeeper1:2181/kafka \
--alter --entity-type topics \
--entity-name test \
--delete-config segment.ms

Topics

Create a topic

bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka \
--topic test --create \
--partitions 3 \
--replication-factor 3 \
--if-not-exists

List all topics

bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka --list

Describe a topic

bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka --topic test --describe

Delete a topic

bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka --topic test --delete

Alter the number of partitions (can only go up)

bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka \
--alter \
--topic test \
--partitions 7

Topics that are not in-sync with all replicas

bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka \
--describe \
--under-replicated-partitions

Topics without a leader replica

bin/kafka-topics.sh --zookeeper zookeeper1:2181/kafka \
--describe \
--unavailable-partitions

Producers

Start a console producer to topic 'test'

bin/kafka-console-producer.sh --broker-list kafka1:9092 --topic test

Add acks flag to your producer (1 | 0 | all)

bin/kafka-console-producer.sh --broker-list kafka1:9092 --topic test --producer-property acks=all

Consumers

Start a console consumer to a topic

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test

Consuming messages from the beginning

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test --from-beginning

Set consumer's formatter & properties

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 \
--topic streams-wordcount-output \
--from-beginning \
--formatter kafka.tools.DefaultMessageFormatter \
--property print.key=true \
--property print.value=true \
--property key.deserializer=org.apache.kafka.common.serialization.StringDeserializer \
--property value.deserializer=org.apache.kafka.common.serialization.LongDeserializer

Consumer groups

Start a consumer group

bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic test --group app1

List the consumer groups

bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --list

Describe a consumer group

bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --describe --group appl1

Describe the active members of the group

bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --describe --group appl1 --members

Describe the state of the group

bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --describe --group appl1 --state

Partition reassigning

1. Create the json file to reassign partitions (topics.json)

{
  "topics": [{ "topic": "test" }],
  "version": 1
}

2. Run the command through a dry-run

./bin/kafka-reassign-partitions.sh --zookeeper zookeeper1:2181/kafka --generate \
--topics-to-move-json-file topics.json \
--broker-list 3,2,1

3. <proposed-reassignment-json-from-the-output-of-the-above-command> > plan.json

4. Execute the reassignment

./bin/kafka-reassign-partitions.sh --zookeeper zookeeper1:2181/kafka --execute \
--reassignment-json-file plan.json

5. Verify the reassignment completed

./bin/kafka-reassign-partitions.sh --zookeeper zookeeper1:2181/kafka --verify \
--reassignment-json-file plan.json

Misc

Output the data from the partition log

./bin/kafka-run-class.sh kafka.tools.DumpLogSegments --print-data-log \
--files <kafka-data-dir>/<partition-dir>/00000000000000000000.log

Zookeeper related

Configure zookeeper id

echo <zookeeper-server-id> > <zookeeper-data-dir>/myid

Zookeeper shell

• bin/zookeeper-shell.sh zookeeper1:2181/kafka
• ls /