java - Flink State过期时触发

Question

我有一个有趣的用例，我想用 Flink 进行测试。我有一个传入的 Message 流，它是 PASS 或 FAIL。现在，如果消息的类型为 FAIL，我有一个下游 ProcessFunction，它保存 Message 状态，然后发送 pause > 对依赖于此的所有内容发出命令。当我收到与之前收到的 FAIL 相关的 PASS 消息(通过消息 ID 键入)时，我向所有内容发送 resume 命令我之前就暂停了。

现在，我计划使用 State TTL 来使存储的 FAIL 状态过期，并在一定的超时后恢复所有内容，即使我没有收到具有相同内容的 PASS 消息消息 ID。这可以单独使用 Flink 来完成吗？还是需要一些外部计时器来向我的程序发送超时消息？

为了让它在 Flink 中运行，我想到了这样的事情:

对于每个消息，添加时间戳并将其传递给一个处理函数，该函数会等到current_ts - timestamp == timeout，然后再发送它以恢复由消息暂停的所有内容模块。有更好的方法吗？或者你们认为这样可以吗？

Answer 1

似乎使用计时器来使状态过期(通过在 onTimer 方法中调用 state.clear() )比使用状态 TTL 更直接。同样的 onTimer 方法也可以安排事情同时恢复。