Kafka-Elasticsearch连接器异常：重复启动后无数据同步且报超时错误

我之前也碰到过几乎一模一样的问题！这个Flush timeout expired with unflushed records错误本质上是连接器在尝试把攒好的批量记录写入Elasticsearch时，超出了预设的等待时间。结合你说的「首次同步正常、后续启动无法同步新数据」的场景，咱们可以一步步排查解决：

1. 先确认Elasticsearch的健康与性能状态 #

这是最常见的触发原因——ES扛不住写入压力了：

• 先跑个命令检查ES集群健康：curl -XGET 'http://<你的ES地址>:<端口>/_cluster/health?pretty'，确保状态是green或者至少是yellow（red的话集群肯定有问题）
• 查看ES的资源使用情况：比如磁盘IO、CPU、内存使用率，如果这些指标飙高，说明ES处理写入请求的速度跟不上连接器的发送速度，自然会超时
• 检查目标索引的分片配置：如果分片数太少，写入请求会集中在少数分片上；分片数太多，又会增加ES的协调开销，都可能导致写入延迟

2. 调整连接器的Flush相关配置 #

打开你的log-platform-elastic.properties，针对性调整这几个关键参数：

• flush.timeout.ms：默认是5000ms（5秒），如果ES写入慢，直接调大这个值，比如改成flush.timeout.ms=30000（给ES30秒时间处理批量请求）
• batch.size：默认是一次发1000条记录，如果ES处理不过来，就把这个数改小，比如batch.size=500，减少单次批量的压力
• linger.ms：可以设置成linger.ms=1000，让连接器等1秒再凑齐一批发送，避免频繁的小批量请求占用ES资源

3. 检查Kafka消费者偏移量是否正常 #

首次同步后，连接器的消费者组可能已经提交了偏移量，但再次启动时有没有正确读取到新数据的偏移量？

• 用Kafka自带工具查看消费者组状态：kafka-consumer-groups.sh --bootstrap-server <你的Kafka地址>:<端口> --group <你的连接器组ID> --describe
• 重点看CURRENT-OFFSET和LOG-END-OFFSET，如果前者远落后于后者，说明连接器没读到新数据；如果两者相等但你确定有新数据，可能需要重置偏移量：

  kafka-consumer-groups.sh --bootstrap-server <Kafka地址>:<端口> --group <连接器组ID> --reset-offsets --to-earliest --topic <你的目标Topic> --execute
  

4. 查看连接器的详细日志找线索 #

默认日志可能不够详细，咱们打开connect-avro-standalone.properties，把log4j.rootLogger改成DEBUG, stdout，然后重新启动连接器，看日志里有没有更具体的错误：

• 如果看到ES返回429 Too Many Requests，说明ES触发了写入限流，这时候可以把连接器的max.in.flight.requests.per.connection改成1，减少并发请求；同时调整ES的indices.memory.index_buffer_size参数，给ES更多内存处理写入
• 如果看到连接超时，那大概率是网络问题

5. 排查网络连接稳定性 #

确认连接器所在机器到ES集群的网络是通畅的：

• 用ping <ES地址>或者traceroute <ES地址>测试延迟和丢包情况
• 检查ES的端口有没有对外开放，防火墙有没有拦截请求

一般按照这个顺序排查，很快就能定位到问题所在。

内容的提问来源于stack exchange，提问作者Oles Rid