Kafka:删除空闲的消费者组ID [英] Kafka: Delete idle consumer group id

问题描述

在某些情况下，我使用 Kafka-stream 对主题的小型内存(哈希图)投影进行建模.K,V 缓存确实需要一些操作，因此它不是 GlobalKTable 的好例子.在这样的缓存"场景中，我希望我所有的兄弟实例都拥有相同的缓存，所以我需要绕过消费者组机制.

In some cases, I use Kafka-stream to model a small in memory (hashmap) projection of a topic. The K,V cache does require some manipulations, so it is not a good case for a GlobalKTable. In such a "caching" scenario, I want all my sibling instances to have the same cache, so I need to bypass the consumer-group mechanism.

要启用此功能，我通常只需使用随机生成的应用程序 ID 启动我的应用程序，因此每个应用程序每次重新启动时都会重新加载主题.唯一需要注意的是，我最终将一些消费者组孤立在 kafka 代理上，直到 offsets.retention.minutes 这对于我们的操作监控工具来说并不理想.知道如何解决这个问题吗?

To enable this, I normally simply start my apps with a randomly generated application Id, so each app will reload the topic each time it restarts. The only caveat to that is that I end up with some consumer group orphaned on the kafka brokers up to the offsets.retention.minutes which is not ideal for our operational monitoring tools. Any idea how to workaround this?

• 我们是否可以将 applicationId 配置为临时的，以便在应用程序关闭后消失?
• 或者我们是否可以强制消费者仅在本地管理其偏移量?
• 或者是否有一些 java adminApi 可以用来在正常关闭应用程序时清理我的消费者组 ID?

谢谢

推荐答案

AdminClient 中有一个名为 deleteConsumerGroups 的 Java API，可用于删除单个 ConsumerGroup.

There is a Java API in the AdminClient called deleteConsumerGroups which can be used to delete individual ConsumerGroups.

您可以在 Kafka 2.5.0 中使用它，如下所示.

You can use it as shown below with Kafka 2.5.0.

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();
  }
}

运行上述代码前的ConsumerGroups列表

$ kafka-consumer-groups --bootstrap-server localhost:9092 --list
console-consumer-65092
console-consumer-53268

运行上述代码后的ConsumerGroups列表

$ kafka-consumer-groups --bootstrap-server localhost:9092 --list
console-consumer-53268

这篇关于Kafka:删除空闲的消费者组ID的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！