Spring Kafka消费者以随机的时间间隔不断断开与broker的连接

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

Producer：调用kafkaTemplate.send((发送消息：

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
kafkaTemplate.send("topic", "message to send");

最后，卡夫卡配置：

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/kafk-server-start.sh-守护进程配置/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

Answer 2

您的Spring客户端超时是在转移注意力。

您的Broker无法连接到Zookeeper，导致崩溃。因此，Spring应用程序也将崩溃。重新启动只会暂时修复它，所以你需要解决为什么代理真的与Zookpeer断开连接(Zookepeer日志看起来还好吗？(