Kafka-Thema löschen

184

Gibt es eine Möglichkeit, das Thema in Kafka zu löschen?

Ich habe eine zu große Nachricht in ein Kafka-Nachrichtenthema auf meinem lokalen Computer verschoben. Jetzt wird eine Fehlermeldung angezeigt:

kafka.common.InvalidMessageSizeException: invalid message size

Das Erhöhen der fetch.sizeist hier nicht ideal, weil ich eigentlich keine so großen Nachrichten annehmen möchte.

Peter Klipfel
quelle

Antworten:

359

Aktualisieren Sie die Aufbewahrungszeit für das Thema vorübergehend auf eine Sekunde:

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

Und in neueren Kafka-Versionen können Sie dies auch tun kafka-configs --entity-type topics

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

Warten Sie dann, bis die Spülung wirksam wird (ca. eine Minute). Stellen Sie nach dem Löschen den vorherigen retention.msWert wieder her.

Steven Appleyard
quelle
8
Das ist eine gute Antwort, aber können Sie bitte eine Beschreibung hinzufügen, wie Sie mit der Überprüfung des aktuellen Wertes für die Aufbewahrung des Themas beginnen können?
Greg Dubicki
28
Ich bin nicht sicher, ob ich die aktuelle Konfiguration überprüfen soll, aber ich glaube, dass das Zurücksetzen auf die Standardeinstellung wie folgt aussieht:bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic MyTopic --deleteConfig retention.ms
aspergillusOryzae
15
Oder je nach Version:--delete-config retention.ms
aspergillusOryzae
3
Nur zu Ihrer Information, für kafka v. 0.9.0.0 heißt es: ubuntu @ ip-172-31-21-201: /opt/kafka/kafka_2.10-0.9.0.0-SNAPSHOT$ bin / kafka-topics.sh - -zookeeper localhost: 2181 --alter --topic room-data --config Retention.ms = 1000 WARNUNG: Das Ändern der Themenkonfiguration in diesem Skript ist veraltet und wird möglicherweise in zukünftigen Versionen entfernt. In Zukunft verwenden Sie bitte kafka-configs.sh für diese Funktionalität
Alper Akture
54
Es scheint, dass seit 0.9.0 die Verwendung von kafka-topics.sh zum Ändern der Konfiguration veraltet ist. Die neue Option ist die Verwendung des Skripts kafka-configs.sh. e.g. kafka-configs.sh --zookeeper <zkhost>:2181 --alter --entity-type topics --entity-name <topic name> --add-config retention.ms=1000 Dies erlaubt Ihnen auch die aktuelle Aufbewahrungsfrist zu überprüfen, zB kafkas-configs --zookeeper <zkhost>: 2181 --describe --entity-Typ Themen --entity-name <topic name>
RHE
69

Um die Warteschlange zu löschen, können Sie das Thema löschen:

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

dann erstelle es neu:

bin/kafka-topics.sh --create --zookeeper localhost:2181 \
    --replication-factor 1 --partitions 1 --topic test
rjaiswal
quelle
14
Denken Sie daran, eine Zeile delete.topic.enable=truein die Datei einzufügen config/server.properties, wie in der Warnung des genannten Befehls angegebenNote: This will have no impact if delete.topic.enable is not set to true.
Patrizio Bertoni
3
Dies ist nicht immer augenblicklich. Manchmal wird es nur zum Löschen markiert und das tatsächliche Löschen erfolgt später.
Gaurav Khare
48

Hier sind die Schritte, die ich befolge, um ein Thema mit dem Namen zu löschen MyTopic:

  1. Beschreiben Sie das Thema und nehmen Sie nicht die Broker-IDs
  2. Stoppen Sie den Apache Kafka-Daemon für jede aufgelistete Broker-ID.
  3. Stellen Sie eine Verbindung zu jedem Broker her und löschen Sie den Themendatenordner, z rm -rf /tmp/kafka-logs/MyTopic-0. Wiederholen Sie diesen Vorgang für andere Partitionen und alle Replikate
  4. Löschen Sie die Themenmetadaten: zkCli.shdannrmr /brokers/MyTopic
  5. Starten Sie den Apache Kafka-Daemon für jeden gestoppten Computer

Wenn Sie Schritt 3 verpassen, meldet Apache Kafka das Thema weiterhin als vorhanden (z. B. wenn Sie es ausführen kafka-list-topic.sh).

