apache-flink - Flink 有状态函数 - org.apache.flink.statefun.sdk.Context::sendAfter 中的不可靠行为

Question

我们使用 SF(有状态函数)开发了一种类似超时的机制，其中我们有一个 TimeoutManagerFunction，它保持和管理由其他函数设置的当前超时状态，并在超时时发回过期消息.

为了实现该行为，我们的 TimeoutManagerFunction 在 X 秒后向自己发送一条消息。 (X表示timeout到实际超时需要经过的秒数，初始设置时从timeout setter函数传过来的)

超时 setter 函数还可以通过向 TimeoutManagerFunction 发送更新消息来更新超时的到期时间，或者可以取消它。 TimeoutManagerFunction 根据收到的消息编辑过期时间或取消它。

我们目前有 10 种不同的函数与此 TimeoutManagerFunction 交互，其中一些几乎每秒都在设置/更新/取消超时。

因此，在 TimeoutManagerFunction 的 set()、update() 和 cancel() 方法上，我们基本上执行如下操作:

context.sendAfter(Duration.ofMillis(cancelableTimeout.getTimeoutDuration()), context.self(), selfTimeout);

并且在 TimeoutManagerFunction 的 timeout() 方法上，我们检查给定的超时是否确实已过期(因为它的过期日期有可能同时更新):

protected void timeout(Context context, SelfTimeout selfTimeout) {
        try {
            // **timeouts** is the internal state of timeouts in TimeoutManagerFunction
            CancelableTimeout cancelableTimeout = timeouts.get(selfTimeout.getId());
            if (cancelableTimeout != null) {
                if (Instant.now().isAfter(Instant.ofEpochMilli(cancelableTimeout.getExpiresAt()))) {
                    /// send back timed out message back to initial timeout setter function...

我们遇到的问题基本上是如果我们执行类似 context.sendAfter(10 秒，超时对象) 的操作，有时需要超过 10 秒才能收到该消息。任何关于如何或为什么会发生这种情况的想法都将受到赞赏。

我们的SF版本是2.2.1，Flink版本为1.11.6

Answer 1

无法保证所要求的时间会得到严格遵守。如果集群相对于正在完成的处理量而言配置不足，那么某些计时器很可能会延迟触发。