Kafka流关闭挂钩和意外异常处理在同一流应用程序中

Question

我的任务是拆除一个开发环境，并从废品中重新设置它，以验证我们的CI-CD流程；唯一的问题是我搞砸了创建一个主题，因此Kafka Streams应用程序退出并出现错误。

我仔细研究了一下，发现了问题并纠正了它，但当我深入研究时，我遇到了另一个奇怪的小问题。

我实现了一个意外的异常处理器，如下所示:

streams.setUncaughtExceptionHandler((t, e) -> {
    logger.fatal("Caught unhandled Kafka Streams Exception:", e);
    // Do some exception handling.
    streams.close();

    // Maybe do some more exception handling.
    // Open a lock that is waiting after streams.start() call 
    // to let application exit normally
    shutdownLatch.countDown();
});

问题是，如果应用程序抛出一个异常，因为一个主题错误时，KafkaStreams::c失去的是调用应用程序似乎死锁在WindowsSeletorImpl::轮询后，尝试调用KafkaStreams::waitOnState。

我认为这可能是在异常处理程序中调用KafkaStreams::close的问题，但我发现这是这样的，Matthias J. Sax的评论说，在异常处理程序中调用KafkaStreams::close应该是可以的，但不要从多个线程中调用KafkaStreams::Close。

问题是，我想实现一个关闭挂钩，以便根据请求优雅地终止steams应用程序，并实现意外异常处理程序，以便在出现异常时进行清理并优雅地终止。

我提出了以下解决方案，该解决方案在调用close之前检查KafkaStreams的状态，它确实有效，但这似乎有点可疑，因为我可以看到除了运行（可能是Pending）之外的其他情况，我们希望确保KafkaStream：：close它调用了。

Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    logger.fatal("Caught Shutdown request");
    // Do some shutdown cleanup.
    if (streams.state().isRunning())
    {
        If this hook is called due to the Main exiting after handling 
        an exception we don't want to call close again. It doesn't 
        cause any errors but logs that the application was closed 
        a second time.
        streams.close(100L, TimeUnit.MILLISECONDS);
    }
    // Maybe do a little bit more clean up before system exits.
    System.exit(0);

}));

streams.setUncaughtExceptionHandler((t, e) -> {
    logger.fatal("Caught unhandled Kafka Streams Exception:", e);
    // Do some exception handling.
    if (streams.state().isRunning())
    {
        streams.close(100L, TimeUnit.MILLISECONDS);
    }
    // Maybe do some more exception handling.

    // Open the Gate to let application exit normally
    shutdownLatch.countDown();
    // Or Optionally call halt to immediately terminate and prevent call to Shutdown hook.
    Runtime.getRuntime().halt(0);
});

关于为什么在异常处理程序中调用KafkaSteams:close会导致这样的问题，或者是否有更好的方法来同时实现关闭钩子和异常处理程序，这将非常受欢迎？

Answer 1

从异常处理程序调用＜code＞close（）＜/code＞与从shutdown钩子调用略有不同close（）如果从shutdown钩子（参见。https://issues.apache.org/jira/browse/KAFKA-4366)因此，您应该使用超时调用它。

此外，该问题与从Jira中描述的未捕获的异常处理程序中调用System.exit（）有关。一般来说，调用System.exit（）非常苛刻，恕我直言，应该避免。

您的解决方案似乎也不是100%健壮的，因为< code>streams.state()。isRunning()可能会导致争用情况。

使用超时的另一种方法可能是仅在关闭挂钩和异常处理程序中设置原子布尔，如果布尔标志设置为true，则使用“main（）”线程调用关闭：

private final static AtomicBoolean stopStreams = new AtomicBoolean(false);

public static void main(String[] args) {
  // do stuff

  KafkaStreams streams = ...
  stream.setUncaughtExceptionHandler((t, e) -> {
    stopStreams.set(true);
  });

  Runtime.getRuntime().addShutdownHook(new Thread(() -> {
    stopStreams.set(true);
  });

  while (!stopStreams.get()) {
    Thread.sleep(1000);
  }
  streams.close();
}