Membersihkan Topik Kafka

185

Apakah ada cara untuk membersihkan topik dalam kafka?

Saya mendorong pesan yang terlalu besar ke topik pesan kafka di komputer lokal saya, sekarang saya mendapat pesan kesalahan:

kafka.common.InvalidMessageSizeException: invalid message size

Menambah fetch.sizeitu tidak ideal di sini, karena saya sebenarnya tidak ingin menerima pesan sebesar itu.

Peter Klipfel
sumber

Jawaban:

360

Perbarui sementara waktu retensi pada topik menjadi satu detik:

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

Dan dalam rilis Kafka yang lebih baru, Anda juga dapat melakukannya dengan kafka-configs --entity-type topics

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

kemudian tunggu pembersihan diberlakukan (sekitar satu menit). Setelah dibersihkan, kembalikan nilai sebelumnya retention.ms.

steven appleyard
sumber
8
Itu jawaban yang bagus tetapi bisakah Anda menambahkan deskripsi bagaimana memulai dengan memeriksa nilai retensi.ms topik saat ini?
Greg Dubicki
28
Saya tidak yakin tentang memeriksa konfigurasi saat ini, tetapi saya percaya pengaturan ulang kembali ke default terlihat seperti:bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic MyTopic --deleteConfig retention.ms
aspergillusOryzae
15
Atau tergantung versi:--delete-config retention.ms
aspergillusOryzae
3
hanya sebuah fyi, untuk kafka v. 0.9.0.0, dikatakan: 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 PERINGATAN: Mengubah konfigurasi topik dari skrip ini telah usang dan dapat dihapus dalam rilis mendatang. Ke depan, silakan gunakan kafka-configs.sh untuk fungsionalitas ini
Alper Akture
54
Tampaknya sejak 0.9.0, menggunakan kafka-topics.sh untuk mengubah konfigurasi sudah usang. Opsi baru adalah menggunakan skrip kafka-configs.sh. e.g. kafka-configs.sh --zookeeper <zkhost>:2181 --alter --entity-type topics --entity-name <topic name> --add-config retention.ms=1000 Ini juga memungkinkan Anda untuk memeriksa periode penyimpanan saat ini, misalnya kafka-configs --zookeeper <zkhost>: 2181 - deskripsikan --entri-jenis topik --entitas-nama <nama topik>
RHE
70

Untuk membersihkan antrian, Anda dapat menghapus topik:

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

lalu buat kembali:

bin/kafka-topics.sh --create --zookeeper localhost:2181 \
    --replication-factor 1 --partitions 1 --topic test
rjaiswal
sumber
14
Ingatlah untuk menambahkan baris delete.topic.enable=truedalam file config/server.properties, sebagaimana peringatan yang dicetak oleh perintah yang disebutkan mengatakanNote: This will have no impact if delete.topic.enable is not set to true.
Patrizio Bertoni
3
Ini tidak selalu instan. Terkadang itu hanya akan menandai untuk penghapusan dan penghapusan yang sebenarnya akan terjadi kemudian.
Gaurav Khare
48

Berikut langkah-langkah yang saya ikuti untuk menghapus topik bernama MyTopic:

  1. Jelaskan topiknya, dan jangan gunakan id broker
  2. Hentikan daemon Apache Kafka untuk setiap ID broker yang terdaftar.
  3. Hubungkan ke setiap broker, dan hapus folder data topik, mis rm -rf /tmp/kafka-logs/MyTopic-0. Ulangi untuk partisi lain, dan semua replika
  4. Hapus metadata topik: zkCli.shlalurmr /brokers/MyTopic
  5. Mulai daemon Apache Kafka untuk setiap mesin yang berhenti

Jika Anda merindukan Anda langkah 3, maka Apache Kafka akan terus melaporkan topik tersebut sebagai hadiah (misalnya saat Anda menjalankan kafka-list-topic.sh).

Diuji dengan Apache Kafka 0.8.0.

