Purge Kafka Topic

¿Hay alguna manera de purgar el tema en kafka?

Introduje un mensaje que era demasiado grande en un tema de mensaje kafka en mi máquina local, ahora recibo un error:

kafka.common.InvalidMessageSizeException: invalid message size

Aumentar el fetch.sizeno es ideal aquí, porque en realidad no quiero aceptar mensajes tan grandes.

Actualice temporalmente el tiempo de retención del tema a un segundo:

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

Y en los nuevos lanzamientos de Kafka, también puedes hacerlo con kafka-configs --entity-type topics

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

luego espere a que la purga surta efecto (aproximadamente un minuto). Una vez purgado, restaure el retention.msvalor anterior .

Para purgar la cola, puede eliminar el tema:

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

luego vuelva a crearlo:

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

Estos son los pasos que sigo para eliminar un tema llamado MyTopic:

1. Describa el tema y no tome los identificadores de los corredores.
2. Detenga el demonio Apache Kafka para cada ID de corredor que se enumere.
3. Conéctese a cada intermediario y elimine la carpeta de datos del tema, por ejemplo rm -rf /tmp/kafka-logs/MyTopic-0. Repita para otras particiones y todas las réplicas
4. Eliminar los metadatos del tema: zkCli.shluegormr /brokers/MyTopic
5. Inicie el demonio Apache Kafka para cada máquina detenida

Si pierde el paso 3, Apache Kafka continuará informando que el tema está presente (por ejemplo, si se ejecuta kafka-list-topic.sh).

Probado con Apache Kafka 0.8.0.

Si bien la respuesta aceptada es correcta, ese método ha quedado en desuso. La configuración del tema ahora debe hacerse a través de kafka-configs.

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

Las configuraciones establecidas a través de este método se pueden mostrar con el comando

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

Probado en Kafka 0.8.2, para el ejemplo de inicio rápido: Primero, agregue una línea al archivo server.properties en la carpeta config:

delete.topic.enable=true

entonces, puedes ejecutar este comando:

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

De kafka 1.1

Purgar un tema

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

espere 1 minuto, para asegurarse de que kafka purgue el tema, elimine la configuración y luego vaya al valor predeterminado

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

kafka no tiene un método directo para el tema de purga / limpieza (Colas), pero puede hacerlo eliminando ese tema y recreándolo.

primero asegúrese de que el archivo sever.properties tenga y si no, agregue delete.topic.enable=true

luego, Eliminar tema bin/kafka-topics.sh --zookeeper localhost:2181 --delete --topic myTopic

luego créelo nuevamente.

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

