【发布时间】:2021-01-10 21:24:24
【问题描述】:
在挖掘了许多 SO 帖子甚至 JIRA 问题之后,我不知道该去哪里找了。 Flink 中的每个检查点都因超时而失败,在作业的异常部分显示以下错误,但作业本身并没有失败:
org.apache.kafka.common.errors.TimeoutException: Timeout expired after 600000milliseconds while awaiting InitProducerId
当禁用检查点时,有关 Kafka 的所有内容都按预期工作,所以我的假设是它可能与等待 Kafka 提交以得到确认的检查点有关(Semantic 设置为 EXACTLY_ONCE)。我记得读过有关超时不匹配导致问题的文章,因此我将 FlinkKafkaProducer 中的 TRANSACTION_TIMEOUT_CONFIG 与 900000 毫秒对齐。
我还调整了 TransactionTimeout 和 MaxBlockMS,正如 this 问题中所建议的那样,现在有很多关于这个完全相同的错误的讨论,但显然没有解决方案。
Flink 书籍“使用 Apache Flink 进行流处理”建议仔细修改 Kafka 配置,例如acks、log.flush.interval.messages、log.flush.interval.ms 和 log.flush.*。虽然我们已经在 Flink 1.9.1 下进行了这项工作,但是自从我们升级到 1.11.1 后,它就不再工作了。我不知道是否有人同时触及了 Kafka 设置,但据我所知,除了log.flush.interval=10000 之外,大部分设置都保留为默认设置。我们和以前一样使用 Confluent 5.3.3,这意味着 Kafka 2.3.1。
另外,Flink 作业部署在单节点环境中,因此它应该可以访问文件系统,整个目录归运行 Flink 服务的用户所有(在另一个 SO 线程中是建议的解决方案)。
有人知道导致这些检查点失败的原因吗?
【问题讨论】: