重播上次成功消息中的Kafka主题

Question

使用Spring云流上通道的标准配置，消息重试3次，然后跳过。如果以下消息处理成功完成，则提交偏移量。这意味着在瞬态异常情况下，消息可能会丢失。

是否可以更改此行为，从而使通道卡在失败消息上，直到修复瞬态条件？

我已经尝试配置重试模板，但您必须指定多次重试，这看起来像是一个无用的参数，因为所需的行为将永远重试。

有人卷入这些麻烦吗？非常感谢。

我还怀疑这会如何干扰max.poll。间隔ms属性。

Answer 1

在活页夹中禁用重试，并使用ListenerContainerCustomizer添加具有无限重试次数的SeekTocurErrorHandler。。。

@SpringBootApplication
public class So63193500Application {

    public static void main(String[] args) {
        SpringApplication.run(So63193500Application.class, args);
    }

    @Bean
    Consumer<String> input() {
        return str -> {
            System.out.println(str);
            throw new RuntimeException("test");
        };
    }

    @Bean
    ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer() {
        return (container, dest, group) -> {
            if (group.equals("so63193500")) {
                container.setErrorHandler(new SeekToCurrentErrorHandler(
                        new FixedBackOff(5_000, FixedBackOff.UNLIMITED_ATTEMPTS)));
            }
        };
    }

}

spring.cloud.stream.bindings.input-in-0.consumer.max-attempts=1
spring.cloud.stream.bindings.input-in-0.group=so63193500

这会导致寻道，只要后退间隔不太长，就不会影响轮询间隔。

也可以使用指数衰减。