A veces, si tiene un clúster saturado (demasiadas particiones, o usa datos de temas cifrados, o usa SSL, o el controlador está en un nodo defectuoso, o la conexión es escasa, tomará mucho tiempo purgar dicho tema .

Sigo estos pasos, especialmente si estás usando Avro.

1: Ejecutar con herramientas kafka:

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

2: Ejecutar en el nodo de registro de esquema:

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: Establezca la retención del tema nuevamente a la configuración original, una vez que el tema esté vacío.

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

Espero que esto ayude a alguien, ya que no se anuncia fácilmente.

ACTUALIZACIÓN: Esta respuesta es relevante para Kafka 0.6. Para Kafka 0.8 y posteriores, ver la respuesta de @Patrick.

Sí, detenga kafka y elimine manualmente todos los archivos del subdirectorio correspondiente (es fácil encontrarlo en el directorio de datos de kafka). Después de reiniciar kafka, el tema estará vacío.

El enfoque más simple es establecer que la fecha de los archivos de registro individuales sea anterior al período de retención. Luego, el corredor debe limpiarlos y eliminarlos en unos segundos. Esto ofrece varias ventajas:

1. No es necesario derribar a los corredores, es una operación en tiempo de ejecución.
2. Evita la posibilidad de excepciones de compensación no válidas (más sobre eso a continuación).

En mi experiencia con Kafka 0.7.x, eliminar los archivos de registro y reiniciar el corredor podría generar excepciones de compensación no válidas para ciertos consumidores. Esto sucedería porque el intermediario reinicia las compensaciones en cero (en ausencia de archivos de registro existentes), y un consumidor que anteriormente consumía el tema se volvería a conectar para solicitar un desplazamiento específico [una vez válido]. Si este desplazamiento cae fuera de los límites de los nuevos registros de temas, entonces no hay daño y el consumidor continúa al principio o al final. Pero, si el desplazamiento se encuentra dentro de los límites de los nuevos registros de temas, el intermediario intenta recuperar el conjunto de mensajes pero falla porque el desplazamiento no se alinea con un mensaje real.

Esto podría mitigarse eliminando también las compensaciones del consumidor en el cuidador del zoológico para ese tema. Pero si no necesita un tema virgen y solo desea eliminar el contenido existente, simplemente 'tocar' algunos registros de temas es mucho más fácil y más confiable, que detener a los intermediarios, eliminar registros de temas y borrar ciertos nodos del cuidador del zoológico .

El consejo de Thomas es excelente, pero desafortunadamente zkClien las versiones antiguas de Zookeeper (por ejemplo, 3.3.6) no parece ser compatible rmr. Por ejemplo, compare la implementación de la línea de comandos en el Zookeeper moderno con la versión 3.3 .

Si se enfrenta a una versión anterior de Zookeeper, una solución es utilizar una biblioteca cliente como zc.zk para Python. Para las personas que no están familiarizadas con Python, debe instalarlo utilizando pip o easy_install . Luego inicia un shell de Python ( python) y puedes hacer:

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

o incluso

zk.delete_recursive('brokers')

si quieres eliminar todos los temas de Kafka.

Para limpiar todos los mensajes de un tema en particular usando su grupo de aplicaciones (GroupName debe ser el mismo que el nombre del grupo kafka de la aplicación).

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

Siguiendo la respuesta de @steven appleyard, ejecuté los siguientes comandos en Kafka 2.2.0 y funcionaron para mí.

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

Aquí hay muchas respuestas geniales, pero entre ellas, no encontré ninguna sobre Docker. Pasé algún tiempo para darme cuenta de que usar el contenedor de intermediarios es incorrecto para este caso (¡¡¡¡¡¡¡¡¡¡obviamente) !!!

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

y debería haber usado en zookeeper:2181lugar de --zookeeper localhost:2181según mi archivo de redacción

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

el comando correcto sería

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

Espero que ahorre tiempo a alguien.

Además, tenga en cuenta que los mensajes no se eliminarán de inmediato y sucederá cuando se cierre el segmento del registro.

No se pudo agregar como comentario debido al tamaño: no estoy seguro de si esto es cierto, además de actualizar retención.ms y retención.bytes, pero noté que la política de limpieza del tema debería ser "eliminar" (predeterminado), si es "compacto", va a retener los mensajes por más tiempo, es decir, si es "compacto", también debe especificar 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

También tuve que monitorear las compensaciones más tempranas / más recientes deben ser las mismas para confirmar que esto sucedió con éxito, también puede verificar 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

El otro problema es que primero debe obtener la configuración actual para que recuerde revertir después de que la eliminación se haya realizado correctamente: ./bin/kafka-configs.sh --zookeeper localhost:2181 --describe --entity-name test-topic-3-100 --entity-type topics

Otro enfoque, más bien manual, para purgar un tema es:

en los corredores:

1. detener el corredor de kafka
   sudo service kafka stop
2. eliminar todos los archivos de registro de partición (debe hacerse en todos los intermediarios)
   sudo rm -R /kafka-storage/kafka-logs/<some_topic_name>-*

en cuidador del zoológico:

3. ejecutar la interfaz de línea de comandos de zookeeper
   sudo /usr/lib/zookeeper/bin/zkCli.sh
4. use zkCli para eliminar los metadatos del tema
   rmr /brokers/topic/<some_topic_name>

en los corredores de nuevo:

5. reiniciar el servicio de agente
   sudo service kafka start

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

Esto debería dar retention.msconfigurado. Luego puede usar el comando alter anterior para cambiar a 1 segundo (y luego volver al valor predeterminado)

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

Desde Java, usando el nuevo en AdminZkClientlugar del obsoleto 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;
  }