【问题标题】:KafkaConsumer never exits .poll method - GroupCoordinatorNotAvailableExceptionKafkaConsumer 永远不会退出 .poll 方法 - GroupCoordinatorNotAvailableException
【发布时间】:2016-11-22 13:07:19
【问题描述】:

我在 java 中有一个 KafkaConsumer 的实现,目前它永远不会退出 .poll 方法。当我在调试模式下深入研究源代码时,我发现它卡在AbstractCoordinator.ensureCoordinatorKnown() 的while 循环中,因为协调器永远找不到。

循环中从sendGroupMetadataRequest() 返回的future 第一次以org.apache.kafka.clients.consumer.internals.SendFailedException 失败,然后每次都以org.apache.kafka.common.errors.GroupCoordinatorNotAvailableException: The group coordinator is not available. 失败。有谁知道为什么会发生这种情况?

如果我使用控制台生产者/消费者,我能够成功发送和接收消息,只有当我使用我的 KafkaConsumer 实现时。此外,消费者确实在我的两台服务器上工作,所以我知道这不是消费者的实现。

以下是创建我的消费者时使用的属性:

Properties props = new Properties();
props.put("bootstrap.servers", "myserver:9000);
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("group.id", groupId);
props.put("enable.auto.commit", "true");
props.put("auto.commit.interval.ms", "1000");
props.put("session.timeout.ms", "30000");

编辑:

主题肯定是在消费者开始之前创建的。

编辑 2: 我删除了集群中的所有代理并重新创建了它们,现在我在不同的点上失败了。在AbstractCoordinator.ensureActiveGroup() 中,尝试重新加入时,从performGroupJoin() 返回的未来反复以org.apache.kafka.common.errors.NotCoordinatorForGroupException: This is not the correct coordinator for this group. 失败。仍然不确定发生了什么。

编辑 3: 我删除了代理并使用不同的 id 重新创建它们,现在 .poll() 方法正在返回并且它正在成功地使用消息。不过,我仍然想知道为什么它一开始就失败了,所以我可以确保它不会再次发生。

【问题讨论】:

  • 哪个版本的 Kafka?经纪人和消费者的版本相同吗? Kafka 真的在监听 myserver:9000(默认为 9092)吗?您可以通过 telnet 从消费者机器连接到 Kafka 吗?
  • 我使用的是 Kafka 0.9.1。是的,kafka 真的在那里听,我为我创建的控制台消费者使用了与 java 相同的代理。刚刚用谷歌搜索了 telnet,我不知道它是否在服务器上运行。当我已经知道我可以从控制台消费者那里消费时,连接到它会证明什么?

标签: java apache-kafka kafka-consumer-api


【解决方案1】:

删除代理并创建新代理解决了这个问题。不过仍然不确定经纪人出了什么问题。

【讨论】:

    猜你喜欢
    • 2018-06-09
    • 1970-01-01
    • 2019-09-23
    • 2014-08-04
    • 2012-09-14
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2020-03-26
    相关资源
    最近更新 更多