Kafka Streams:错误退出的正确方法

Question

我已经成功地获得了一个使用,转换和生成数据的流应用程序,但我注意到,流处理器将定期转换到状态,ERROR并且进程将在不退出的情况下坐在那里.

显示我的日志:

All stream threads have died. The instance will be in error state and should be closed.

有没有办法告诉Streams应用程序一旦到达ERROR状态就退出？也许是各种监视器线程？

我看到Kafka Streams代码的注释中的引用对于需要在应用程序达到此状态时关闭应用程序的用户,但是,我无法在文档中找到提及此任务的内容.

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

---

可能是错误的方式可能关闭错误

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

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

结果是:

• 记录异常消息
• INFO org.apache.kafka.streams.KafkaStreams ... State transition from ERROR to PENDING_SHUTDOWN
• INFO org.apache.kafka.streams.processor.internals.StreamThread ... Informed to shut down

然后,不幸的是,这个过程似乎没有退出.

FWIW我觉得这可能是误用了 setUncaughtExceptionHandler

Answer 1

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