Purge Kafka Topic

Question

Is there a way to purge the topic in kafka   I pushed a message that was too big into a kafka message topic on my local machine  now I m getting an error   kafka common InvalidMessageSizeException  invalid message size   Increasing the fetch size is not ideal here  because I don t actually want to accept messages that big

User · Answer

While the accepted answer is correct  that method has been deprecated   Topic configuration should now be done via kafka-configs   kafka-configs --zookeeper localhost 2181 --entity-type topics --alter --add-config retention ms 1000 --entity-name MyTopic   Configurations set via this method can be displayed with the command  kafka-configs --zookeeper localhost 2181 --entity-type topics --describe --entity-name MyTopic

User · Answer

UPDATE  This answer is relevant for Kafka 0 6  For Kafka 0 8 and later see answer by  Patrick   Yes  stop kafka and manually delete all files from corresponding subdirectory  it s easy to find it in kafka data directory   After kafka restart the topic will be empty

User · Answer

Thomas  advice is great but unfortunately zkCli in old versions of  Zookeeper  for example 3 3 6  do not seem to support rmr  For example compare the command line implementation in modern Zookeeper with version 3 3   If you are faced with an old version of Zookeeper one solution is to use a client library such as zc zk for Python  For people not familiar with Python you need to install it using pip or easy install  Then start a Python shell  python  and you can do   import zc zk zk   zc zk ZooKeeper  localhost 2181   zk delete recursive  brokers MyTopic      or even  zk delete recursive  brokers     if you want to remove all the topics from Kafka

User · Answer

Another  rather manual  approach for purging a topic is   in the brokers    stop kafka broker sudo service kafka stop delete all partition log files  should be done on all brokers  sudo rm -R  kafka-storage kafka-logs  lt some topic name gt -    in zookeeper    run zookeeper command line interface sudo  usr lib zookeeper bin zkCli sh use zkCli to remove the topic metadata rmr  brokers topic  lt some topic name gt    in the brokers again    restart broker service sudo service kafka start

User · Answer

have you considered having your app simply use a new renamed topic   i e  a topic that is named like the original topic but with a  quot 1 quot  appended at the end   That would also give your app a fresh clean topic

User · Answer

Sometimes  if you ve a saturated cluster  too many partitions  or using encrypted topic data  or using SSL  or the controller is on a bad node  or the connection is flaky  it ll take a long time to purge said topic    I follow these steps  particularly if you re using Avro    1  Run with kafka tools     bash kafka-configs sh --alter --entity-type topics --zookeeper zookeeper01 kafka com --add-config retention ms 1 --entity-name  lt topic-name gt    2  Run on Schema registry node   kafka-avro-console-consumer --consumer-property security protocol SSL --consumer-property ssl truststore location  etc schema-registry secrets trust jks --consumer-property ssl truststore password password --consumer-property ssl keystore location  etc schema-registry secrets identity jks --consumer-property ssl keystore password password --consumer-property ssl key password password --bootstrap-server broker01 kafka com 9092 --topic  lt topic-name gt  --new-consumer --from-beginning  3  Set topic retention back to the original setting  once topic is empty    bash kafka-configs sh --alter --entity-type topics --zookeeper zookeeper01 kafka com --add-config retention ms 604800000 --entity-name  lt topic-name gt    Hope this helps someone  as it isn t easily advertised

User · Answer

Here are the steps I follow to delete a topic named MyTopic    Describe the topic  and take not of the broker ids Stop the Apache Kafka daemon for each broker ID listed   Connect to each broker  and delete the topic data folder  e g  rm -rf  tmp kafka-logs MyTopic-0  Repeat for other partitions  and all replicas Delete the topic metadata  zkCli sh then rmr  brokers MyTopic Start the Apache Kafka daemon for each stopped machine     If you miss you step 3  then Apache Kafka will continue to report the topic as present  for example when if you run kafka-list-topic sh    Tested with Apache Kafka 0 8 0

User · Answer

