Jak już tu wspomniałem Oczyść kolejkę Kafki :

Testowane w Kafka 0.8.2, dla przykładu quick-start: najpierw dodaj jedną linię do serwera.Plik Właściwości w folderze konfiguracyjnym:

delete.topic.enable=true

Następnie możesz uruchomić następujące polecenie:

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

Testowane z kafka 0.10

1. stop zookeeper & Kafka server,
2. then go to 'kafka-logs' folder , there you will see list of kafka topic folders, delete folder with topic name
3. go to 'zookeeper-data' folder , delete data inside that.
4. start zookeeper & kafka server again.

Uwaga: jeśli usuwasz foldery tematyczne wewnątrz kafka-logs, ale nie z folderu zookeeper-data, wtedy zobaczysz, że tematy nadal tam są.

Jako brudne obejście można dostosować ustawienia przechowywania dla poszczególnych tematów, np. bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic my_topic --config retention.bytes=1 (retencja.bajtów = 0 może również działać)

Po krótkim czasie kafka powinna zwolnić miejsce. Nie jestem pewien, czy ma to jakieś konsekwencje w porównaniu do ponownego tworzenia tematu.

Ps. Lepiej przywrócić ustawienia retencji, gdy kafka skończy czyszczenie.

Możesz również użyć retention.ms do przechowywania danych historycznych

Poniżej znajdują się skrypty do opróżniania i usuwania tematu Kafka zakładając localhost jako serwer zookeeper i Kafka_Home jest ustawiona w katalogu instalacyjnym:

Poniższy skrypt opróżni temat, ustawiając jego czas przechowywania na 1 sekundę, a następnie usuwając konfigurację:

#!/bin/bash
echo "Enter name of topic to empty:"
read topicName
/$Kafka_Home/bin/kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name $topicName --add-config retention.ms=1000
sleep 5
/$Kafka_Home/bin/kafka-configs --zookeeper localhost:2181 --alter --entity-type topics --entity-name $topicName --delete-config retention.ms

Do całkowicie usuń tematy należy zatrzymać odpowiedni broker kafka i usunąć jego katalog z katalogu kafka log (domyślnie: / tmp / kafka-logs), a następnie uruchomić ten skrypt do Usuń temat z zookeeper. Aby zweryfikować, że został usunięty z zookeeper, wyjście ls / brokers / topics nie powinno już zawierać tematu:

#!/bin/bash
echo "Enter name of topic to delete from zookeeper:"
read topicName
/$Kafka_Home/bin/zookeeper-shell localhost:2181 <<EOF
rmr /brokers/topics/$topicName
ls /brokers/topics
quit
EOF

Próbowaliśmy z umiarkowanym powodzeniem tego, co opisują inne odpowiedzi. To, co naprawdę dla nas działało (Apache Kafka 0.8.1), to polecenie klasy

Sh kafka-run-class.sh kafka.admin.DeleteTopicCommand --topic yourtopic --zookeeper localhost: 2181

Dla użytkowników browaru

Jeśli używasz brew jak ja i marnujesz dużo czasu na szukanie niesławnego folderu kafka-logs, nie bój się więcej. (i proszę daj mi znać, czy to działa dla Ciebie i wielu różnych wersji Homebrew, Kafka itp:)) {]}

Prawdopodobnie znajdziesz go pod:

Lokalizacja:

/usr/local/var/lib/kafka-logs

Jak znaleźć tę ścieżkę

(jest to również pomocne w zasadzie dla każdej aplikacji, którą instalujesz przez brew)

1) brew services list

Kafka zaczęła matbhz / Users / matbhz / Library/LaunchAgents / homebrew.mxcl.kafka.plist

2) Otwórz i przeczytaj, że plist znalazłeś powyżej

3) znajdź linię definiującą server.properties lokację otwórz ją, w moim przypadku:

• /usr/local/etc/kafka/server.properties

4) poszukaj linii log.dirs:

Log.dirs = /usr/local/var/lib / kafka-logs

5) przejdź do tej lokalizacji i usuń logi dla tematów, które wish

6) Restart Kafka z brew services restart kafka

Wszystkie dane o tematach i ich partycjach są przechowywane w tmp/kafka-logs/. Ponadto są one przechowywane w formacie topic-partionNumber, więc jeśli chcesz usunąć temat newTopic, możesz:

