使用 spring 云流 kafka 绑定消费者和 DLQ 启用覆盖重试尝试

Question

我正在使用 spring 云流与 kafka 绑定。使用 spring.cloud.stream.kafka.bindings.function 配置并启用 DLQ 来配置消费者。

spring.cloud.stream.kafka.bindings.consumeTest-in-0:
    consumer:
        batch-mode: false
        ackEachRecord: true
        dlqName: test.file.error
        enableDlq: true
        dlqProducerProperties.configuration.key.serializer: org.apache.kafka.common.serialization.StringSerializer
        dlqProducerProperties.configuration.value.serializer: org.apache.kafka.common.serialization.StringSerializer
        maxAttempts: 5
        backOffInitialInterval: 5000
        backOffMaxInterval: 5000

当我消费一个主题的消息时，出现异常，我看到它重试了3次，并将消息发送到DLQ。我已将 maxAttempts 配置为 5，但我无法覆盖默认值 3。

我正在使用 spring kafka (2.7.8) 和 spring cloud (2020.0.4)。如何覆盖重试尝试和退避间隔参数？

我注意到如果我在默认消费者级别（如下所示）添加参数，它们会被选中，但不会在函数绑定级别：

spring.cloud.stream.default.consumer:
    maxAttempts: 5
    defaultRetryable: true
    backOffInitialInterval: 5000
    backOffMaxInterval: 5000
    backOffMultiplier: 1

谢谢

Answer 1

把它放在spring.cloud.stream.bindings.consumeTest-in-0.consumer.maxAttempts=5下面，而不是你的spring.cloud.stream.kafka.bindings.consumeTest-in-0.consumer.maxAttempts=5。

看这里：Common Properties

使用 spring 云流 kafka 绑定消费者和 DLQ 启用覆盖重试尝试

Override retry attempts with spring cloud stream kafka bindings consumer and DLQ enabled

spring-boot

spring-cloud-stream

spring-kafka