Purger le sujet Kafka

Existe-t-il un moyen de purger le sujet dans kafka?

J'ai poussé un message trop gros dans un sujet de message kafka sur ma machine locale, maintenant j'obtiens une erreur:

kafka.common.InvalidMessageSizeException: invalid message size

Augmenter le fetch.sizen'est pas idéal ici, car je ne veux pas vraiment accepter de messages aussi gros.

Mettez temporairement à jour le temps de rétention du sujet sur une seconde:

kafka-topics.sh --zookeeper <zkhost>:2181 --alter --topic <topic name> --config retention.ms=1000

Et dans les nouvelles versions de Kafka, vous pouvez également le faire avec kafka-configs --entity-type topics

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

puis attendez que la purge prenne effet (environ une minute). Une fois purgé, restaurez la retention.msvaleur précédente .

Pour purger la file d'attente, vous pouvez supprimer le sujet:

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

puis recréez-le:

bin/kafka-topics.sh --create --zookeeper localhost:2181 \
    --replication-factor 1 --partitions 1 --topic test

Voici les étapes que je suis pour supprimer un sujet nommé MyTopic:

1. Décrivez le sujet et ne prenez pas les identifiants du courtier
2. Arrêtez le démon Apache Kafka pour chaque ID de courtier répertorié.
3. Connectez-vous à chaque courtier et supprimez le dossier de données de rubrique, par exemple rm -rf /tmp/kafka-logs/MyTopic-0. Répétez pour les autres partitions et toutes les répliques
4. Supprimer les métadonnées du sujet: zkCli.shpuisrmr /brokers/MyTopic
5. Démarrez le démon Apache Kafka pour chaque machine arrêtée

Si vous manquez l'étape 3, Apache Kafka continuera à signaler le sujet comme présent (par exemple, si vous exécutez kafka-list-topic.sh).

Testé avec Apache Kafka 0.8.0.

Bien que la réponse acceptée soit correcte, cette méthode est obsolète. La configuration du sujet doit maintenant être effectuée via kafka-configs.

kafka-configs --zookeeper localhost:2181 --entity-type topics --alter --add-config retention.ms=1000 --entity-name MyTopic

Les configurations définies via cette méthode peuvent être affichées avec la commande

kafka-configs --zookeeper localhost:2181 --entity-type topics --describe --entity-name MyTopic

Testé dans Kafka 0.8.2, pour l'exemple de démarrage rapide: Tout d'abord, ajoutez une ligne au fichier server.properties sous le dossier config:

delete.topic.enable=true

ensuite, vous pouvez exécuter cette commande:

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

Depuis 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 1 minute, pour être sûr que kafka purge le sujet, supprimez la configuration, puis passez à 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

kafka n'a pas de méthode directe pour purger / nettoyer le sujet (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 contient et sinon ajoutez delete.topic.enable=true

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

puis créez-le à nouveau.

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

Parfois, si vous avez un cluster saturé (trop de partitions, ou en utilisant des données de sujet chiffrées, ou en utilisant SSL, ou le contrôleur est sur un mauvais nœud, ou la connexion est irrégulière, il faudra beaucoup de temps pour purger ledit sujet .

Je suis ces étapes, en particulier si vous utilisez Avro.

1: Exécutez avec les outils kafka:

bash kafka-configs.sh --alter --entity-type topics --zookeeper zookeeper01.kafka.com --add-config retention.ms=1 --entity-name <topic-name>

2: Exécuter sur le nœud de registre Schema:

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 <topic-name> --new-consumer --from-beginning

3: redéfinissez la rétention des sujets sur le paramètre d'origine, une fois que le sujet est vide.

bash kafka-configs.sh --alter --entity-type topics --zookeeper zookeeper01.kafka.com --add-config retention.ms=604800000 --entity-name <topic-name>

J'espère que cela aide quelqu'un, car ce n'est pas facilement annoncé.

MISE À JOUR: Cette réponse est pertinente pour Kafka 0.6. Pour Kafka 0.8 et versions ultérieures, 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.

L'approche la plus simple consiste à définir la date des fichiers journaux individuels pour qu'elle soit antérieure à la période de rétention. Ensuite, le courtier doit les nettoyer et les supprimer pour vous en quelques secondes. Cela offre plusieurs avantages:

1. Pas besoin de faire tomber les courtiers, c'est une opération d'exécution.
2. Évite la possibilité d'exceptions de décalage non valides (plus d'informations ci-dessous).

D'après mon expérience avec Kafka 0.7.x, la suppression des fichiers journaux et le redémarrage du courtier pourraient entraîner des exceptions de décalage invalides pour certains consommateurs. Cela se produit parce que le courtier redémarre les décalages à zéro (en l'absence de tous les fichiers journaux existants), et un consommateur qui consommait auparavant à partir de la rubrique se reconnecterait pour demander un décalage [une fois valide] spécifique. Si ce décalage tombe en dehors des limites des nouveaux journaux de rubrique, alors aucun dommage et le consommateur reprend au début ou à la fin. Mais, si le décalage tombe dans les limites des nouveaux journaux de rubrique, le courtier tente d'extraire l'ensemble de messages mais échoue car le décalage ne s'aligne pas sur un message réel.