Thomas Bratt
sumber
2
di 0.8.1 ./zookeeper-shell.sh localhost:2181dan./kafka-topics.sh --list --zookeeper localhost:2181
pdeschen
Dapat digunakan zookeeper-clientsebagai gantinya zkCli.sh(dicoba di Cloudera CDH5)
Martin Tapp
1
Ini menghapus topik, bukan data di dalamnya. Ini mengharuskan Broker dihentikan. Ini adalah hack terbaik. Jawaban Steven Appleyard benar-benar yang terbaik.
Jeff Maass
1
Ini adalah satu-satunya cara pada saat itu ditulis.
Thomas Bratt
2
Bekerja untuk saya di Kafka 0.8.2.1, meskipun topis di zookeeper berada di bawah / broker / topik / <nama topik di sini>
codecraig
44

Meskipun jawaban yang diterima benar, metode itu sudah usang. Konfigurasi topik sekarang harus dilakukan melalui kafka-configs.

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

Konfigurasi yang diatur melalui metode ini dapat ditampilkan dengan perintah

kafka-configs --zookeeper localhost:2181 --entity-type topics --describe --entity-name MyTopic
Shane Perry
sumber
2
Ini juga layak ditambahkan:kafka-configs --zookeeper localhost:2181 --entity-type topics --alter --delete-config retention.ms --entity-name MyTopic
NoBrainer
38

Diuji dalam Kafka 0.8.2, untuk contoh mulai cepat: Pertama, Tambahkan satu baris ke file server.properties di bawah folder config:

delete.topic.enable=true

lalu, Anda dapat menjalankan perintah ini:

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

Dari kafka 1.1

Bersihkan topik

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

tunggu 1 menit, agar aman bahwa kafka bersihkan topik hapus konfigurasi, dan kemudian pergi ke nilai default

bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name tp_binance_kline --delete-config retention.ms
pengguna644265
sumber
1
Saya pikir Anda memiliki panah tambahan. Di tanganku, aku bisa berlaribin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name my-topic --add-config rentention.ms=100
Will
4

kafka tidak memiliki metode langsung untuk membersihkan / membersihkan topik (Antrian), tetapi dapat melakukan ini dengan menghapus topik itu dan membuatnya kembali.

pertama-tama pastikan file sever.properties telah dan jika tidak ditambahkan delete.topic.enable=true

lalu, Hapus topik bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic myTopic

lalu buat lagi.

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

Kadang-kadang, jika Anda memiliki gugus jenuh (terlalu banyak partisi, atau menggunakan data topik terenkripsi, atau menggunakan SSL, atau pengontrol ada di node yang buruk, atau koneksi tidak stabil, akan butuh waktu lama untuk membersihkan topik tersebut .

Saya mengikuti langkah-langkah ini, terutama jika Anda menggunakan Avro.

1: Jalankan dengan alat kafka:

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

2: Jalankan pada simpul registri Skema:

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: Atur retensi topik kembali ke pengaturan awal, setelah topik kosong.

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

Semoga ini bisa membantu seseorang, karena tidak mudah diiklankan.

Ben Coughlan
sumber
Catatan: kafka-avro-console-consumertidak perlu
OneCricketeer
4

UPDATE: Jawaban ini relevan untuk Kafka 0.6. Untuk Kafka 0.8 dan yang lebih baru lihat jawaban oleh @Patrick.

Ya, hentikan kafka dan hapus semua file secara manual dari subdirektori yang sesuai (mudah ditemukan di direktori data kafka). Setelah kafka restart topiknya akan kosong.

Kebakaran
sumber
Ini membutuhkan penghancuran Broker, dan paling-paling merupakan peretasan. Jawaban Steven Appleyard benar-benar yang terbaik.
Jeff Maass
@ MaasSql saya setuju. :) Jawaban ini berumur dua tahun, tentang versi 0.6. Fungsionalitas "ubah topik" dan "hapus topik" telah diterapkan kemudian.
Wildfire
Jawaban Steven Appleyard sama gagalnya dengan jawaban ini.
Banjocat
Memiliki pegangan aplikasi menghapus datanya sendiri dengan cara yang didukung jauh lebih tidak hacky daripada mematikan aplikasi tersebut dan menghapus apa yang Anda pikir semua file datanya kemudian menyalakannya kembali.
Nick
3

