How can I delete all the messages from a Kafka topic

Question

Announcement! Career Guide 2019 is out now. Explore careers to become a Big Data Developer or Architect!

I created a Kafka topic and pushed large messages to that kafka topic. Now I am getting error saying:

kafka.common.InvalidMessageSizeException: invalid message size

How I want to purge the topic so, that I can set the fetch.size for the topic ?

Comments

Please tell me how to answer an existing topic?  
 Maybe I'm not writing correctly?
 Need your help.
Thank you.

---

How to reply to a topic?  
 Maybe I'm not writing correctly?
 Please tell me.  
 Yours faithfully.

---

Can someone help how to answer a new topic ??
What am I doing wrong?  
 Please tell me.  
Thank you.

---

How to reply to a topic?  
 Maybe I'm doing something wrong?  
 Need your help.
Thank you.

---

Please tell me how to answer an existing topic?  
What am I doing wrong?  
 Need your help.
 Yours faithfully.

Answer 1

To purge the Kafka topic, you need to change the retention time of that topic. The default retention time is 168 hours, i.e. 7 days. So, you have to change the retention time to 1 second, after which the messages from the topic will be deleted. Then, you can go ahead and change the retention time of the topic back to 168 hours.

To change the retention of the topic:

kafka-topics.sh --zookeeper localhost:2181 --alter --topic topic1 --config retention.ms=1000

Comments

How do we verify that the topic is deleted?
Below command still shows the topic.
./kafka-topics.sh --zookeeper localhost:2181 --list | grep <topic_name>

---

Hi@Harshi,

You need to remove the log dir of the topic. Check the below-given answers. You will get the steps.

Answer 2

To delete manually:

1. Shutdown the cluster
2. Clean kafka log dir (specified by the log.dir attribute in kafka config file ) as well the zookeeper data
3. Restart the cluster

Comments

Not a good way

How we know that which topic consumed which not and only consumed should be deleted ?

is there any way to delete topics after consume?

Answer 3

Follow these steps:

1. Stop kafka
2. Clean kafka log specific to partition, kafka stores its log file in a format of "logDir/topic-partition" so for a topic named "MyTopic" the log for partition id 0 will be stored in /tmp/kafka-logs/MyTopic-0 where /tmp/kafka-logs is specified by the log.dirattribute
3. Restart kafka

Answer 4

Add one line to server.properties file under config folder:

delete.topic.enable=true

then, run this command:

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

Answer 5

This script will empty a topic:

#!/bin/bash
echo "Enter name of topic to empty:"
read topicName
/$Kafka_Home/bin/kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name $topicName --add config retention.ms=1000
sleep 5
/$Kafka_Home/bin/kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name $topicName --delete-config retention.ms

Comments

it is --add-config instead of --add config

---

Doesnt it matter how many messages do we have in the topic? Will this still work for a topic with hundred millions of messages? Should we still check whether there are still messages in the topic after changing the retention?

Answer 6

Try this:

• stop kafka
• delete the files rm -rf /tmp/kafka-logs/newTopic-*

Answer 7

One of the possible reason for this error could be your consumer's fetch size is smaller than the largest message it is trying to consume. Try increasing your fetch.size to make it larger than the max.message.size on your producers

Answer 8

Adding to @Bilal's answer, there could be another possible reason for this error, the log on the server could be corrupted. To verify this, try running the
DumpLogSegments tool on the log segments for that topic/partition and see if it reports a problem

Answer 9

sh kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic yourtopic --zookeeper localhost:2181

Answer 10

What I did was change the per-topic run-tine retention:

bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my_topic --config retention.bytes=1

Comments

Follow these steps:

1. Stop kafka
2. Clean kafka log specific to partition, kafka stores its log file in a format of "logDir/topic-partition" so for a topic named "MyTopic" the log for partition id 0 will be stored in /tmp/kafka-logs/MyTopic-0 where /tmp/kafka-logs is specified by the log.dirattribute
3. Restart kafka

Answer 11

I dont know.

Answer 12

As of kafka 2.3.0 version, there is an alternate way to soft deletion of Kafka (old approach are deprecated ).

Update retention.ms to 1 sec (1000ms) then set it again after a min, to default setting i.e 7 days (168 hours, 604,800,000 in ms )

Soft deletion:- (rentention.ms=1000) (using kafka-configs.sh).

bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --alter --entity-name kafka_topic3p3r --entity-type topics  --add-config retention.ms=1000
Completed Updating config for entity: topic 'kafka_topic3p3r'.

Setting to default:- 7 days (168 hours , retention.ms= 604800000)

bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --alter --entity-name kafka_topic3p3r --entity-type topics  --add-config retention.ms=604800000

Answer 13

To delete all the messages from a Kafka topic. 

There are many approach : - 

Approach 1: Using Retention Time ( We can also say Soft Deletion without interruption. ) By Default Retention Time is : 7 Days

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-name kafka_keyword --entity-type topics  --add-config retention.ms=1000.  [ Here Identified for 1 Sec only ] 1 second = .001 ms

Approach 2 : Soft Delete ( Just Delete Topic and create new one with same name)

bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic keyword

Approach 3: Hard Delete 

- Stop the server

- Clean kafka log dir (specified by the log.dir attribute in kafka config file ) as well the zookeeper data

- Restart the broker

I hope it is enough to get idea. 

Answer 14

Hi,

You can delete the messages by adding one line to the server.properties file under config folder.

delete.topic.enable=true

After that, you can run the below command.

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