[ https://issues.apache.org/jira/browse/FLINK-21413?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17287862#comment-17287862 ]
Yu Li commented on FLINK-21413: ------------------------------- Checking the [state TTL FLIP document|https://cwiki.apache.org/confluence/display/FLINK/FLIP-25%3A+Support+User+State+TTL+Natively#FLIP25:SupportUserStateTTLNatively-TTLbehaviour] I cannot find description on whether TTL for a whole map is supported for {{MapState}}, but according to the current implementation the answer is no (TTL is only checked against value of each map entry). What's more, in [HeapMapState#remove|https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapMapState.java#L130-L132] we could see the whole map will be removed if become empty, so I don't think {{TtlIncrementalCleanup}} need to take care of the empty map case. Accordingly, I think we should have a fast path in {{TtlMapState#getUnexpiredOrNull}} to check whether {{ttlValue}} is empty and return it directly (instead of returning {{NULL}}) if so, and returning {{NULL}} iif {{ttlValue}} is not empty but all values expired ({{unexpired.size()}} is zero). And similar logic should be applied to {{TtlListState#getUnexpiredOrNull}}. Please let me know your thoughts [~wind_ljy]. Thanks. > TtlMapState and TtlListState cannot be clean completely with Filesystem > StateBackend > ------------------------------------------------------------------------------------ > > Key: FLINK-21413 > URL: https://issues.apache.org/jira/browse/FLINK-21413 > Project: Flink > Issue Type: Bug > Components: Runtime / State Backends > Affects Versions: 1.9.0 > Reporter: Jiayi Liao > Priority: Major > Attachments: image-2021-02-19-11-13-58-672.png > > > Take the #TtlMapState as an example, > > {code:java} > public Map<UK, TtlValue<UV>> getUnexpiredOrNull(@Nonnull Map<UK, > TtlValue<UV>> ttlValue) { > Map<UK, TtlValue<UV>> unexpired = new HashMap<>(); > TypeSerializer<TtlValue<UV>> valueSerializer = > ((MapSerializer<UK, TtlValue<UV>>) > original.getValueSerializer()).getValueSerializer(); > for (Map.Entry<UK, TtlValue<UV>> e : ttlValue.entrySet()) { > if (!expired(e.getValue())) { > // we have to do the defensive copy to update the > value > unexpired.put(e.getKey(), > valueSerializer.copy(e.getValue())); > } > } > return ttlValue.size() == unexpired.size() ? ttlValue : unexpired; > } > {code} > > The returned value will never be null and the #StateEntry will exists > forever, which leads to memory leak if the key's range of the stream is very > large. Below we can see that 20+ millison uncleared TtlStateMap could take up > several GB memory. > > !image-2021-02-19-11-13-58-672.png! -- This message was sent by Atlassian Jira (v8.3.4#803005)