在Kafka中寻求操作

Question

我有一个场景做一些管理任务与Kafka。

现在管理员想，

例1：将app2的偏移量从当前偏移量位置（X）向前移动5个位置。对于此任务，是否应该停止app2并创建一个KafkaConsumer（不是spring)，其组id与app2相同，即“Group2”，并提交新的偏移量x+5并启动app2？

案例2：将主题A的所有应用程序的偏移量从它们当前的偏移量位置(X)向前移动5个位置，假设所有应用程序都处于相同的偏移量X。对于这个任务，是否应该停止所有应用程序并创建一个具有空组id“”的KafkaConsumer，提交新的偏移量X+5并启动所有应用程序？或者对每个应用程序单独执行类似于案例-1的操作？

或者还有什么更好的方法来处理Kafka？

Answer 1

ConsumerSeekAware对您没有帮助吗？

为了查找，您的侦听器必须实现ConsumerSeekAware，它具有以下方法：

void registerSeekCallback(ConsumerSeekCallback callback);

void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

void onIdleContainer(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback);

若要在运行时任意搜索，请使用RegisterSeekCallback对适当线程的回调引用。

https://docs.spring.io/spring-kafka/docs/2.0.1.release/reference/html/_reference.html#seek