Working with Kafka

Kafka CLI commands

Create a topic

kafka-topics.sh --zookeeper 127.0.0.1:2181 --topic first_topic --create --partitions 3 --replication-factor 1

List topics

kafka-topics.sh --zookeeper 127.0.0.1:2181 --list

➜  kafka_2.12-2.5.0 kafka-topics.sh --zookeeper 127.0.0.1:2181 --topic first_topic --describe

Topic: first_topic PartitionCount: 3 ReplicationFactor: 1 Configs: 

Topic: first_topic Partition: 0 Leader: 0 Replicas: 0 Isr: 0

Topic: first_topic Partition: 1 Leader: 0 Replicas: 0 Isr: 0

Topic: first_topic Partition: 2 Leader: 0 Replicas: 0 Isr: 0

Delete topic

kafka-topics.sh --zookeeper 127.0.0.1:2181 --topic first_topic --delete

kafka-console-producer

➜  kafka_2.12-2.5.0 kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic first_topic

>hello prashanth

>learning kafkA

>:-)

>^C%    

kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic first_topic --producer-property acks=all

Not recommended

kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic new_topic

Creates the new_topic, with default partitions configured in server.properties

kafka-console-consumer

Reads messages as the producer puts the message

kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic first_topic

Reads the messages from the beginning

kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic first_topic --from-beginning

Kafka consumers in group

kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic first_topic --group my-ap

kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic first_topic --group my-app2 --from-beginning

Will read all the messages from beginning, as it's commits the offset for this group when you run the same command, it doesn't return anything.

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

 kafka_2.12-2.5.0 kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --describe --group  my-app2

Consumer group 'my-app2' has no active members.

GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID     HOST            CLIENT-ID

my-app2         first_topic     0          7               7               0               -               -               -

my-app2         first_topic     1          8               8               0               -               -               -

my-app2         first_topic     2          8               8               0               -               -   

Reset the offsets for a consumer group

➜  kafka_2.12-2.5.0 kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group my-app --reset-offsets --to-earliest --execute --topic first_topic

GROUP                          TOPIC                          PARTITION  NEW-OFFSET     

my-app                         first_topic                    0          0              

my-app                         first_topic                    1          0              

my-app                         first_topic                    2          0  

➜  kafka_2.12-2.5.0 kafka-consumer-groups.sh --bootstrap-server 127.0.0.1:9092 --group my-app --reset-offsets --shift-by -2 --execute --topic first_topic

GROUP                          TOPIC                          PARTITION  NEW-OFFSET     

my-app                         first_topic                    0          5              

my-app                         first_topic                    1          6              

my-app                         first_topic                    2          6     

Kafka Streams