Spring Cloud Stream 3.x - 重播消息策略

Question

我正在寻找有关使用 Spring Cloud Stream 3.x / Kafka binder implementations 从 Kafka 主题重播消息策略的一些指导 -

1. 重播特定消息[例如。通过时间戳窗口]来自一个主题。如何重置消费者组中所有或部分消费者的偏移量？

2. 是否可以从主题的特定分区重播[如果我们知道我们有兴趣重播的消息的分区]？

一般来说，关于消息重播的最佳做法是什么。感谢您的宝贵时间。

Answer 1

添加一个重新平衡侦听器 bean，它将被连接到 binder...

@Bean
KafkaBindingRebalanceListener rebal() {
    return new KafkaBindingRebalanceListener() {

        @Override
        public void onPartitionsAssigned(String bindingName, Consumer<?, ?> consumer,
                Collection<TopicPartition> partitions, boolean initial) {

            consumer.seekToBeginning(partitions);
        }

    };
}

您可以使用任何消费者查找操作；您也可以拨打consumer.offsetsForTimes(...)等。

initial 标志对于第一次重新平衡为 true，对于其他标志为 false。