如何使用Kafka来传递状态？

我们有一个Flink Job需要一些自定义的规则，希望能够动态添加、更新、删除。规则的数量在百到千条。目前设计的结构是RDB+ Kafka + Flink。

RDB存储规则的完整快照，以展示给Web应用作增删改查。改动通过Kafka发送消息至Flink，通过BroadcastState传播规则。

目前有一个问题没有解决：如何使用Kafka来传递状态。我想了一下，大概有几种方案：

1. 消息标记Add、Upadte、Delete类型，在Flink中写逻辑来处理状态以和RDB中状态保持一致。 目前的问题是，每次重启Job，都需要从头读Kafka，来回放状态的更新。Kafka中的状态消息也需要持久保存。担心长期会堆积很多消息。
2. 使用Kafka Compact Log来解决1的问题。这个方案主要是之前没有使用过Compact Log，不清楚会不会有坑。 3.使用方案1，但是启动时Flink从RDB拉取全量规则。
3. 规则更新后Kafka消息发送全量规则，启动时Flink只拉取最新一条消息。

各位大佬是否有经验可以分享，怎么处理是比较科学的？不胜感激！

*来自志愿者整理的flink邮件归档