kafka command

kafka_command.sh

Kafka shell command:

1) Get list all topic:
./kafka-topics.sh --zookeeper <zookeeper_host:port> --list
./bin/kafka-topics.sh --zookeeper kafka1:2181 --list
./bin/kafka-topics.sh --zookeeper kafka1:2181 --describe

2) Create topic:
./kafka-topics.sh --create --zookeeper <zookeeper_host:port> --replication-factor <number_replica> --partitions <number_partition> --topic <topic_name>
./bin/kafka-topics.sh --create --zookeeper kafka1:2181 --replication-factor 3 --partitions 8 --topic proximity

3) Delete topic:
./kafka-topics.sh --delete --zookeeper <zookeeper_host:port> --topic <topic_name>
./kafka-topics.sh --delete --zookeeper kafka1:2181 --topic test-topic
./bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --topic test-topic --delete --group <consumer_group_name>
./bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --delete --group mobio-profiling-pv

3.1) Read topic:
./bin/kafka-topics.sh --zookeeper kafka1:2181 --describe --topic <topic_name>

4) Read data from kafka topic and write output to console:
*) Lệnh cũ, bản mới ko chạy được lệnh này
./kafka-console-consumer.sh --zookeeper <zookeeper_host:port> --topic <topic_name>
./kafka-console-consumer.sh --zookeeper kafka1:2181 --topic proximity
*) Lệnh mới
./kafka-console-consumer.sh --bootstrap-server <kafka_host:port> --topic <topic_name> --from-beginning
./kafka-console-consumer.sh --bootstrap-server kafka1:19091 --topic url_record --from-beginning

5) Read data from standard output and write it to a Kafka topic:
./kafka-console-producer.sh --broker-list <kafka_host:port>,<kafka_host:port> --topic <topic_name>
./kafka-console-producer.sh --broker-list kafka1:9092,kakfa2:9092 --topic proximity

6) View offsets of consumer group:
*) For old consumers that store offsets in ZooKeeper
./kafka-consumer-groups.sh --zookeeper <zookeeper_host:port> --describe --group <group_id>
./kafka-consumer-groups.sh --zookeeper kafka1:2181 --describe --group tong-hop-session
*) For new consumers that store offsets in Kafka
./kafka-consumer-groups.sh --bootstrap-server <kafka_host:port> --group <group_id> --describe
./kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group tong-hop-session --describe

7) Reset the consumer offset for a topic:
*) Preview offsets after reset (but not actually reset)
./kafka-consumer-groups.sh --bootstrap-server <kafka_host:port> --group <group_id> --reset-offsets --to-latest --topic <topic_name>
./kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group tong-hop-session --reset-offsets --to-earliest --topic location2_site
*) Reset offsets
./kafka-consumer-groups.sh --bootstrap-server <kafka_host:port> --group <group_id> --reset-offsets --to-earliest --topic <topic_name> --execute
./kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group tong-hop-session --reset-offsets --to-earliest --topic location2_site --execute

8) Reset the consumer offset for all topic:
*) Preview offsets after reset (but not actually reset)
./kafka-consumer-groups.sh --bootstrap-server <kafka_host:port> --group <group_id> --reset-offsets --to-earliest --all-topics
./kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group tong-hop-session --reset-offsets --to-earliest --all-topics
*) Reset offsets
./kafka-consumer-groups.sh --bootstrap-server <kafka_host:port> --group <group_id> --reset-offsets --to-earliest --all-topics --execute
./kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --group tong-hop-session --reset-offsets --to-earliest --all-topics --execute

9) Adding Partitions to a Topic
./bin/kafka-topics.sh --alter --zookeeper kafka1:2181 --topic nm_rollback_email --partitions 16

10) View all Groups
./bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --list

11) Rereassignment Factor 

{"version":1,
"partitions":[{"topic":"chattool-send-socket-online-nm","partition":0,"replicas":[10,20,30]}]}

Then, use the json file with the --execute option to start the reassignment process: 
./bin/kafka-reassign-partitions.sh --zookeeper kafka1:2181 --reassignment-json-file increase-replication-factor.json --execute

Save this to use as the --reassignment-json-file option during rollback
Successfully started reassignment of partitions
{"version":1,
"partitions":[{"topic":"foo","partition":0,"replicas":[5,6,7]}]}

The --verify option can be used with the tool to check the status of the partition reassignment. Note that the same increase-replication-factor.json 

./bin/kafka-reassign-partitions.sh --zookeeper kafka1:2181 --reassignment-json-file increase-replication-factor.json --verify

12) Send Message to topic
./bin/kafka-console-producer.sh --topic upsert-profile-with-callback --bootstrap-server localhost:9092


./bin/kafka-topics.sh --alter --zookeeper kafka1:2181 --topic process-profile-change-history --partitions 16 

13) change retention

./bin/kafka-topics.sh --zookeeper kafka1:2181 --alter --topic process-profile-change-history --config retention.ms=259200000
./bin/kafka-configs.sh --bootstrap-server kafka1:9092 --entity-type topics --entity-name adp-logging-system2 --alter --add-config retention.ms=259200000

./bin/zookeeper-shell.sh kafka1:2181 ls /brokers/ids

echo srvr | nc kafka2 2181

14) list all groups and topics belong
./bin/kafka-consumer-groups.sh --bootstrap-server kafka1:9092 --all-groups --describe