Spring Kafka：如何在执行 seek() 后丢弃 poll() 已检索到的消息？

Question

这是对 - Reading the same message several times from Kafka 的后续问题。如果有更好的方法可以在不发布新问题的情况下提出这个问题，请告诉我。在这篇文章中，Gary 提到了

“但是如果它们已经被检索到，你仍然会首先看到后面的消息，所以你也必须丢弃它们。”

在调用 seek() 之后，是否有一种干净的方法可以丢弃 poll() 已经读取的消息？我开始通过保存初始偏移量（在recordOffset中）来实现逻辑，并在成功时增加它。失败时，我调用 seek() 并将 recordOffset 的值设置为 record.offset()。然后对于每条新消息，我都会检查 record.offset() 是否大于 recordOffset。如果是，我只需调用确认（），从而“丢弃”所有先前读取的消息。这是代码 -

    // in onMessage()...
    if (record.offset() > recordOffset){
        acknowledgment.acknowledge();
        return;
    }

    try {
        processRecord(record);
        recordOffset = record.offset()+1;
        acknowledgment.acknowledge();
    } catch (Exception e) {
        recordOffset = record.offset();
        consumerSeekCallback.seek(record.topic(), record.partition(), record.offset());
    }

这种方法的问题在于它会因多个分区而变得复杂。有没有更简单/更清洁的方法？

编辑 1 根据下面 Gary 的建议，我尝试添加这样的 errorHandler -

@KafkaListener(topicPartitions =
        {@org.springframework.kafka.annotation.TopicPartition(topic = "${kafka.consumer.topic}", partitions = { "1" })},
        errorHandler = "SeekToCurrentErrorHandler")

当我收到“无法解析方法'errorHandler'”时，此语法是否有问题？

编辑 2 在 Gary 解释了 2 个错误处理程序之后，我删除了上面的 errorHandler 并在下面添加到配置文件中 -

@Bean
public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
    factory.setConsumerFactory(new DefaultKafkaConsumerFactory<>(kafkaProps()));
    factory.getContainerProperties().setAckOnError(false);
    factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
    factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
    return factory;
}

当我启动应用程序时，我现在收到此错误...

java.lang.NoSuchMethodError: org.springframework.util.Assert.state(ZLjava/util/function/Supplier;)V 在 org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter.determineInferredType(MessagingMessageListenerAdapter.java:396)

这是第 396 行 -

Assert.state(!this.isConsumerRecordList || validParametersForBatch,
            () -> String.format(stateMessage, "ConsumerRecord"));
Assert.state(!this.isMessageList || validParametersForBatch,
            () -> String.format(stateMessage, "Message<?>"));

Answer 1

从2.0.1版本开始，如果容器的ErrorHandler是RemainingRecordsErrorHandler，比如SeekToCurrentErrorHandler，剩下的记录（包括失败的记录）将被发送到错误处理程序而不是监听器。

这允许SeekToCurrentErrorHandler 重新定位每个分区，以便下一次轮询将返回未处理的记录。

/**
 * An error handler that seeks to the current offset for each topic in the remaining
 * records. Used to rewind partitions after a message failure so that it can be
 * replayed.
 *
 * @author Gary Russell
 * @since 2.0.1
 *
 */
public class SeekToCurrentErrorHandler implements RemainingRecordsErrorHandler 

编辑

有两种类型的错误处理程序。 KafkaListenerErrorHandler（在注释中指定）在侦听器级别工作；它连接到调用 @KafkaListener 注释的侦听器适配器，因此只能访问当前记录。

第二个错误处理程序（在侦听器容器上配置）在容器级别工作，因此可以访问剩余的记录。 SeekToCurrentErrorHandler 是容器级错误处理程序。

在容器工厂的容器属性上配置...

@Bean
public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
    factory.setConsumerFactory(this.consumerFactory);
    factory.getContainerProperties().setAckOnError(false);
    factory.getContainerProperties().setErrorHandler(new SeekToCurrentErrorHandler());
    factory.getContainerProperties().setAckMode(AckMode.RECORD);
    return factory;
}

Answer 2

你走对了，是的，你还必须处理不同的分区。有FilteringMessageListenerAdapter，但还是要自己写逻辑。