Чистка Кафка Тема

Есть ли способ очистить тему в кафке?

Я поместил сообщение, которое было слишком большим, в тему сообщения kafka на моей локальной машине, и теперь я получаю сообщение об ошибке:

kafka.common.InvalidMessageSizeException: invalid message size

Увеличение fetch.sizeне является идеальным здесь, потому что я не хочу принимать такие большие сообщения.

Временно обновите время хранения темы до одной секунды:

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

И в более новых выпусках Kafka, вы также можете сделать это с kafka-configs --entity-type topics

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

затем подождите, пока очистка не вступит в силу (около минуты). После очистки восстановите предыдущее retention.msзначение.

Для очистки очереди вы можете удалить тему:

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

затем воссоздайте его:

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

Вот шаги, которые я выполняю, чтобы удалить тему с именем MyTopic:

1. Опишите тему, и не берите идентификаторы брокера
2. Остановите демон Apache Kafka для каждого из перечисленных идентификаторов брокера.
3. Подключитесь к каждому брокеру и удалите папку данных темы, например rm -rf /tmp/kafka-logs/MyTopic-0. Повторите для других разделов и всех реплик
4. Удалить метаданные темы: zkCli.shзатемrmr /brokers/MyTopic
5. Запустите демон Apache Kafka для каждой остановленной машины.

Если вы пропустите шаг 3, то Apache Kafka продолжит сообщать о теме как о существующей (например, если вы запускаете kafka-list-topic.sh).

Протестировано с Apache Kafka 0.8.0.

Хотя принятый ответ верен, этот метод устарел. Настройка темы теперь должна быть сделана через kafka-configs.

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

Конфигурации, установленные с помощью этого метода, можно отобразить с помощью команды

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

Протестировано в Kafka 0.8.2, для примера быстрого запуска: Сначала добавьте одну строку в файл server.properties в папке config:

delete.topic.enable=true

Затем вы можете запустить эту команду:

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

Из кафка 1.1

Очистить тему

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

подождите 1 минуту, чтобы быть уверенным, что кафка очистит тему, удалите конфигурацию, а затем перейдите к значению по умолчанию

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

У kafka нет прямого метода для очистки / очистки темы (Очереди), но это можно сделать, удалив эту тему и воссоздав ее.

сначала убедитесь, что файл sever.properties имеет и если не добавить delete.topic.enable=true

затем Удалить тему bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic myTopic

затем создайте его снова.

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

Иногда, если у вас насыщенный кластер (слишком много разделов, или используются зашифрованные данные раздела, или используется SSL, или контроллер находится на поврежденном узле, или соединение ненадежно, для очистки указанной темы может потребоваться много времени). ,

Я следую этим шагам, особенно если вы используете Avro.

1: Запуск с инструментами Кафки:

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

2: Запустить на узле реестра схемы:

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: Установите сохранение темы обратно к первоначальной настройке, когда тема пуста.

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

Надеюсь, это кому-то поможет, так как это нелегко рекламировать.

ОБНОВЛЕНИЕ: Этот ответ актуален для Кафки 0.6. Для Кафки 0,8 и позже смотрите ответ @Patrick.

Да, остановите kafka и вручную удалите все файлы из соответствующего подкаталога (это легко найти в каталоге данных kafka). После перезагрузки кафки тема будет пустой.

Самый простой подход состоит в том, чтобы установить дату отдельных файлов журнала, которая будет старше, чем срок хранения. Затем брокер должен очистить их и удалить их для вас в течение нескольких секунд. Это предлагает несколько преимуществ:

1. Нет необходимости сбивать брокеров, это операция во время выполнения.
2. Предотвращает возможность недопустимых исключений смещения (подробнее об этом ниже).

По моему опыту работы с Kafka 0.7.x удаление файлов журнала и перезапуск посредника могут привести к недопустимым исключениям смещения для определенных потребителей. Это может произойти, потому что посредник перезапускает смещения с нуля (при отсутствии каких-либо существующих файлов журнала), а потребитель, который ранее потреблял из этой темы, переподключается, чтобы запросить конкретное [когда-то действительное] смещение. Если это смещение выходит за границы журналов новых тем, то это не повредит, и потребитель возобновляет работу в начале или в конце. Но, если смещение попадает в границы новых журналов тем, брокер пытается получить набор сообщений, но не удается, потому что смещение не совпадает с реальным сообщением.

Это может быть смягчено также очисткой смещения потребителей в zookeeper для этой темы. Но если вам не нужна нетронутая тема и вы просто хотите удалить существующее содержимое, то просто «прикоснуться» к нескольким журналам тем гораздо проще и надежнее, чем останавливать посредников, удалять журналы тем и очищать определенные узлы zookeeper ,

Совет Томаса великолепен, но, к сожалению, zkCliв старых версиях Zookeeper (например, 3.3.6), похоже, не поддерживается rmr. Например, сравните реализацию командной строки в современном Zookeeper с версией 3.3 .

Если вы столкнулись со старой версией Zookeeper, одним из решений является использование клиентской библиотеки, такой как zc.zk для Python. Для тех, кто не знаком с Python, вам необходимо установить его с помощью pip или easy_install . Затем запустите оболочку Python ( python), и вы можете сделать:

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

или даже

zk.delete_recursive('brokers')

если вы хотите удалить все темы из Кафки.

Чтобы очистить все сообщения от определенной темы, используя вашу группу приложений (GroupName должно совпадать с именем группы приложения kafka).

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

После ответа @steven appleyard я выполнил следующие команды на Kafka 2.2.0, и они работали для меня.

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

Здесь много хороших ответов, но среди них я не нашел ни одного о докере. Я потратил некоторое время, чтобы понять, что использование контейнера брокера в этом случае не подходит (очевидно !!!)

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

и я должен был использовать zookeeper:2181вместо того, чтобы --zookeeper localhost:2181в соответствии с моим файлом составить

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

правильная команда будет

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

Надеюсь, это сэкономит кому-то время.

Также помните, что сообщения не будут удалены немедленно, и это произойдет, когда сегмент журнала будет закрыт.

Не удалось добавить в качестве комментария из-за размера: Не уверен, что это правда, кроме обновления retention.ms и retention.bytes, но я заметил, что политика очистки темы должна быть «delete» (по умолчанию), если «compact», она собирается задерживать сообщения дольше, т. е. если оно «компактное», вам также нужно указать 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

Также нужно было отслеживать самые ранние / последние смещения, чтобы подтвердить, что это произошло успешно, также можете проверить 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

Другая проблема заключается в том, что вы должны сначала получить текущую конфигурацию, чтобы вы не забыли вернуться после успешного удаления: ./bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-name test-topic-3-100 --entity-type topics

Другой, довольно ручной подход для очистки темы:

в брокерах:

1. стоп кафка брокер
   sudo service kafka stop
2. удалить все файлы журналов разделов (должно быть сделано на всех брокерах)
   sudo rm -R /kafka-storage/kafka-logs/<some_topic_name>-*

в зоопарке:

3. запустить интерфейс командной строки zookeeper
   sudo /usr/lib/zookeeper/bin/zkCli.sh
4. используйте zkCli для удаления метаданных темы
   rmr /brokers/topic/<some_topic_name>

опять в брокерах:

5. перезапустить брокерскую службу
   sudo service kafka start

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

Это должно дать retention.msнастроено. Затем вы можете использовать указанную выше команду alter, чтобы изменить значение на 1 секунду (и позже вернуться к значению по умолчанию).

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

Из Java, используя новое AdminZkClientвместо устаревшего 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;
  }