确保已使用REST代理从Kafka主题读取了所有消息

我是Kafka的新手，我们的团队正在研究服务间通信的模式。

目标

我们有两个服务，P（生产者）和C（消费者）。对于C所需的一组数据，P是真理的源头。C启动时，需要将所有当前数据从P加载到其缓存中，然后订阅更改通知。（换句话说，我们要在服务之间同步数据。）

数据总量相对较低，并且更改很少。短暂的同步延迟是可以接受的（最终一致性）。

我们希望解耦服务，以便P和C不需要彼此了解。

提案

当P启动时，它将所有数据发布到启用了日志压缩的Kafka主题。每个消息都是其ID为键的聚合。

C启动时，它将从主题的开头读取所有消息，并填充其缓存。然后，它继续从其偏移量读取数据，以通知更新。

当P更新其数据时，它将为已更改的聚合发布一条消息。（此消息与原始消息具有相同的架构。）

C收到新消息时，将更新其缓存中的相应数据。

约束条件

我们正在使用Confluent REST代理与Kafka进行通信。

问题

当C启动时，如何知道何时从该主题读取了所有消息，以便可以安全地开始处理？

如果C没有立即注意到P第二秒钟发送的消息，这是可以接受的。如果C在消费一个小时前发送的消息之前开始处理，这是不可接受的。请注意，我们不知道何时更新P的数据。

我们不希望C在消耗每条消息之后不必等待REST代理的轮询间隔。