Как удалить группу потребителей kafka (созданную с помощью нового API-интерфейса потребителя)?

Я создал потребителей kafka с помощью нового потребительского API.
Я использую kafka 2.10-0.9.0.1.
У нас есть 1 группа потребителей с 1 экземпляром потребителя в каждой группе.

Сценарий Kafka 'kafka-consumer-groups.sh' предоставляет способ удаления пользователя, но это применимо только для старых групп потребителей. Запуск команды:

bin / kafka-consumer-groups.sh

Дает

--delete: ВНИМАНИЕ! Удаление группы работает только для старых групп потребителей на основе ZK, и нужно использовать его осторожно, чтобы удалять только неактивные группы.

Итак, я хочу спросить, есть ли способ удалить группу потребителей, созданную с помощью нового потребительского API?


spring-boot java apache-kafka web-services kafka-consumer-api
person nikhil7610    schedule 10.06.2016    source источник

Ответы (2)


arrow_upward
5
arrow_downward

Нет необходимости удалять с новым потребителем. Вот что выводит скрипт при попытке удаления:

Обратите внимание, что нет необходимости удалять метаданные группы для нового потребителя, поскольку они автоматически удаляются, когда последний участник уходит.

Это краткий ответ. Подробнее: Под «метаданными» подразумеваются две вещи. Во-первых, это просто информация о потребителях и группах потребителей, которая хранится как часть координатора членства в группе. Он автоматически удаляется, если все потребители в группе ушли.

Во-вторых, группа потребителей сохранила подтвержденные смещения в теме Kafka (когда используется новый потребитель. Раньше они сохранялись в Zookeeper). Эта тема не удаляется сразу после исчезновения группы потребителей. Если группа потребителей появится снова, она автоматически найдет предыдущие смещения в этом разделе. Он может использовать их или игнорировать. Если группа потребителей больше не появляется, эти сохраненные смещения в конечном итоге автоматически собираются мусором.

Короче говоря, при использовании нового потребителя ничего удалять не нужно.

person Eno Thereska    schedule 14.06.2016
comment
Вы знаете, как часто эти офсеты собираются или хранятся в кафке? Есть ли способ удалить эти смещения? - person nikhil7610; 29.06.2016
comment
@ nikhil7610 так не думаю: Error while executing topic command : Topic __consumer_offsets is a kafka internal topic and is not allowed to be marked for deletion. - person Raf; 14.11.2016
comment
Как потребитель может игнорировать ранее сохраненные смещения? Есть ли для этого параметр конфигурации? - person pederpansen; 21.03.2017

arrow_upward
0
arrow_downward

В Kafka 2.5.0 в AdminClient есть Java API под названием deleteConsumerGroups, который можно использовать для удаления отдельных групп потребителей.

Вы можете использовать его, как показано ниже:

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;

public class DeleteConsumerGroups {
  public static void main(String[] args) {
    System.out.println("*** Starting AdminClient to delete a Consumer Group ***");

    final Properties properties = new Properties();
    properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
    properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, "1000");
    properties.put(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, "5000");

    AdminClient adminClient = AdminClient.create(properties);
    String consumerGroupToBeDeleted = "console-consumer-65092";
    DeleteConsumerGroupsResult deleteConsumerGroupsResult = adminClient.deleteConsumerGroups(Arrays.asList(consumerGroupToBeDeleted));

    KafkaFuture<Void> resultFuture = deleteConsumerGroupsResult.all();
    try {
      resultFuture.get();
    } catch (InterruptedException e) {
      e.printStackTrace();
    } catch (ExecutionException e) {
      e.printStackTrace();
    }

    adminClient.close();
  }
}
person mike    schedule 30.09.2020