Y a-t-il un moyen de purger le sujet dans Kafka ?

Question

J'ai poussé un message qui était trop gros dans un sujet de message kafka sur ma machine locale, maintenant je reçois une erreur :

kafka.common.InvalidMessageSizeException: invalid message size

Augmenter le fetch.size n'est pas l'idéal ici, car je ne veux pas vraiment accepter des messages de cette taille.

Answer 1

La commande suivante peut être utilisée pour supprimer tous les messages existants dans le sujet kafka :

kafka-delete-records --bootstrap-server <kafka_server:port> --offset-json-file delete.json

La structure du fichier delete.json doit être la suivante :

{ "partitions" : [ { "sujet" : "foo", "partition" : 1, "offset" : -1 } ], "version" : 1 }

où offset :-1 supprimera tous les enregistrements (Cette commande a été testée avec kafka 2.0.1.

Answer 2

De kafka 1.1

Purger un sujet

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name tp_binance_kline --add-config retention.ms=100

attendez au moins 1 minute, pour être sûr que kafka purge le sujet supprimer la configuration, puis passer à la valeur par défaut

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name tp_binance_kline --delete-config retention.ms

Answer 3

Kafka n'a pas de méthode directe pour purger/nettoyer les sujets (files d'attente), mais peut le faire en supprimant ce sujet et en le recréant.

Tout d'abord, assurez-vous que le fichier sever.properties existe et si ce n'est pas le cas ajoutez delete.topic.enable=true

puis, Supprimer le sujet bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic myTopic

puis le créer à nouveau.

bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic myTopic --partitions 10 --replication-factor 2

Answer 4

Suite à la réponse de @steven appleyard, j'ai exécuté les commandes suivantes sur Kafka 2.2.0 et elles ont fonctionné pour moi.

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

bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name <topic-name> --alter --add-config retention.ms=1000

bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name <topic-name> --alter --delete-config retention.ms

Answer 5

MISE À JOUR : cette réponse est pertinente pour Kafka 0.6. Pour Kafka 0.8 et plus, voir la réponse de @Patrick.

Oui, arrêtez kafka et supprimez manuellement tous les fichiers du sous-répertoire correspondant (il est facile de le trouver dans le répertoire de données kafka). Après le redémarrage de kafka, le sujet sera vide.