• stop Kafce
• Usuń pliki rm -rf /tmp/kafka-logs/newTopic-*

1. Stop ZooKeeper i Kafka
2. na serwerze.właściwości, dziennik zmian.retencja.wartość godzin. Możesz skomentować log.retention.hours i dodać log.retention.ms=1000. To utrzymywałoby rekord na temat Kafki tylko przez sekundę.
3. Start zookeeper i kafka.
4. sprawdź konsolę konsumencką. Kiedy po raz pierwszy otworzyłem konsolę, była tam Płyta. Ale kiedy ponownie otworzyłem konsolę, rekord został usunięty.
5. później możesz ustawić wartość log.retention.hours Na żądaną tak.

Od wersji kafka 2.3.0 istnieje alternatywny sposób na miękkie usunięcie Kafki (stare podejście jest przestarzałe ).

Update retention.ms na 1 sek (1000ms), a następnie ustawić go ponownie po min, do domyślnego ustawienia tj. 7 dni (168 godzin, 604,800,000 w ms)

Soft deletion: - (rentention. ms=1000) (using kafka-configs.sh)

bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --alter --entity-name kafka_topic3p3r --entity-type topics  --add-config retention.ms=1000
Completed Updating config for entity: topic 'kafka_topic3p3r'.

Ustawienie na domyślne: - 7 dni (168 godzin , retention.ms= 604800000)

bin/kafka-configs.sh --zookeeper 192.168.1.10:2181 --alter --entity-name kafka_topic3p3r --entity-type topics  --add-config retention.ms=604800000

W ręcznym usuwaniu tematu z klastra Kafki, możesz to sprawdzić https://github.com/darrenfu/bigdata/issues/6 Ważnym krokiem w większości rozwiązań jest usunięcie /config/topics/<topic_name> w ZK.

Używam tego skryptu:

#!/bin/bash
topics=`kafka-topics --list --zookeeper zookeeper:2181`
for t in $topics; do 
    for p in retention.ms retention.bytes segment.ms segment.bytes; do
        kafka-topics --zookeeper zookeeper:2181 --alter --topic $t --config ${p}=100
    done
done
sleep 60
for t in $topics; do 
    for p in retention.ms retention.bytes segment.ms segment.bytes; do
        kafka-topics --zookeeper zookeeper:2181 --alter --topic $t --delete-config ${p}
    done
done

Używam poniższego narzędzia do czyszczenia po uruchomieniu testu integracji.

Używa najnowszego api AdminZkClient. Starsze api zostało przestarzałe.

import javax.inject.Inject
import kafka.zk.{AdminZkClient, KafkaZkClient}
import org.apache.kafka.common.utils.Time

class ZookeeperUtils @Inject() (config: AppConfig) {

  val testTopic = "users_1"

  val zkHost = config.KafkaConfig.zkHost
  val sessionTimeoutMs = 10 * 1000
  val connectionTimeoutMs = 60 * 1000
  val isSecure = false
  val maxInFlightRequests = 10
  val time: Time = Time.SYSTEM

  def cleanupTopic(config: AppConfig) = {

    val zkClient = KafkaZkClient.apply(zkHost, isSecure, sessionTimeoutMs, connectionTimeoutMs, maxInFlightRequests, time)
    val zkUtils = new AdminZkClient(zkClient)

    val pp = new Properties()
    pp.setProperty("delete.retention.ms", "10")
    pp.setProperty("file.delete.delay.ms", "1000")
    zkUtils.changeTopicConfig(testTopic , pp)
    //    zkUtils.deleteTopic(testTopic)

    println("Waiting for topic to be purged. Then reset to retain records for the run")
    Thread.sleep(60000L)

    val resetProps = new Properties()
    resetProps.setProperty("delete.retention.ms", "3000000")
    resetProps.setProperty("file.delete.delay.ms", "4000000")
    zkUtils.changeTopicConfig(testTopic , resetProps)

  }


}

Istnieje opcja Usuń temat. Ale to oznacza temat do usunięcia. Zookeeper później usuwa temat. Ponieważ może to być nieprzewidywalnie długie, wolę retention.ms podejście