kubernetes - Apache Flink-作业部署期间重复的消息处理，以ActiveMQ作为源

Question

给定，

我有一个Flink作业，它从ActiveMQ源读取并写入mysql数据库-键入标识符。我每隔一秒钟就为该作业启用检查点。我将检查点指向Minio实例，并验证了检查点是否与jobid一起使用。我部署的工作是Openshift(位于下面的Kubernetes)-我可以根据需要按＆扩展/缩小该工作。

问题

当作业由于错误/错误而被部署(滚动)或作业中断时，并且如果ActiveMQ中有未使用的消息或Flink中有未确认的消息(但写入了数据库)，则该作业何时恢复(或新的作业)部署)作业进程已处理的消息，导致在数据库中插入重复的记录。

问题

检查点是否应帮助工作从其离开的地方恢复过来？

(滚动)部署新作业之前，我应该接受检查点吗？

如果作业因错误或集群故障而退出，会发生什么？

由于jobid在每个部署中都会不断变化，因此恢复如何进行？

编辑由于无法从数据库中获得幂等性，为了避免重复项保存到数据库中(Exactly-Once)，我可以编写特定于数据库的查询(upsert)查询以更新给定记录是否存在，如果不存在则插入？

Answer 1

JDBC当前至少支持一次，这意味着在恢复时会收到重复的消息。当前有一个草案添加了对exactly once的支持，该草案可能会在1.11中发布。

Shouldn't the checkpoints help the job recover from where it left?

是的，但是最后一次成功检查点与恢复之间的时间可能会产生观察到的重复项。我在一个有点相关的话题上做了更详细的 answer。

Should I take the checkpoint before I (rolling) deploy new job?

绝对。您实际上应该使用带有保存点的取消。这是更改拓扑的唯一可靠方法。此外，使用保存点取消可避免数据中的任何重复，因为它可以正常关闭作业。

What happens if the job quit with error or cluster failure?

它应该自动重启(取决于您的重启设置)。它将使用最新的检查点进行恢复。那肯定会导致重复。

As the jobid keeps changing on every deployment, how does the recovery happens?

通常，您通常明确指向同一检查点目录(在S3上？)。

As I cannot expect idempotency from the database, is upsert the only way to achieve Exactly-Once processing?

目前，我看不到解决方法。它应该更改为1.11。