Spring Kafka 消费者以随机时间间隔与代理断开连接

Question

我的kafka配置很简单：

@Bean
public NewTopic generalTopic() {
    return TopicBuilder.name("topic")
            .partitions(5)
            .replicas(5)
            .build();
}

@KafkaListener(id= "anyID", topics="topic")
public void consumer(String message) {
    System.out.println(message);
}

生产者：发送消息调用 kafkaTemplate.send()：

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

kafkaTemplate.send("topic", "message to send");

最后，kafka 配置：

spring.kafka.bootstrap-servers=localhost

spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer

spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer= org.apache.kafka.common.serialization.StringSerializer

我的应用程序是一个桌面应用程序，因此每个客户端都有自己的生产者和消费者，其配置与上面完全相同（不包括@KafkaListener（id=“anyID”，其中 anyID 是用户名）。

启动应用程序时（关闭后立即），有时会生成此错误消息：

2022-10-21 15:13:42.130 ERROR 3524 --- [JavaFX-Launcher] o.springframework.kafka.core.KafkaAdmin  : Could not configure topics

org.springframework.kafka.KafkaException: Timed out waiting to get existing topics; nested exception is java.util.concurrent.TimeoutException

检查 kafka server.log（在 logs/server.log 下），有这个堆栈跟踪：

    [2022-10-21 18:11:28,519] WARN Unexpected exception (org.apache.zookeeper.server.NIOServerCnxn)
EndOfStreamException: Unable to read additional data from client, it probably closed the socket: address = /127.0.0.1:34456, session = 0x100a5e36a580005
    at org.apache.zookeeper.server.NIOServerCnxn.handleFailedRead(NIOServerCnxn.java:163)
    at org.apache.zookeeper.server.NIOServerCnxn.doIO(NIOServerCnxn.java:326)
    at org.apache.zookeeper.server.NIOServerCnxnFactory$IOWorkRequest.doWork(NIOServerCnxnFactory.java:522)
    at org.apache.zookeeper.server.WorkerService$ScheduledWorkRequest.run(WorkerService.java:154)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:750)
[2022-10-21 18:11:43,670] INFO Expiring session 0x100a5e36a580005, timeout of 18000ms exceeded (org.apache.zookeeper.server.ZooKeeperServer)

我确定错误消息与日志堆栈跟踪有关，但我不知道是什么原因造成的，也不知道如何解决。

为了让它恢复工作，我只是重新启动kafka

bin/kafka-server-start.sh -daemon config/server.properties

所以基本上我每次遇到这个错误时都必须重新启动kafka。

有什么想法吗？

作为参考，这是我在客户中遇到的错误的完整参考：

    org.springframework.kafka.KafkaException: Timed out waiting to get existing topics; nested exception is java.util.concurrent.TimeoutException
    at org.springframework.kafka.core.KafkaAdmin.lambda$checkPartitions$8(KafkaAdmin.java:388) ~[spring-kafka-2.8.8.jar:2.8.8]
    at java.base/java.util.HashMap.forEach(HashMap.java:1421) ~[na:na]
    at org.springframework.kafka.core.KafkaAdmin.checkPartitions(KafkaAdmin.java:367) ~[spring-kafka-2.8.8.jar:2.8.8]
    at org.springframework.kafka.core.KafkaAdmin.addOrModifyTopicsIfNeeded(KafkaAdmin.java:263) ~[spring-kafka-2.8.8.jar:2.8.8]
    at org.springframework.kafka.core.KafkaAdmin.initialize(KafkaAdmin.java:200) ~[spring-kafka-2.8.8.jar:2.8.8]
    at org.springframework.kafka.core.KafkaAdmin.afterSingletonsInstantiated(KafkaAdmin.java:167) ~[spring-kafka-2.8.8.jar:2.8.8]
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:974) ~[spring-beans-5.3.22.jar:5.3.22]
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:918) ~[spring-context-5.3.22.jar:5.3.22]
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:583) ~[spring-context-5.3.22.jar:5.3.22]
    at org.springframework.boot.SpringApplication.refresh(SpringApplication.java:734) ~[spring-boot-2.7.3.jar:2.7.3]
    at org.springframework.boot.SpringApplication.refreshContext(SpringApplication.java:408) ~[spring-boot-2.7.3.jar:2.7.3]
    at org.springframework.boot.SpringApplication.run(SpringApplication.java:308) ~[spring-boot-2.7.3.jar:2.7.3]
    at org.springframework.boot.builder.SpringApplicationBuilder.run(SpringApplicationBuilder.java:164) ~[spring-boot-2.7.3.jar:2.7.3]
    at com.nume.main.MirroredMain.springBootApplicationContext(MirroredMain.java:40) ~[classes/:na]
    at com.nume.main.MirroredMain.init(MirroredMain.java:61) ~[classes/:na]
    at javafx.graphics@19/com.sun.javafx.application.LauncherImpl.launchApplication1(LauncherImpl.java:825) ~[javafx.graphics.jar:na]
    at javafx.graphics@19/com.sun.javafx.application.LauncherImpl.lambda$launchApplication$2(LauncherImpl.java:196) ~[javafx.graphics.jar:na]
    at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]
Caused by: java.util.concurrent.TimeoutException: null
    at java.base/java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1960) ~[na:na]
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:2095) ~[na:na]
    at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:180) ~[kafka-clients-3.1.1.jar:na]
    at org.springframework.kafka.core.KafkaAdmin.lambda$checkPartitions$8(KafkaAdmin.java:370) ~[spring-kafka-2.8.8.jar:2.8.8]
    ... 17 common frames omitted

Answer 1

我想你忘了在你的配置中提到端口

在kafka配置中是这样的：

spring.kafka.bootstrap-servers=localhost:9092