From Java  using the new AdminZkClient instead of the deprecated AdminUtils     public void reset         try  KafkaZkClient zkClient   KafkaZkClient apply  localhost 2181   false  200 000          5000  10  Time SYSTEM   metricGroup    metricType             for  Map Entry lt String  List lt PartitionInfo gt  gt  entry   listTopics   entrySet              deleteTopic entry getKey    zkClient                        private void deleteTopic String topic  KafkaZkClient zkClient            skip Kafka internal topic     if  topic startsWith                return             System out println  Resetting Topic      topic       AdminZkClient adminZkClient   new AdminZkClient zkClient       adminZkClient deleteTopic topic           deletions are not instantaneous     boolean success   false      int maxMs   5 000      while  maxMs  gt  0  amp  amp   success          try           maxMs -  100          adminZkClient createTopic topic  1  1  new Properties    null           success   true          catch  TopicExistsException ignored                       if   success          Assert fail  failed to create     topic                private Map lt String  List lt PartitionInfo gt  gt  listTopics         Properties props   new Properties        props put  bootstrap servers   kafkaContainer getBootstrapServers         props put  group id    test-container-consumer-group        props put  key deserializer    org apache kafka common serialization StringDeserializer        props put  value deserializer    org apache kafka common serialization StringDeserializer         KafkaConsumer lt String  String gt  consumer   new KafkaConsumer lt  gt  props       Map lt String  List lt PartitionInfo gt  gt  topics   consumer listTopics        consumer close         return topics

User · Answer

kafka don t have direct method for purge clean-up topic  Queues   but can do this via deleting that topic and recreate it   first of make sure sever properties file has and if not add delete topic enable true  then  Delete topic  bin kafka-topics sh --zookeeper localhost 2181 --delete --topic myTopic  then create it again   bin kafka-topics sh --zookeeper localhost 2181 --create --topic myTopic --partitions 10 --replication-factor 2

User · Answer

Following command can be used to delete all the existing messages in kafka topic  kafka-delete-records --bootstrap-server  lt kafka server port gt  --offset-json-file delete json  The structure of the delete json file should be following      quot partitions quot        quot topic quot    quot foo quot    quot partition quot   1   quot offset quot   -1       quot version quot   1    where offset  -1 will delete all the records  This command has been tested with kafka 2 0 1

User · Answer

Could not add as comment because of size  Not sure if this is true  besides updating retention ms and retention bytes  but I noticed topic cleanup policy should be  delete   default   if  compact   it is going to hold on to messages longer  i e   if it is  compact   you have to specify delete retention ms also     bin kafka-configs sh --zookeeper localhost 2181 --describe --entity-name test-topic-3-100 --entity-type topics Configs for topics test-topic-3-100 are retention ms 1000 delete retention ms 10000 cleanup policy delete retention bytes 1  Also had to monitor earliest latest offsets should be same to confirm this successfully happened  can also check the du -h  tmp kafka-logs test-topic-3-100-     bin kafka-run-class sh kafka tools GetOffsetShell --broker-list  BROKER 9095  --topic test-topic-3-100 --time -1   awk -F       sum     3  END  print sum   26599762    bin kafka-run-class sh kafka tools GetOffsetShell --broker-list  BROKER 9095  --topic test-topic-3-100 --time -2   awk -F       sum     3  END  print sum   26599762  The other problem is  you have to get current config first so you remember to revert after deletion is successful    bin kafka-configs sh --zookeeper localhost 2181 --describe --entity-name test-topic-3-100 --entity-type topics

User · Answer

The simplest approach is to set the date of the individual log files to be older than the retention period  Then the broker should clean them up and remove them for you within a few seconds  This offers several advantages     No need to bring down brokers  it s a runtime operation   Avoids the possibility of invalid offset exceptions  more on that below     In my experience with Kafka 0 7 x  removing the log files and restarting the broker could lead to invalid offset exceptions for certain consumers  This would happen because the broker restarts the offsets at zero  in the absence of any existing log files   and a consumer that was previously consuming from the topic would reconnect to request a specific  once valid  offset  If this offset happens to fall outside the bounds of the new topic logs  then no harm and the consumer resumes at either the beginning or the end  But  if the offset falls within the bounds of the new topic logs  the broker attempts to fetch the message set but fails because the offset doesn t align to an actual message   This could be mitigated by also clearing the consumer offsets in zookeeper for that topic  But if you don t need a virgin topic and just want to remove the existing contents  then simply  touch -ing a few topic logs is far easier and more reliable  than stopping brokers  deleting topic logs  and clearing certain zookeeper nodes

User · Answer

