Kafka Streams 在 TopicAuthorizationException 上关闭

Question

我们在 IBM Cloud 上的 Kafka 代理中存在一个潜在问题，我们的 kafka 消费者（Spring ListenerConsumer 和 KafkaStreams）偶尔会遇到 org.apache 异常。 kafka.common.errors.TopicAuthorizationException 在运行/轮询时

我们通过设置 AuthorizationExceptionRetryInterval 克服了 Kafka 监听器中的这个问题，这意味着 kafka 将重试直到异常解决。这里都很好

[ consumer-0-C-1] o.a.k.c.consumer.internals.Fetcher       : [Consumer clientId=consumerxxx_processing00b773de-064b-4c6f-8d0b-31a06de5ea93-14, 
groupId=xxx-00b773de-064b-4c6f-8d0b-31a06de5ea93] 
Not authorized to read from partition xxx_topic_name-2.

2021-11-02 01:15:10.086 ERROR [XXX] 1 --- [ consumer-0-C-1] essageListenerContainer$ListenerConsumer : Authorization Exception, retrying in 30000 ms

但是，当涉及到 kafka 流时，我们找不到类似的东西，因此我们的流正在关闭

2021-11-02 01:15:10.094 ERROR [xxx] 1 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [xxx-8c944f68-9bb5-488c-89c69271200e84c4-StreamThread-1] 
Encountered the following exception during processing and the thread is going to shut down:

org.apache.kafka.common.errors.TopicAuthorizationException: Not authorized to access topics: [xxx_topic_name]

我们的应用可以检测到流已关闭，因此会在一段时间后重新启动。这似乎并不理想。会喜欢 spring kafka 在引擎盖下为我们处理它。

谁能想到更好的解决方法？

Answer 1

正如 Gary 正确建议的那样，我们可以将答案总结如下

启动时出现问题 - 将 StreamsBuilderFactoryBean autoStartup 属性设置为 false 以防止 Spring 在上下文初始化期间尝试启动它。并以编程方式循环启动bean（有延迟），直到成功；您可以为此使用自己的逻辑，或者使用 spring-retry 中的 RetryTemplate 之类的东西。

Stream 已在运行时出现问题 - 将 UncaughtExceptioHandler（由即将发布的 spring-kafka 2.8 中的 StreamsUncaughtExceptionHandler 替换）添加到 StreamsBuilderFactoryBean 并使用它来安排重新启动

Answer 2

这与Spring无关；使用 Apache Kafka 流，它只是帮助以 Spring 友好的方式创建流；它与流本身的运行时特性无关。在使用消费者和生产者 API 时，它拥有更多控制权。

您可以尝试的一件事是将 StreamsBuilderFactoryBean autoStartup 属性设置为 false 以防止 Spring 在上下文初始化期间尝试启动它。

假设在start() 方法期间抛出异常，您可以以编程方式在循环中启动bean（有延迟）直到成功；您可以为此使用自己的逻辑，或者类似 spring-retry 中的RetryTemplate。

如果授权异常发生在运行时（流运行一段时间后），我对此没有任何建议（除了您当前正在做的事情）。