Pendekatan paling sederhana adalah mengatur tanggal file log individual menjadi lebih lama dari periode penyimpanan. Kemudian broker harus membersihkannya dan menghapusnya untuk Anda dalam beberapa detik. Ini menawarkan beberapa keuntungan:

  1. Tidak perlu menjatuhkan broker, ini adalah operasi runtime.
  2. Hindari kemungkinan pengecualian offset yang tidak valid (selengkapnya di bawah ini).

Dalam pengalaman saya dengan Kafka 0.7.x, menghapus file log dan memulai kembali broker dapat menyebabkan pengecualian offset tidak valid untuk konsumen tertentu. Ini akan terjadi karena broker me-restart offset di nol (tanpa adanya file log yang ada), dan seorang konsumen yang sebelumnya mengkonsumsi dari topik akan menyambung kembali untuk meminta offset spesifik [sekali valid]. Jika offset ini jatuh di luar batas log topik baru, maka tidak ada salahnya dan konsumen kembali pada awal atau akhir. Tetapi, jika offset berada dalam batas log topik baru, broker mencoba untuk mengambil set pesan tetapi gagal karena offset tidak selaras dengan pesan yang sebenarnya.

Ini bisa dikurangi dengan juga membersihkan offset konsumen di zookeeper untuk topik itu. Tetapi jika Anda tidak memerlukan topik perawan dan hanya ingin menghapus konten yang ada, maka cukup 'menyentuh'-log beberapa topik jauh lebih mudah dan lebih dapat diandalkan, daripada menghentikan broker, menghapus log topik, dan membersihkan node zookeeper tertentu .

Andrew Carter
sumber
bagaimana cara "mengatur tanggal file log individual menjadi lebih lama dari periode penyimpanan"? terima kasih
bylijinnan
3

Saran Thomas sangat bagus tetapi sayangnya zkClidi Zookeeper versi lama (misalnya 3.3.6) tampaknya tidak mendukung rmr. Misalnya membandingkan implementasi command line di Zookeeper modern dengan versi 3.3 .

Jika Anda dihadapkan dengan Zookeeper versi lama, salah satu solusinya adalah menggunakan pustaka klien seperti zc.zk untuk Python. Untuk orang yang tidak terbiasa dengan Python, Anda perlu menginstalnya menggunakan pip atau easy_install . Kemudian mulai shell Python ( python) dan Anda dapat melakukannya:

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

atau bahkan

zk.delete_recursive('brokers')

jika Anda ingin menghapus semua topik dari Kafka.

Mark Butler
sumber
2

Untuk membersihkan semua pesan dari topik tertentu menggunakan grup aplikasi Anda (GroupName harus sama dengan nama grup aplikasi kafka).

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

pengguna4713340
sumber
Ada masalah dengan pendekatan ini (diuji dalam 0.8.1.1). Jika aplikasi berlangganan dua (atau lebih) topik: topic1 dan topic2 dan konsumen konsol membersihkan topic1, sayangnya itu juga menghapus offset konsumen yang tidak terkait untuk topic2, yang menyebabkan pengulangan semua pesan dari topic2.
jsh
2

Mengikuti jawaban appleyard @steven, saya menjalankan perintah berikut pada Kafka 2.2.0 dan mereka bekerja untuk saya.

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
sumber
Ini sepertinya menggandakan jawaban lain
OneCricketeer
2

