Collections 的 Flink State TTL 过期

Question

感谢您花时间阅读本文，想请教各位高手关于 Flink 1.8.0 中的 Flink 状态 TTL 特性，看完后 这个，对我来说还是很模糊的。

https://ci.apache.org/projects/flink/flink-docs-release-1.8/dev/stream/state/state.html#state-time-to-live-ttl

我想确定在哪里启用了 TTL 功能，是在键字段上还是在值字段上。特别是，假设我有一个这样的 mapState 结构：

mapState = Map[String,List[String]]
e.g. val mapState = Map("haha" -> List("foo","bar")) in Scala
where "haha" is the key of the mapState and List("foo","bar") is the value

如果我要通过 StateTtlConfig 在 mapState 上设置 1 分钟的 TTL，然后立即（不到 1 分钟）写入 List 中的一个值，例如“foo”。

那么1分钟后，当TTL触发时，是key“haha”过期还是value“bar”过期？

换句话说，如果它要在 key 上过期，我的理解是 mapState 将保持不变

mapState = Map("haha" -> List("foo","bar"))

因为写入值“foo”会重置键上的 TTL，因此整个 mapState 保持不变

另一种情况是，如果它在值上过期，mapState 将变为

mapState = Map("haha" -> List("foo"))

因为值“bar”会在 1 分钟后过期而不会被访问。

希望我把问题说清楚了，提前感谢您提供任何形式的帮助。

Answer 1

通过 TTL 访问状态的代码如下：

<SE extends Throwable, CE extends Throwable, CLE extends Throwable, V> TtlValue<V> getWrappedWithTtlCheckAndUpdate(
        SupplierWithException<TtlValue<V>, SE> getter,
        ThrowingConsumer<TtlValue<V>, CE> updater,
        ThrowingRunnable<CLE> stateClear) throws SE, CE, CLE {
        TtlValue<V> ttlValue = getter.get();
        if (ttlValue == null) {
            return null;
        } else if (expired(ttlValue)) {
            stateClear.run();
            if (!returnExpired) {
                return null;
            }
        } else if (updateTsOnRead) {
            updater.accept(rewrapWithNewTs(ttlValue));
        }
        return ttlValue;
    }

验证值是否过期的方法如下：

    static <V> boolean expired(@Nullable TtlValue<V> ttlValue, long ttl, long currentTimestamp) {
        return ttlValue != null && expired(ttlValue.getLastAccessTimestamp(), ttl, currentTimestamp);
    }

这基本上意味着在这种情况下，它将检查整个列表的 TTL，而不是单独的元素。因此，取决于StateTtlConfig，整个列表将过期或整个列表不会过期。

可用的 TTL 配置是 OnReadAndWrite 和 OnCreateAndWrite。所以，基本上为了保持一致，如果你想更新你的值列表，你需要在MapState 上使用put()。