Cela pourrait être atténué en effaçant également les compensations des consommateurs dans zookeeper pour ce sujet. Mais si vous n'avez pas besoin d'un sujet vierge et que vous voulez simplement supprimer le contenu existant, il est beaucoup plus facile et plus fiable de simplement `` toucher '' quelques journaux de sujets, que d'arrêter les courtiers, de supprimer les journaux de sujets et d'effacer certains nœuds de gardien de zoo. .

Les conseils de Thomas sont excellents mais malheureusement zkClidans les anciennes versions de Zookeeper (par exemple 3.3.6) ne semblent pas le supporter rmr. Par exemple, comparez l'implémentation de la ligne de commande dans Zookeeper moderne avec la version 3.3 .

Si vous êtes confronté à une ancienne version de Zookeeper, une solution consiste à utiliser une bibliothèque cliente telle que zc.zk pour Python. Pour les personnes qui ne sont pas familières avec Python, vous devez l'installer à l'aide de pip ou easy_install . Ensuite, démarrez un shell Python ( python) et vous pouvez faire:

import zc.zk
zk = zc.zk.ZooKeeper('localhost:2181')
zk.delete_recursive('brokers/MyTopic') 

ou même

zk.delete_recursive('brokers')

si vous souhaitez supprimer tous les sujets de Kafka.

Pour nettoyer tous les messages d'une rubrique particulière à l'aide de votre groupe d'applications (GroupName doit être identique au nom du groupe kafka de l'application).

./kafka-path/bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic topicName --from-beginning --group application-group

Suite à la réponse @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

Beaucoup de bonnes réponses ici, mais parmi elles, je n'en ai pas trouvé sur docker. J'ai passé un certain temps à comprendre que l'utilisation du conteneur de courtier est incorrecte pour ce cas (évidemment !!!)

## 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.<init>(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)

et j'aurais dû utiliser zookeeper:2181au lieu de --zookeeper localhost:2181selon mon fichier de composition

## 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

la commande correcte serait

docker exec broker1 kafka-configs --zookeeper zookeeper:2181 --alter --entity-type topics --entity-name dev_gdn_urls --add-config retention.ms=12800000

J'espère que cela fera gagner du temps à quelqu'un.

Sachez également que les messages ne seront pas supprimés immédiatement et que cela se produira lorsque le segment du journal sera fermé.

Impossible d'ajouter en tant que commentaire en raison de la taille: Je ne sais pas si c'est vrai, en plus de la mise à jour de retention.ms et retention.bytes, mais j'ai remarqué que la politique de nettoyage de la rubrique devrait être "delete" (par défaut), si "compact", elle va conserver les messages plus longtemps, c'est-à-dire que s'il est "compact", vous devez également spécifier delete.retention.ms .

./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

Il fallait également surveiller les décalages les plus anciens / les plus récents pour confirmer que cela s'est produit avec succès, peut également vérifier le 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

L'autre problème est, vous devez obtenir config actuelle d' abord si vous vous souvenez de revenir après la suppression est réussie: ./bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-name test-topic-3-100 --entity-type topics

Une autre approche, plutôt manuelle, pour purger un sujet est:

chez les courtiers:

1. arrêter le courtier kafka
   sudo service kafka stop
2. supprimer tous les fichiers journaux de partition (devrait être fait sur tous les courtiers)
   sudo rm -R /kafka-storage/kafka-logs/<some_topic_name>-*

dans zookeeper:

3. exécuter l'interface de ligne de commande de zookeeper
   sudo /usr/lib/zookeeper/bin/zkCli.sh
4. utilisez zkCli pour supprimer les métadonnées du sujet
   rmr /brokers/topic/<some_topic_name>

dans les courtiers à nouveau:

5. redémarrer le service de courtier
   sudo service kafka start

./kafka-topics.sh --describe --zookeeper zkHost:2181 --topic myTopic

Cela devrait donner retention.msconfiguré. Ensuite, vous pouvez utiliser la commande alter ci-dessus pour passer à 1 seconde (et revenir plus tard à la valeur par défaut).

Topic:myTopic   PartitionCount:6        ReplicationFactor:1     Configs:retention.ms=86400000

Depuis Java, en utilisant le nouveau AdminZkClientau lieu du obsolète AdminUtils:

  public void reset() {
    try (KafkaZkClient zkClient = KafkaZkClient.apply("localhost:2181", false, 200_000,
        5000, 10, Time.SYSTEM, "metricGroup", "metricType")) {

      for (Map.Entry<String, List<PartitionInfo>> 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 > 0 && !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<String, List<PartitionInfo>> 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<String, String> consumer = new KafkaConsumer<>(props);
    Map<String, List<PartitionInfo>> topics = consumer.listTopics();
    consumer.close();

    return topics;
  }