Comment supprimer/nettoyer Kafka messages en file d'attente sans la suppression de Sujet

Est-il un moyen de supprimer les messages en file d'attente sans la suppression de Kafka sujets?

Je souhaite supprimer les messages en file d'attente lors de l'activation de la consommation.

Je sais qu'il ya plusieurs façons comme:

Réinitialisation temps de rétention

$ ./bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic MyTopic --config retention.ms=1000
La suppression de kafka fichiers

$ rm -rf /data/kafka-logs/<topic/Partition_name>

Le temps de rétention astuce vous mentionner d'abord est beaucoup mieux. La deuxième façon provoque des problèmes pour les répliqué sujets et rend les métadonnées pour le sujet non conformes à la réalité. Notez que le décalage ne vais pas revenir à zéro.

InformationsquelleAutor Tachikoma | 2017-09-14

apache-kafka

16

De 0,11 ou plus, vous pouvez exécuter l'bin/kafka-delete-records.sh commande pour marquer les messages à supprimer.

https://github.com/apache/kafka/blob/trunk/bin/kafka-delete-records.sh

Par exemple publier des 100 messages
```
seq 100 | ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic mytest
```
puis supprimer 90 de ces 100 messages avec la nouvelle kafka-delete-records.sh
outil de ligne de commande
```
./bin/kafka-delete-records.sh --bootstrap-server localhost:9092 --offset-json-file ./offsetfile.json
```
où offsetfile.json contient
```
 {"partitions": [{"topic": “mytest", "partition": 0, "offset": 90}], "version":1 }
```
et consomment les messages depuis le début pour vérifier que 90 des 100
les messages sont en effet marqués comme supprimés.
```
./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic mytest --from-beginning
91
92
93
94
95
96
97
98
99
100
```
- Merci Hans pour la réponse! Il est très proche de ce que j'ai envie. Savez-vous, par hasard, si je peux supprimer tous les offset sans savoir combien de messages sont placés? Serais-je capable de le faire en modifiant le json?
- Oui, vous pouvez supprimer tous les messages. Aussi, vous pouvez passer directement à l'aide de cet outil, et de regarder le code source et d'écrire votre propre programme qui appelle directement les même API pour supprimer des enregistrements avant tout la compensation (y compris la dernière offset) ou vous pouvez recherche de compensations par heure pour supprimer tous les enregistrements avant un certain temps. L'API utilise cet outil doit être mieux documentée dans Kafka 1.0 qui est sorti en octobre 2017
- J'ai trouvé que {"partitions": [{ "rubrique": "test", "partition": 0, "offset": -1 }], "version":1 } pouvez effacer tous les messages dans le sujet. Le prochain problème que j'ai, c'est que cela supprime la même rubrique d'un groupe différent. Par exemple, si la rubrique "mytest" a été dans cosumer groupe "groupe1" et "groupe2" a été à regarder la rubrique "mytest", l'outil va supprimer tous les messages dans les deux groupes. Est-il un moyen pour que cela cesse?
- Il y a seulement une validation partagée journal de Kafka donc, si vous supprimez les messages, ils sont partis pour tout le monde indépendamment de tout les groupes de consommateurs. Si vous souhaitez modifier l'un des groupes de consommateurs, des compensations à ignorer les messages, mais les laisser pour les autres consommateurs, puis utilisez bin/kafka-consumer-groups.sh --reset-compense. Voir KIP-122 détails ici cwiki.apache.org/confluence/display/KAFKA/...
InformationsquelleAutor Hans Jespersen

Pour supprimer tous les messages dans un sujet spécifique, vous pouvez exécuter kafka-delete-records.sh

Par exemple, j'ai un sujet appelé test, qui a 4 partitions.

Créer un Json fichier , par exemple j.json:

{

"partitions": [

    {

        "topic": "test",

        "partition": 0,

        "offset": -1

    }, {

        "topic": "test",

        "partition": 1,

        "offset": -1

    }, {

        "topic": "test",

        "partition": 2,

        "offset": -1

    }, {

        "topic": "test",

        "partition": 3,

        "offset": -1

    }

],

"version": 1

}

maintenant supprimer tous les messages à cette commande :

/opt/kafka/confluent-4.1.1/bin/kafdelete-records --bootstrap-server 192.168.XX.XX:9092 --offset-json-file j.json

Après l'exécution de la commande, ce message sera affiché

Records delete operation completed:
partition: test-0   low_watermark: 7
partition: test-1   low_watermark: 7
partition: test-2   low_watermark: 7
partition: test-3   low_watermark: 7

c'est vraiment lourd ..

InformationsquelleAutor Alihossein shahabi

Vous devez vous connecter pour publier un commentaire.