【问题标题】:Proper error handling for kafka batch listenerkafka 批处理侦听器的正确错误处理
【发布时间】:2022-01-13 06:05:35
【问题描述】:

我正在处理错误处理实施,并有一个问题。
让我解释一下问题:
我收到一批消息,我在 for 循环中对每个消息进行数据库查找,然后我需要收集列表中所有查找的对象,并使用此对象列表调用批量插入存储过程和批量更新存储过程。

现在让我们假设在查找过程中发生了一些异常。我想重试这条消息。对于这种情况,我尝试使用DefaultErrorHandler。但是有一个问题,根据文档,当抛出带有元素索引的 BatchListenerFailedException 时,它会提交索引之前记录的偏移量。但正如我所说,我需要在查找后执行批量插入和更新,所以我不想在索引之前提交偏移量,这些记录还没有插入/更新到数据库中。

这是否意味着我唯一的选择是使用RetryingBatchErrorHandler 每次都重试整个批次?我可以以某种方式继续处理不产生错误的消息吗?

另外,如果RetryingBatchErrorHandler 是唯一的选择,我怎么能确定在较长的退避期(指数退避)的情况下,kafka 不会杀死我的消费者并且不会启动重新平衡?

我目前的实现:

RetryingBatchErrorHandler retryingBatchErrorHandler =
                new RetryingBatchErrorHandler(backoff,
                        (consumerRecord, e) ->
                                log.error("Backoff attempts exhausted for the record with offset={}, partition={}, value={}, offset committed.",
                consumerRecord.offset(), consumerRecord.partition(), consumerRecord.value()));

factory.setBatchErrorHandler(retryingBatchErrorHandler);

更新:请参阅 Artem 答案中的 cmets。
这就是如何将查找步骤包装到 retryTemplate

LookedUpRequest lookedUpRequest = retryTemplate.execute(ctx -> {
   //Lookup step
   return lookup.process(request);
});

如果它失败了,那么它将进一步为批处理错误处理程序抛出异常,RetryingBatchErrorHandler 根据其策略重试批处理

【问题讨论】:

    标签: java spring spring-boot apache-kafka spring-kafka


    【解决方案1】:

    查看它的 JavaDocs:

    /**
     * A batch error handler that invokes the listener according to the supplied
     * {@link BackOff}. The consumer is paused/polled/resumed before each retry in order to
     * avoid a rebalance. If/when retries are exhausted, the provided
     * {@link ConsumerRecordRecoverer} is invoked for each record in the batch. If the
     * recoverer throws an exception, or the thread is interrupted while sleeping, seeks are
     * performed so that the batch will be redelivered on the next poll.
     *
     * @author Gary Russell
     * @since 2.3.7
     *
     */
    public class RetryingBatchErrorHandler extends KafkaExceptionLogLevelAware
            implements ListenerInvokingBatchErrorHandler {
    

    因此,没有重新平衡,因为消费者在两者之间暂停。

    您可能需要考虑为您的查找部分缓存一些东西,这样您就不会在失败之前对您已经请求的那些记录施加压力。

    您也可以考虑自己重试该查找。见RetryTemplate。但在这种情况下,您需要确保整个操作的时间不够长(请参阅max.poll.interval.ms)以使消费者离开其组。

    【讨论】:

    • 我看到 RetryTemplate 在最新版本中已被弃用,所以我只是使用带有指数退避的 RetryingBatchErrorHandler 来重试整个批次。看来这是我唯一的选择了。遗憾的是 RetryingBatchErrorHandler 不支持异常分类。我添加了我现在如何实现它的代码。感谢您确认消费者不会被杀死。
    • RetryTemplate 未被弃用。只是它的功能被 Kafka 侦听器容器中的 RetryingBatchErrorHandler 取代。您错过了我的意思:我建议您直接在代码中使用RetryTemplate.execute(),以避免将异常推送回 Kafka 侦听器容器。但这仅适用于查找功能:我不会自己重试插入和更新...
    • 现在我明白了,我用 retryTemplate.execute 包装了查找操作,谢谢!但是为什么你不建议包装插入和更新呢?
    • 因为你自己说的:整批都是单次操作。所以,让 Kafka 容器用它自己的重试功能将它作为一个整体带回给你。
    • 但在这种情况下,它必须再次进行查找
    猜你喜欢
    • 1970-01-01
    • 2019-12-10
    • 2021-09-13
    • 2021-08-31
    • 2022-11-11
    • 2021-02-03
    • 2021-08-31
    • 1970-01-01
    • 2023-04-10
    相关资源
    最近更新 更多