Spring cloud stream kafka绑定器以编程方式将kafka主题（多个分区）偏移量重置为任意数字

Question

我需要将偏移量重置为一个数字。

详细要求：我的应用程序正在使用来自 kafka 主题的消息并将其转储到 DB 中，假设 DB 在处理已使用的消息时出现故障（offset=10），直到数据库关闭时应用程序使用的消息，直到偏移量 20。

现在DB在处理第20个偏移量消息时再次出现，现在我想再次将偏移量重置为10，以便将数据保存在数据库中。

我可以通过编程（Spring启动）来实现吗？我正在使用Spring云流活页夹Kafka。

Answer 1

如果数据库已关闭，为什么还要继续接收消息？

只需将侦听器容器配置为继续重试当前记录，直到数据库再次可用。

@Bean
ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> listenerContainerCustomizer() {
    return (container, topic, group) -> {
        container.setErrorHandler(...);
    };
}

看见https://docs.spring.io/spring-kafka/docs/current/reference/html/#seek-到当前