Getestet mit Apache Kafka 0.8.0.

Thomas Bratt
quelle
2
in 0.8.1 ./zookeeper-shell.sh localhost:2181und./kafka-topics.sh --list --zookeeper localhost:2181
pdeschen
Kann zookeeper-clientanstelle von zkCli.sh(auf Cloudera CDH5 ausprobiert)
Martin Tapp
1
Dadurch wird das Thema gelöscht, nicht die darin enthaltenen Daten. Dies erfordert, dass der Broker gestoppt wird. Dies ist bestenfalls ein Hack. Steven Appleyards Antwort ist wirklich die absolut beste.
Jeff Maass
1
Dies war der einzige Weg zu der Zeit, als es geschrieben wurde.
Thomas Bratt
2
Arbeitete für mich an Kafka 0.8.2.1, obwohl die Topis in Zookeeper unter / brokers / topic / <
Themenname
43

Obwohl die akzeptierte Antwort korrekt ist, ist diese Methode veraltet. Die Themenkonfiguration sollte jetzt über erfolgen kafka-configs.

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

Mit dieser Methode festgelegte Konfigurationen können mit dem Befehl angezeigt werden

kafka-configs --zookeeper localhost:2181 --entity-type topics --describe --entity-name MyTopic
Shane Perry
quelle
2
Es lohnt sich auch hinzuzufügen:kafka-configs --zookeeper localhost:2181 --entity-type topics --alter --delete-config retention.ms --entity-name MyTopic
NoBrainer
38

In Kafka 0.8.2 für das Schnellstartbeispiel getestet: Fügen Sie zunächst eine Zeile zur Datei server.properties im Konfigurationsordner hinzu:

delete.topic.enable=true

Dann können Sie diesen Befehl ausführen:

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

Von kafka 1.1

Ein Thema löschen

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

Warten Sie 1 Minute, um sicherzugehen, dass kafka das Thema löscht, entfernen Sie die Konfiguration und wechseln Sie dann zum Standardwert

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name tp_binance_kline --delete-config retention.ms
user644265
quelle
1
Ich denke du hast einen zusätzlichen Pfeil. Auf meinem konnte ich laufenbin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name my-topic --add-config rentention.ms=100
Will
4

kafka hat keine direkte Methode zum Löschen / Bereinigen von Themen (Warteschlangen), kann dies jedoch tun, indem Sie dieses Thema löschen und neu erstellen.

Stellen Sie zunächst sicher, dass die Datei Sever.Properties vorhanden ist, und fügen Sie sie hinzu, falls nicht delete.topic.enable=true

dann Thema löschen bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic myTopic

dann erstelle es erneut.

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

Wenn Sie einen gesättigten Cluster haben (zu viele Partitionen oder verschlüsselte Themendaten oder SSL verwenden oder der Controller sich auf einem fehlerhaften Knoten befindet oder die Verbindung nicht funktioniert), dauert es manchmal lange, bis das Thema gelöscht ist .

Ich folge diesen Schritten, insbesondere wenn Sie Avro verwenden.

1: Mit Kafka-Tools ausführen:

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

2: Auf dem Schema-Registrierungsknoten ausführen:

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: Setzen Sie die Themenaufbewahrung auf die ursprüngliche Einstellung zurück, sobald das Thema leer ist.

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

Hoffe das hilft jemandem, da es nicht einfach zu bewerben ist.

Ben Coughlan
quelle
Hinweis: kafka-avro-console-consumerist nicht erforderlich
OneCricketeer
4

UPDATE: Diese Antwort ist relevant für Kafka 0.6. Für Kafka 0.8 und höher siehe Antwort von @Patrick.

Ja, stoppen Sie kafka und löschen Sie alle Dateien manuell aus dem entsprechenden Unterverzeichnis (es ist einfach, sie im kafka-Datenverzeichnis zu finden). Nach dem Neustart von kafka ist das Thema leer.

Lauffeuer
quelle
Dies erfordert das Herunterfahren des Brokers und ist bestenfalls ein Hack. Steven Appleyards Antwort ist wirklich die absolut beste.
Jeff Maass
@ MaasSql Ich stimme zu. :) Diese Antwort ist zwei Jahre alt, ungefähr Version 0.6. Die Funktionen "Thema ändern" und "Thema löschen" wurden später implementiert.
Wildfire
Steven Appleyards Antwort ist genauso hackig wie diese.
Banjocat
Es ist weitaus weniger schwierig, eine Anwendung dazu zu bringen, ihre eigenen Daten auf unterstützte Weise zu löschen, als diese Anwendung auszuschalten und alle Ihrer Meinung nach alle Datendateien zu löschen und dann wieder einzuschalten.
Nick
3

