java - 卡夫卡流 : Proper way to exit on error

Question

我已经成功地让流应用程序使用、转换和生成数据，但我注意到流处理器会周期性地转换到 ERROR 状态并且进程将坐在那里不退出。

向我显示如下日志:

所有流线程都已死亡。该实例将处于错误状态，应关闭。

有没有办法告诉 Streams 应用程序在达到 ERROR 状态后退出？也许是某种监视器线程？

我看到 references在 Kafka Streams 代码的注释中，用户需要在应用程序达到此状态后关闭应用程序，但是，我无法在文档中找到提及此任务的信息。

是否有一种简单的方法来执行此关闭步骤？

---

关闭错误的可能不正确的方法

我的意图是在 KafkaStreams 对象上设置 UncaughtExceptionHandler 方法，以执行以下操作:

• 记录错误
• 在原始 KafkaStreams 对象上使用 close 方法关闭流

结果是:

• 记录异常消息
• INFO org.apache.kafka.streams.KafkaStreams ... 从 ERROR 到 PENDING_SHUTDOWN 的状态转换
• INFO org.apache.kafka.streams.processor.internals.StreamThread ... 通知关闭

然后，不幸的是，这个过程似乎没有退出就挂了。

FWIW 我觉得这可能是对 setUncaughtExceptionHandler

的误用

Answer 1

使用 UncaughtExceptionHandler 是正确的。但是，如果您在处理程序回调中调用 KafkaStreams#close()，您可能会遇到死锁。因此，您应该只设置一个标志，并在回调之外调用 #close()，或者使用带超时的 close()。如果超时到期，则强制关机。