InvalidPidMappingException 导致 kafka 流应用程序关闭答案

【问题标题】：InvalidPidMappingException causes kafka stream application to shut downInvalidPidMappingException 导致 kafka 流应用程序关闭
【发布时间】：2022-01-06 16:25:52
【问题描述】：

我有这个用 kafka 流编写的应用程序。它时不时地抛出 InvalidPidMappingException。

Caused by: org.apache.kafka.common.KafkaException: org.apache.kafka.common.errors.InvalidPidMappingException: The producer attempted to use a producer id which is not currently assigned to its transactional id.

我有这段代码，它为应用程序设置未捕获的异常处理程序

streams.setUncaughtExceptionHandler(
            (Thread thread, Throwable exp) -> {
              log.error("Unhandled exception in thread with name ", exp);
                SpringApplication.exit(applicationContext, () -> 1);
            }
    );

我了解当协调器在未收到任何交易状态更新后使生产者的交易 ID 到期时，会发生此异常。关于这个例外，我有几个问题：

我以为在这个异常之后，生产者会重试将其事务 id 与协调器同步并恢复，而不会导致 kafka 流线程被杀死。即使我将上面的代码 sn-p 更改为在 InvalidPidMappingException 上不退出 Spring 应用程序，它仍然会杀死流线程。有没有办法避免 InvalidPidMappingException 上的流线程死亡？当存在 UnknownProducerIdException 时，我已经看到了所需的行为。还是我在这里遗漏了什么？
除了事务 ID 过期之外，是否还有其他原因导致此异常发生？
为什么 InvalidPidMappingException 的处理方式不同于 UnknownProducerIdException？前者杀死了Stream Tread，后者恢复得很好。

我正在使用以下版本的库：

spring-kafka-version = '2.5.5.RELEASE'
apache-kafka-clientVersion = '2.5.1'
confluent-version = '5.4.2'

【问题讨论】：

标签： apache-kafka apache-kafka-streams

【解决方案1】：

晚了几个月，但discussion on this Apacha Kafka Jira 很有帮助。总而言之，使用 Apache Kafka 流库 v2.8.0 及更高版本，您可以使用 org.apache.kafka.streams.KafkaStreams 中新的 setUncaughtExceptionHandler() 方法来处理任何未捕获的异常并通过终止当前线程并通过返回 org.apache.kafka.streams.errors.StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD 创建一个新线程以供将来处理来保持流运行。例如

kafkaStreams.setStreamsUncaughtExceptionHandler(e -> {
    return StreamsUncaughtExceptionHandler.StreamThreadExceptionResponse.REPLACE_THREAD;
});

Javadocs for KafkaStreams::setUncaughtExceptionHandler method added in 2.8.0

【讨论】：