Banyak jawaban bagus di sini tetapi di antara mereka, saya tidak menemukan satu tentang buruh pelabuhan. Saya menghabiskan beberapa waktu untuk mencari tahu bahwa menggunakan wadah broker adalah salah untuk kasus ini (jelas !!!)

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

dan saya seharusnya menggunakan zookeeper:2181bukan --zookeeper localhost:2181sebagai file menulis saya

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

perintah yang benar adalah

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

Semoga ini akan menghemat waktu seseorang.

Perlu diketahui juga bahwa pesan tidak akan segera dihapus dan itu akan terjadi ketika segmen log akan ditutup.

Vladimir Semashkin
sumber
Anda dapat mengeksekusi ke broker dengan baik. Masalahnya adalah localhost:2181... Misalnya Anda salah memahami fitur jaringan Docker. Selain itu, tidak semua wadah Zookeeper memiliki kafka-topics, jadi yang terbaik adalah tidak menggunakannya seperti itu. Instalasi Kafka terbaru memungkinkan untuk --bootstrap-serversmengubah topik alih-alih--zookeeper
OneCricketeer
1
Namun, exec ke dalam wadah Zookeeper tampaknya salah. you can use --zookeeper zookeeper: 2181` dari wadah Kafka adalah poin saya. Atau bahkan mengeluarkan garis Zookeeper dari file server.properties
OneCricketeer
@ cricket_007 hei, terima kasih untuk ini benar-benar, saya mengoreksi jawabannya, beri tahu saya jika ada sesuatu yang salah di sana
Vladimir Semashkin
1

Tidak dapat menambahkan sebagai komentar karena ukuran: Tidak yakin apakah ini benar, selain memperbarui retention.ms dan retention.bytes, tetapi saya perhatikan kebijakan pembersihan topik harus "delete" (default), jika "compact", itu akan tahan pesan lebih lama, yaitu, jika "kompak", Anda harus menentukan delete.retention.ms juga.

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

Juga harus memantau paling awal / offset terbaru harus sama untuk mengkonfirmasi ini berhasil terjadi, juga dapat memeriksa du-h / tmp / kafka-log / test-topik-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

Masalah lainnya adalah, Anda harus mendapatkan konfigurasi saat ini terlebih dahulu sehingga Anda ingat untuk mengembalikan setelah penghapusan berhasil: ./bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-name test-topic-3-100 --entity-type topics

kisna
sumber
1

Pendekatan lain, agak manual, untuk membersihkan suatu topik adalah:

di broker:

  1. hentikan broker kafka
    sudo service kafka stop
  2. hapus semua file log partisi (harus dilakukan pada semua broker)
    sudo rm -R /kafka-storage/kafka-logs/<some_topic_name>-*

di penjaga kebun binatang:

  1. jalankan antarmuka baris perintah zookeeper
    sudo /usr/lib/zookeeper/bin/zkCli.sh
  2. gunakan zkCli untuk menghapus metadata topik
    rmr /brokers/topic/<some_topic_name>

di broker lagi:

  1. restart layanan broker
    sudo service kafka start
Danny Mor
sumber
Anda perlu berhenti dan menghapus file dari masing-masing broker dengan replika, yang berarti Anda bisa membuat klien downtime ketika melakukan ini
OneCricketeer
1
Anda benar, ini hanya membiarkan Anda benar-benar melihat di mana beberapa hal disimpan dan dikelola oleh Kafka. tetapi pendekatan brute-force ini jelas bukan untuk sistem yang menjalankan produksi.
Danny Mor
1
./kafka-topics.sh --describe --zookeeper zkHost:2181 --topic myTopic

Ini harus memberi retention.msdikonfigurasi. Kemudian Anda dapat menggunakan perintah alter di atas untuk mengubah ke 1detik (dan kemudian kembali ke default).

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

Dari Jawa, gunakan yang baru dan AdminZkClientbukan yang sudah usang 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
sumber
Anda tidak membutuhkan Zookeeper. Gunakan AdminClientatauKafkaAdminClient
OneCricketeer