Am einfachsten ist es, das Datum der einzelnen Protokolldateien so festzulegen, dass es älter als die Aufbewahrungsdauer ist. Dann sollte der Broker sie bereinigen und innerhalb weniger Sekunden für Sie entfernen. Dies bietet mehrere Vorteile:

  1. Es ist nicht erforderlich, Broker herunterzufahren, es handelt sich um eine Laufzeitoperation.
  2. Vermeidet die Möglichkeit ungültiger Offset-Ausnahmen (mehr dazu weiter unten).

Nach meiner Erfahrung mit Kafka 0.7.x kann das Entfernen der Protokolldateien und das Neustarten des Brokers bei bestimmten Verbrauchern zu ungültigen Offset-Ausnahmen führen. Dies würde passieren, weil der Broker die Offsets bei Null neu startet (wenn keine vorhandenen Protokolldateien vorhanden sind) und ein Verbraucher, der zuvor das Thema verwendet hat, erneut eine Verbindung herstellt, um einen bestimmten [einmal gültigen] Offset anzufordern. Wenn dieser Versatz außerhalb der Grenzen der neuen Themenprotokolle liegt, ist dies kein Schaden, und der Verbraucher wird entweder am Anfang oder am Ende wieder aufgenommen. Wenn der Versatz jedoch innerhalb der Grenzen der neuen Themenprotokolle liegt, versucht der Broker, den Nachrichtensatz abzurufen, schlägt jedoch fehl, da der Versatz nicht an einer tatsächlichen Nachricht ausgerichtet ist.

Dies könnte gemildert werden, indem auch die Verbraucher-Offsets in zookeeper für dieses Thema gelöscht werden. Wenn Sie jedoch kein jungfräuliches Thema benötigen und nur den vorhandenen Inhalt entfernen möchten, ist das einfache Berühren einiger Themenprotokolle viel einfacher und zuverlässiger als das Stoppen von Brokern, das Löschen von Themenprotokollen und das Löschen bestimmter Zookeeper-Knoten .

Andrew Carter
quelle
Wie kann man "das Datum der einzelnen Protokolldateien so einstellen, dass es älter als die Aufbewahrungsfrist ist"? danke
bylijinnan
3

Der Rat von Thomas ist großartig, aber leider zkCliin alten Versionen von Zookeeper (zum Beispiel 3.3.6) nicht zu unterstützen rmr. Vergleichen Sie beispielsweise die Befehlszeilenimplementierung in Modern Zookeeper mit Version 3.3 .

Wenn Sie mit einer alten Version von Zookeeper konfrontiert sind, besteht eine Lösung darin, eine Clientbibliothek wie zc.zk für Python zu verwenden. Für Leute, die nicht mit Python vertraut sind, müssen Sie es mit pip oder easy_install installieren . Starten Sie dann eine Python-Shell ( python) und Sie können Folgendes tun:

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

oder auch

zk.delete_recursive('brokers')

wenn Sie alle Themen aus Kafka entfernen möchten.

Mark Butler
quelle
2

So bereinigen Sie alle Nachrichten eines bestimmten Themas mithilfe Ihrer Anwendungsgruppe (Gruppenname sollte mit dem Namen der Anwendungs-Kafka-Gruppe identisch sein).

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

user4713340
quelle
Bei diesem Ansatz gibt es ein Problem (getestet in 0.8.1.1). Wenn eine Anwendung zwei (oder mehr) Themen abonniert: Thema1 und Thema2 und der Konsolenkonsument Thema1 bereinigt, wird leider auch der nicht verwandte Konsumentenoffset für Thema2 gelöscht, wodurch alle Nachrichten von Thema2 wiedergegeben werden.
Jsh
2

Nach der Antwort von @steven appleyard habe ich die folgenden Befehle in Kafka 2.2.0 ausgeführt und sie haben für mich funktioniert.

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
abbas
quelle
Dies scheint andere Antworten zu duplizieren
OneCricketeer
2