To clean up all the messages from a particular topic using your application group  GroupName should be same as application kafka group name      kafka-path bin kafka-console-consumer sh --zookeeper localhost 2181 --topic topicName  --from-beginning --group application-group

User · Answer

kafka-topics sh --describe --zookeeper zkHost 2181 --topic myTopic   This should give retention ms configured  Then you can use above alter command to change to 1second  and later revert back to default    Topic myTopic   PartitionCount 6        ReplicationFactor 1     Configs retention ms 86400000

User · Answer

From kafka 1 1  Purge a topic  bin kafka-configs sh --zookeeper localhost 2181 --alter --entity-type topics --entity-name tp binance kline --add-config retention ms 100   wait 1 minute  to be secure that kafka purge the topic remove the configuration  and then go to default value  bin kafka-configs sh --zookeeper localhost 2181 --alter --entity-type topics --entity-name tp binance kline --delete-config retention ms

User · Answer

Following  steven appleyard answer I executed the following commands on Kafka 2 2 0 and they worked for me   bin kafka-configs sh --zookeeper localhost 2181 --entity-type topics --entity-name  lt topic-name gt  --describe  bin kafka-configs sh --zookeeper localhost 2181 --entity-type topics --entity-name  lt topic-name gt  --alter --add-config retention ms 1000  bin kafka-configs sh --zookeeper localhost 2181 --entity-type topics --entity-name  lt topic-name gt  --alter --delete-config retention ms

User · Answer

Tested in Kafka 0 8 2  for the quick-start example  First  Add one line to server properties file under config folder   delete topic enable true   then  you can run this command   bin kafka-topics sh --zookeeper localhost 2181 --delete --topic test

User · Answer

A lot of great answers over here but among them  I didn t find one about docker  I spent some time to figure out that using the broker container is wrong for this case  obviously         this is wrong  docker exec broker1 kafka-topics --zookeeper localhost 2181 --alter --topic mytopic --config retention ms 1000   Exception in thread  main  kafka zookeeper ZooKeeperClientTimeoutException  Timed out waiting for connection while in state  CONNECTING         at kafka zookeeper ZooKeeperClient  anonfun waitUntilConnected 3 ZooKeeperClient scala 258          at scala runtime java8 JFunction0 mcV sp apply JFunction0 mcV sp java 23          at kafka utils CoreUtils  inLock CoreUtils scala 253          at kafka zookeeper ZooKeeperClient waitUntilConnected ZooKeeperClient scala 254          at kafka zookeeper ZooKeeperClient  lt init gt  ZooKeeperClient scala 112          at kafka zk KafkaZkClient  apply KafkaZkClient scala 1826          at kafka admin TopicCommand ZookeeperTopicService  apply TopicCommand scala 280          at kafka admin TopicCommand  main TopicCommand scala 53          at kafka admin TopicCommand main TopicCommand scala    and I should have used zookeeper 2181 instead of --zookeeper localhost 2181 as per my compose file     this might be an option  but as per comment below not all zookeeper images can have this script included docker exec zookeper1 kafka-topics --zookeeper localhost 2181 --alter --topic mytopic --config retention ms 1000   the correct command would be   docker exec broker1 kafka-configs --zookeeper zookeeper 2181 --alter --entity-type topics --entity-name dev gdn urls --add-config retention ms 12800000   Hope it will save someone s time   Also  be aware that the messages won t be deleted immediately and it will happen when the segment of the log will be closed

User · Answer

To purge the queue you can delete the topic   bin kafka-topics sh --zookeeper localhost 2181 --delete --topic test   then re-create it   bin kafka-topics sh --create --zookeeper localhost 2181       --replication-factor 1 --partitions 1 --topic test

User · Answer

Temporarily update the retention time on the topic to one second   kafka-topics sh --zookeeper  lt zkhost gt  2181 --alter --topic  lt topic name gt  --config retention ms 1000   And in newer Kafka releases  you can also do it with kafka-configs --entity-type topics  kafka-configs sh --zookeeper  lt zkhost gt  2181 --entity-type topics --alter --entity-name  lt topic name gt  --add-config retention ms 1000   then wait for the purge to take effect  about one minute   Once purged  restore the previous retention ms value

[apache-kafka] Purge Kafka Topic

Examples related to apache-kafka

Examples related to purge