使用在 Confluent 上运行的 Kafka 对 Kafka CLI 进行身份验证答案

【问题标题】：Authenticate Kafka CLI with Kafka running on Confluent使用在 Confluent 上运行的 Kafka 对 Kafka CLI 进行身份验证
【发布时间】：2022-02-16 03:19:35
【问题描述】：

我有一个在 Confluent Cloud 上运行的 Kafka 集群，但我无法从 UI 重置提交偏移量。因此，我尝试通过 Kafka 的 CLI 进行如下操作：

kafka-consumer-groups --bootstrap-server=my_cluster.confluent.cloud:9092 --list

但是，我遇到了以下错误。而且我认为这与我如何进行身份验证有关。

Error: Executing consumer group command failed due to org.apache.kafka.common.KafkaException: Failed to find brokers to send ListGroups
java.util.concurrent.ExecutionException: org.apache.kafka.common.KafkaException: Failed to find brokers to send ListGroups
    at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:396)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2073)
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:165)
    at kafka.admin.ConsumerGroupCommand$ConsumerGroupService.listConsumerGroups(ConsumerGroupCommand.scala:203)
    at kafka.admin.ConsumerGroupCommand$ConsumerGroupService.listGroups(ConsumerGroupCommand.scala:198)
    at kafka.admin.ConsumerGroupCommand$.run(ConsumerGroupCommand.scala:70)
    at kafka.admin.ConsumerGroupCommand$.main(ConsumerGroupCommand.scala:59)
    at kafka.admin.ConsumerGroupCommand.main(ConsumerGroupCommand.scala)
Caused by: org.apache.kafka.common.KafkaException: Failed to find brokers to send ListGroups
    at org.apache.kafka.clients.admin.KafkaAdminClient$24.handleFailure(KafkaAdminClient.java:3368)
    at org.apache.kafka.clients.admin.KafkaAdminClient$Call.handleTimeoutFailure(KafkaAdminClient.java:838)
    at org.apache.kafka.clients.admin.KafkaAdminClient$Call.fail(KafkaAdminClient.java:804)
    at org.apache.kafka.clients.admin.KafkaAdminClient$TimeoutProcessor.handleTimeouts(KafkaAdminClient.java:934)
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.timeoutPendingCalls(KafkaAdminClient.java:1013)
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.processRequests(KafkaAdminClient.java:1367)
    at org.apache.kafka.clients.admin.KafkaAdminClient$AdminClientRunnable.run(KafkaAdminClient.java:1331)
    at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment. Call: findAllBrokers

【问题讨论】：

我正试图做同样的事情，试图抵消给定的值。 @knl，你有什么运气吗？

标签： apache-kafka confluent-cloud

【解决方案1】：

这是一个列出消费者群体的示例

kafka-consumer-groups --bootstrap-server <ccloud kafka>:9092 --command-config consumer.properties --list

consumer.properties

bootstrap.servers=<ccloud kafka>:9092
ssl.endpoint.identification.algorithm=https
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule 
required username="<KEY>" password="<SECRET>";

【讨论】：

【解决方案2】：

您需要使用 --command-config 选项来设置包含您的 CCLoud 凭据的属性文件

【讨论】：

谢谢！你知道我是否可以使用 Confluent 的 CLI 为消费者组重置主题的偏移量吗？或者我应该进行身份验证然后使用 Apache Kafka 的命令行工具？谢谢！
我不使用 Confluent ccloud cli，但我认为它没有那个功能
是的，我想是的，谢谢！
我正试图做同样的事情，试图抵消给定的值。 @knl，你有什么运气吗？
@Sri 给定 TopicPartition +Offset 的 seek 调用应该更容易在代码本身中完成