重放上一条成功消息中的 Kafka 主题

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

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

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

有人陷入过这些烦恼吗？谢谢。

我也有点怀疑这会如何干扰max.poll.interval.ms财产。

在活页夹中禁用重试并使用ListenerContainerCustomizer添加一个SeekToCurrentErrorHandler无限重试...

@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

这会导致查找，并且只要退避间隔不太长，就不会影响轮询间隔。

您还可以使用ExponentialBackOff.