Kafka_zookeeper_commands

To list the topics

/opt/kafka/bin/kafka-topics.sh --list --zookeeper localhost:2181

To run the console consumer

/opt/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic <topic name> /opt/kafka/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic ntpprod.learning.events.failed --from-beginning

To get the topics offset

/opt/kafka/bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic <topic name> --time -1

To describe the topic

/opt/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic <topic name>

To alter the topic or To increase the partion

/opt/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic <topic name> --partitions 6

To reassign the partion or increase the replication of topics

/opt/kafka/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/sample.json --execute /opt/kafka/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/sample.json --verify

To consume the events from specific offset

/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --partition <partition number> --offset <offset number> --topic <topic name>

Load the data to specific topic

/opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic <topic name> < pipeline_test_data.json

To describe the consumer group

/opt/kafka/bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --describe --group <group name>

TO update the retention hours

/opt/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic <topic name> --config retention.ms=<>

TO update the topic max.message.bytes setting

/opt/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic <topic name> --config max.message.bytes=<>

To delete the topics from the zookeeper level forcefully

rmr /brokers/topics/{topic_name}
rmr /admin/delete_topics/{topic_name}
OR
sudo ./zookeeper-shell.sh localhost:2181 rmr /brokers/topics/your_topic

To update the retention period

/opt/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic <topicname> --config retention.ms=0

To increase the replication factor for topic partition or partition reassignment

NOTE: These steps can be used when we are scaling the Kafka cluster and want to reassign the partitions to the new node

• Describe the topic for which you want to increase the replication factor and check the replication factor count by using the below command.

  • /opt/kafka/bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic <topic name>

• Keep the below JSON in one file(This file will be used in next step) and fill out the topic name, partition number and replicas based on requirement.

{
  "version": 1, 
  "partitions":[
     {"topic":"ntpprod.telemetry.ingest","partition":0,"replicas":[3,1]},
     {"topic":"ntpprod.telemetry.ingest","partition":1,"replicas":[1,2]},
     {"topic":"ntpprod.telemetry.ingest","partition":2,"replicas":[2,3]},
     {"topic":"ntpprod.telemetry.ingest","partition":3,"replicas":[3,1]}
  ]
}

• Execute the below command in one of the kafka broker to start the partition reassignment.

  • /opt/kafka/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file <json-file-name> --execute

• Check the status of partition reassignment by using below command.

  • /opt/kafka/bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file <json-file-name> --verify

• You will see the output of above command as below if the partition reassignment is completed successfully.

Status of partition reassignment: 
Reassignment of partition env.telemetry.ingest-0 completed successfully
Reassignment of partition env.telemetry.ingest-1 completed successfully
Reassignment of partition env.telemetry.ingest-2 completed successfully
Reassignment of partition env.telemetry.ingest-3 completed successfully