弹簧靴。卡夫卡。断开与节点的连接答案

【问题标题】：Spring boot. Kafka. Disconnect from Node弹簧靴。卡夫卡。断开与节点的连接
【发布时间】：2022-11-28 08:12:18
【问题描述】：

尝试时我与节点断开连接听到订阅的主题。我不需要生成消息，它已经实现了。 VPN 用于连接 Kafka。我使用 Spring boot 2.7.0，Java 17。

配置：

pom.xml:

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>2.9.2</version>
</dependency>

配置类：

@EnableKafka
@Configuration
public class KafkaConsumerConfig {
@Bean
public ConsumerFactory<String, String> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.36.12.5:2181");
    props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id");
    props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
    return new DefaultKafkaConsumerFactory<>(props);
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String>
            factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    return factory;
}
}

听众：

@Component
public class KafkaListenersService {
    @KafkaListener(topics = "ift.notification.clientId.request", groupId = "group-id")
    public void listen(String message) {
        System.out.println("Received Message in group - group-id: " + message);
    }

}

我已经完成了哪些步骤：

我将主机域及其 IP 地址添加到 /etc/hosts。所以它正在被正确解决。
我使用Offset Explorer 2作为kafka工具，成功连接到指定主机。我找到了我需要的主题并设法从中读取消息。我认为这意味着我可以在本地连接到kafka，所以这意味着我可以做到也来自 Java.

我还尝试将 Kafka 的设置从 @Configuration 类移动到 application.yml。它看起来像这样：

spring:
  kafka:
    consumer:
      bootstrap-servers: 10.36.12.5:2181
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

las，我断开了连接，也无法阅读任何消息。

我在日志中得到的：

2022-11-22 20:29:21.715  INFO 5005 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka version: 3.2.3
2022-11-22 20:29:21.716  INFO 5005 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka commitId: 50029d3ed8ba576f
2022-11-22 20:29:21.716  INFO 5005 --- [           main] o.a.kafka.common.utils.AppInfoParser     : Kafka startTimeMs: 1669134561713
2022-11-22 20:29:21.719  INFO 5005 --- [           main] o.a.k.clients.consumer.KafkaConsumer     : [Consumer clientId=consumer-group-id-1, groupId=group-id] Subscribed to topic(s): ift.notification.clientId.request
2022-11-22 20:29:21.743  INFO 5005 --- [           main] insure.pulse.Main                        : Started Main in 2.153 seconds (JVM running for 2.83)
2022-11-22 20:29:22.265  INFO 5005 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-group-id-1, groupId=group-id] Node -1 disconnected.
2022-11-22 20:29:22.268  INFO 5005 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-group-id-1, groupId=group-id] Cancelled in-flight API_VERSIONS request with correlation id 1 due to node -1 being disconnected (elapsed time since creation: 149ms, elapsed time since send: 149ms, request timeout: 30000ms)
2022-11-22 20:29:22.268  WARN 5005 --- [ntainer#0-0-C-1] org.apache.kafka.clients.NetworkClient   : [Consumer clientId=consumer-group-id-1, groupId=group-id] Bootstrap broker 10.36.12.5:2181 (id: -1 rack: null) disconnected

之后警告不断重复。我认为这是因为 KafkaListener 一直在尝试连接到 Node。

任何帮助都感激不尽。也可以随意询问任何其他信息，我很乐意提供。

【问题讨论】：

您确定 kafka 节点已启动并正在运行吗？

标签： java spring-boot apache-kafka

【解决方案1】：

我实际上发现了原因是什么。我使用通过 Offset Explorer 2 连接到 kafka 作为参考。我看到 Zookeeper 端口是2181: 所以我也将它用作我的 application.yml 中引导服务器的端口：

spring:
  kafka:
    consumer:
      bootstrap-servers: 10.36.12.5:2181

那是我的错误。看起来 zookeeper 不是引导服务器，但它可以为您提供引导服务器的地址。所以应该在 application.yml 中指定引导服务器本身的地址（及其特定端口）：

spring:
  kafka:
    consumer:
      bootstrap-servers: 10.36.12.5:9092

我也在 Offset Explorer 中找到了 bootstrap-server 端口：

当我修复它时，一切正常。

【讨论】：