Useful kafka commands

For past few months I have been working with many kafka production issues and I am still trying to learn many new things about it. But while working on any kafka issue, you always want some commands to be very handy and just one click away in a file or in a blog.

Here are few kafka commands that may be very helpful to you while learning or solving any of your kafka issues. 

Here I would be referring KAFKA_HOME as your path to kafka home.

Push data to kafka topic

KAFKA_HOME/bin/kafka-console-producer.sh --broker-list <kafka_broker_ip>:9092 --topic <topic_name> --property "parse.key=true" --property "key.separator=::"

>your_kafka_packet_key::your_kafka_packet_value

Read data from kafka topic

KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server <kafka_broker_ip>:9092 --topic <topic_name>

Read data from kafka topic from beginning

KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server <kafka_broker_ip>:9092 --topic <topic_name> --from-beginning

Read kafka topic data at particular offset

KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server <kafka_broker_ip>:9092 --topic <topic_name> --max-messages 1  --partition <parition> --offset <offset>

Check offset value of all partitions in a consumer group (position of all consumers in a consumer group)

KAFKA_HOME/bin/kafka-consumer-groups.sh --bootstrap-server <kafka_broker_ip>:9092 --group <Consumer_group_name> --describe

Get all consumer groups across all topics

KAFKA_HOME/bin/kafka-consumer-groups.sh --bootstrap-server <kafka_broker_ip>:9092 --list

Dry run command for setting commit offset value in a partition

KAFKA_HOME/bin/kafka-consumer-groups.sh --bootstrap-server <kafka_broker_ip>:9092 --group <Consumer_group_name> --topic <topic_name>:<partition_number> --reset-offsets --to-offset <offset_number> --dry-run

Execute command for setting commit offset value in a partition

KAFKA_HOME/bin/kafka-consumer-groups.sh --bootstrap-server <kafka_broker_ip>:9092 --group <Consumer_group_name> --topic <topic_name>:<partition_number> --reset-offsets --to-offset <offset_number> --execute

Create kafka topic with partition=3, replica=2 and packet retention time=14 days

KAFKA_HOME/bin/kafka-topics.sh --create --zookeeper <zookeeper_ip>:2181 --replication-factor 2 --partitions 3 --topic <topic_name> --config retention.ms=1209600000

List topic names

KAFKA_HOME/bin/kafka-topics.sh --list --zookeeper <zookeeper_ip>:2181

Describe topic

KAFKA_HOME/bin/kafka-topics.sh --zookeeper <zookeeper_ip>:2181 --describe --topic <topic_name>

Change topic retention time

KAFKA_HOME/bin/kafka-topics.sh --zookeeper <zookeper_ip>:2181 --alter --topic <topic_name> --config retention.ms=<time_in_ms>

Get end offset of a topic for each partition

cd KAFKA_HOME/bin

./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list <broker_ip>:9092 --topic <topic_name>  --time -1

Get start offset of a topic for each partition

cd KAFKA_HOME/bin

./kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list <broker_ip>:9092 --topic <topic_name>  --time -2

I hope you find these kafka commands very helpful. Post your comments for any improvements. You can also post comments if you find any other helpful kafka command.

Thanks