Viele gute Antworten hier, aber unter ihnen habe ich keine über Docker gefunden. Ich habe einige Zeit damit verbracht herauszufinden, dass die Verwendung des Broker-Containers in diesem Fall falsch ist (offensichtlich !!!)

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

und ich hätte zookeeper:2181anstelle von --zookeeper localhost:2181gemäß meiner Erstellungsdatei verwenden sollen

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

der richtige Befehl wäre

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

Hoffe, es wird jemandem Zeit sparen.

Beachten Sie außerdem, dass die Nachrichten nicht sofort gelöscht werden und dass das Segment des Protokolls geschlossen wird.

Vladimir Semashkin
quelle
Sie können in den Broker ganz gut ausführen. Das Problem ist localhost:2181... ZB Sie verstehen die Docker-Netzwerkfunktionen falsch. Darüber hinaus sind nicht alle Zookeeper-Container vorhanden. kafka-topicsVerwenden Sie sie daher am besten nicht auf diese Weise. Neueste Kafka-Installationen ermöglichen --bootstrap-serversdas Ändern eines Themas anstelle von--zookeeper
OneCricketeer
1
Trotzdem scheint die Ausführung in den Zookeeper-Container falsch zu sein. you can use --zookeeper zookeeper: 2181` aus dem Kafka Container ist mein Punkt. Oder greifen Sie sogar die Zookeeper-Zeile aus der Datei
server.properties heraus
@ Cricket_007 Hey, danke dafür wirklich, ich habe die Antwort korrigiert, lass es mich wissen, wenn da
drüben
1

Aufgrund der Größe konnte kein Kommentar hinzugefügt werden: Ich bin mir nicht sicher, ob dies zutrifft, abgesehen von der Aktualisierung von Retention.ms und Retention.bytes. Ich habe jedoch festgestellt, dass die Richtlinie zur Bereinigung von Themen "Löschen" (Standard) sein sollte. Wenn "Kompakt", wird dies der Fall sein Halten Sie Nachrichten länger fest, dh wenn es "kompakt" ist, müssen Sie auch delete.retention.ms angeben .

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

Auch mussten früheste / späteste Offsets gleich überwacht werden, um zu bestätigen, dass dies erfolgreich passiert ist, kann auch das du -h / tmp / kafka-logs / test-topic-3-100- * überprüfen

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

Das andere Problem ist, müssen Sie aktuelle Konfiguration erhalten zuerst , so dass Sie zurückkehren denken Sie daran, nach dem Löschen ist erfolgreich: ./bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-name test-topic-3-100 --entity-type topics

kisna
quelle
1

Ein anderer, eher manueller Ansatz zum Löschen eines Themas ist:

in den Maklern:

  1. Stoppen Sie Kafka Broker
    sudo service kafka stop
  2. Löschen Sie alle Partitionsprotokolldateien (sollte auf allen Brokern durchgeführt werden).
    sudo rm -R /kafka-storage/kafka-logs/<some_topic_name>-*

im Tierpfleger:

  1. Führen Sie die zookeeper-Befehlszeilenschnittstelle aus
    sudo /usr/lib/zookeeper/bin/zkCli.sh
  2. Verwenden Sie zkCli, um die Themenmetadaten zu entfernen
    rmr /brokers/topic/<some_topic_name>

wieder in den Maklern:

  1. Starten Sie den Broker-Service neu
    sudo service kafka start
Danny Mor
quelle
Sie müssen Dateien von jedem Broker mit einem Replikat anhalten und entfernen, was bedeutet, dass Sie dabei Ausfallzeiten des Clients haben können
OneCricketeer
1
Sie haben Recht, hier können Sie nur sehen, wo einige Dinge von Kafka gespeichert und verwaltet werden. Dieser Brute-Force-Ansatz ist jedoch definitiv nicht für ein Produktionsbetriebssystem geeignet.
Danny Mor
1
./kafka-topics.sh --describe --zookeeper zkHost:2181 --topic myTopic

Dies sollte retention.mskonfiguriert geben . Dann können Sie den obigen Änderungsbefehl verwenden, um auf 1 Sekunde zu wechseln (und später auf die Standardeinstellungen zurückzukehren).

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

Verwenden Sie von Java aus das Neue AdminZkClientanstelle des Veralteten 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;
  }
Michael Böckling
quelle
Du brauchst keinen Zookeeper. Verwenden Sie AdminClientoderKafkaAdminClient
OneCricketeer