Scala Embedded Kafka结合Kafka Streams的生产者线程泄漏问题

你遇到的核心问题是测试结束后Kafka Streams的生产者线程未被彻底终止，残留的线程持续尝试连接已关闭的Broker，从而重复输出错误日志。以下是针对性的解决方案，按优先级排序：

1. 使用带超时参数的close()方法，确保资源完全清理 #

Kafka Streams的无参close()会使用默认超时时间，但如果你的测试中有未完成的消息发送（比如向自身主题发送删除事件的逻辑），这个时间可能不足以让Streams完成所有收尾工作。显式指定一个足够长的超时时间，给足资源释放的时间：

finally {
  // 根据测试场景调整时长，比如30秒
  streams.close(Duration.ofSeconds(30))
}

2. 清理多余的KafkaStreams实例 #

看你的测试代码，开头定义了一个未使用的streams实例：

val streams = new KafkaStreams(topology,properties)

这个实例没有被关闭，直接导致了线程泄漏！请移除这个多余的定义，确保测试中只创建并管理一个KafkaStreams实例。

3. 关闭后调用cleanUp()清理本地状态 #

如果你的应用使用了状态存储，关闭Streams后调用cleanUp()可以彻底清理本地状态文件和关联资源，避免残留后台线程：

finally {
  streams.close(Duration.ofSeconds(30))
  streams.cleanUp() // 清理本地状态存储
}

4. 检查自定义生产者的资源泄漏 #

测试中使用的publishToKafka方法如果内部创建了独立的KafkaProducer实例，一定要确保它被正确关闭。比如修改方法实现，或者在测试中主动关闭：

// 假设publishToKafka返回生产者引用
val testProducer = publishToKafka(eventTopic, key = keyMSite1UID1, message = event11a)
try {
  // ... 其他发布操作
} finally {
  testProducer.close(Duration.ofSeconds(5)) // 关闭测试用生产者
}

5. 等待线程终止（最后手段） #

如果以上方法无效，可以在关闭后主动等待Streams线程终止，必要时中断残留线程：

finally {
  streams.close(Duration.ofSeconds(30))
  // 等待所有Stream线程结束
  streams.localThreads().forEach { thread =>
    try {
      thread.join(10000) // 等待10秒
      if (thread.isAlive) {
        thread.interrupt()
      }
    } catch {
      case e: InterruptedException => Thread.currentThread().interrupt()
    }
  }
}

• 替换固定延迟为状态监听：不要用Thread.sleep等待Streams初始化，而是监听KafkaStreams.State变化，直到进入RUNNING状态再发送消息，更可靠。
• 检查拓扑循环逻辑：因为应用会向自身消费的主题发删除事件，可能存在循环处理的情况，导致关闭时还有未完成的消息循环。可以在测试中添加开关，让拓扑收到关闭信号时停止发送事件。

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