使用偏移量/时间戳将信息从一个Kafka主题复制到另一个

Question

对于一些数据处理，我们需要重新处理两个时间戳之间的所有消息，比如1月1日至1月15日。

为了控制上限，我们计划创建一个具有这些消息的新主题，这样一旦任务完成，我们也可以删除该主题。新主题将具有来自源主题的特定偏移量的数据。

分区1-从偏移量100

分区2-从偏移量2400.诸若此类

对此最合适的解决方案是什么？其中大约有10条this消息。

Answer 1

1. 从源主题创建一个使用者。
2. 调用.assign用于要为这些分区的每个起始偏移量复制
3. Call .seek的分区。您可以使用method.
4. Create方法获取特定时间戳；然后，您可以将这些消息传递给Producer
5. Start Producer
6. Start一个轮询循环(理想情况下，每个线程都带有创建的producer).
7. As轮询的引用)，如果记录时间戳超过正在读取的日期，检查记录
   • 的时间戳，停止轮询循环/线程
   • ，通过生产者将数据发送到输出topic

。