当Kafka群集出现故障时,KafkaStreams应用程序不会失败

我有一个运行Kafka Streams的应用程序(0.10.2.1).当我关闭Kafka群集时,流应用程序继续等待下一条消息,当群集重新启动时,它将恢复消耗消息.对于群集关闭的持续时间,应用程序似乎正常工作.我测试了这个超过45分钟.

我希望卡夫卡能够抛出异常或停止.我已经配置了一个StateListener在KafkaStreams关闭时记录,但是从不调用它.

kafkaStreams.setStateListener((newState, _) => {
  if (newState == KafkaStreams.State.NOT_RUNNING) {
    Log.error("Kafka died unexpectedly.")
  }
})

如何在无法连接到群集时让Kafka抛出异常或关闭？

注意:这假定应用程序启动后群集关闭