我有一个kafka流应用程序正在运行（0.10.2.1）。当我关闭kafka集群时，streams应用程序继续等待下一条消息，当集群恢复时，它将继续使用消息。在集群关闭期间，应用程序似乎工作正常。我已经测试了45分钟了。
我希望Kafka抛出一个例外或停止。我有一个 StateListener 当kafkastreams关闭时记录，但是它从未被调用。
kafkaStreams.setStateListener((newState, _) => {
if (newState == KafkaStreams.State.NOT_RUNNING) {
Log.error("Kafka died unexpectedly.")
}
})当kafka无法连接到集群时，如何让它抛出异常或关闭？
注意：这假设集群在应用程序启动后关闭
1条答案
按热度按时间yb3bgrhw1#
你为什么想让Kafka流应用程序宕机？
应用程序应该对代理失败有弹性，也就是说，耐心地运行，直到代理恢复，似乎这就是它正在做的。如果您有多个kafka streams应用程序示例，其中一个示例与代理失去了连接，那么负载将重新平衡到其余示例上。如果每个失去连接的示例都关闭了自己，那么即使代理连接恢复了，您也将失去示例，并随之失去冗余和并行性。现在的Kafka是为恢复力而设计的。我认为这是正确的行为。
如果您想检测代理（或连接）失败，那么这是一个用于监视的用例，而不是将失败引入kafka streams应用程序。