如何在 Kafka Streams 应用程序中处理偏移提交期间的超时异常答案

【问题标题】：How to handle Timeout exceptions during Offset Commit in a Kafka Streams application如何在 Kafka Streams 应用程序中处理偏移提交期间的超时异常
【发布时间】：2020-10-22 23:37:14
【问题描述】：

如果在 Kafka Streams 应用程序中提交偏移量时出现超时异常，ProductionExceptionHandler 会起作用吗？

Timeout of 60000ms expired before successfully committing offsets

我正在使用基于 Spring 云流的 Kafka 流应用程序，并且我确实将 ProductionExceptionHandler 实现配置为 LogAndContinue。但似乎控件并没有出现在这个超时异常的异常处理程序中，并且所有流线程都死了。

处理这个问题的正确方法是什么？这是完整的错误。

[-StreamThread-1] o.a.k.s.p.i.AssignedStreamsTasks         : stream-thread [stream-app-0d77ace1-4d20-413b-a71a-94cca3e620e0-StreamThread-1] Failed to commit stream task 0_3 due to the following error:
org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before successfully committing offsets {example-topic-3=OffsetAndMetadata{offset=139191, leaderEpoch=null, metadata='AQAAAXU0ut1W'}}
2020-10-17 04:05:23.410 ERROR [ratelimit-transformer,,,] 19 --- [-StreamThread-1] o.a.k.s.p.internals.StreamThread         : stream-thread [stream-app-0d77ace1-4d20-413b-a71a-94cca3e620e0-StreamThread-1] Encountered the following unexpected Kafka exception during processing, this usually indicate Streams internal errors:
org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before successfully committing offsets {example-topic-3=OffsetAndMetadata{offset=139191, leaderEpoch=null, metadata='AQAAAXU0ut1W'}}
2020-10-17 04:05:24.149 ERROR [ratelimit-transformer,,,] 19 --- [-StreamThread-1] org.apache.kafka.streams.KafkaStreams    : stream-client [stream-app-0d77ace1-4d20-413b-a71a-94cca3e620e0] All stream threads have died. The instance will be in error state and should be closed.
Exception in thread "stream-app-0d77ace1-4d20-413b-a71a-94cca3e620e0-StreamThread-1" org.apache.kafka.common.errors.TimeoutException: Timeout of 60000ms expired before successfully committing offsets {example-topic-3=OffsetAndMetadata{offset=139191, leaderEpoch=null, metadata='AQAAAXU0ut1W'}}

【问题讨论】：

标签： java apache-kafka-streams spring-cloud-stream

【解决方案1】：

目前TimeoutException 没有被 Kafka Streams 很好地管理，最终会导致您的 StreamThread 崩溃。大多数情况下，由于网络问题或 kafka 不可用而发生超时。

Kafka Streams 记录的错误消息是明确的“所有流线程都已死亡。实例将处于错误状态，应该关闭。”;因此，您可以停止并重新启动 KafkaStreams 应用程序，但如果是网络问题，重新启动您的应用程序将无济于事。

请注意，有一些工作正在进行中：

您可以在 KafkaStreams 实例上配置 java.lang.Thread.UncaughtExceptionHandler 来处理该异常 (KafkaStreams JavaDoc)

【讨论】：

搜索时我确实到达了 KIP。我了解如果存在网络问题，预计会出现超时错误。我想知道流应用程序是否可以通过使用退避重试来处理故障和偶尔出现的错误，以便在连接问题得到解决后开始工作。我想我必须寻找方法来添加钩子以在发生此类错误时关闭应用程序。我想知道这里提到的“重试”配置是否有效——docs.confluent.io/current/streams/developer-guide/…
除了配置我的应用程序重新启动之外，没有其他方法可以处理这个问题。借助 spring 的执行器与 kafka 流的集成，我能够获得 kafka 流服务的 DOWN 状态，并让我的部署模块在一定次数的重试后负责重新启动 DOWN 应用程序。 cloud.spring.io/spring-cloud-stream-binder-kafka/…
retries 配置不适用于提交偏移量，但是，如果请求超时，消费者无论如何都会在内部重试。您可以增加default.api.timeout.ms，如KafkaConsumer#commitSync JavaDocs 中所述。