[FLINK-21413] TtlMapState and TtlListState cannot be clean completely with Filesystem StateBackend

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.9.0
• Fix Version/s: 1.13.0
• Component/s: Runtime / State Backends
• Labels: pull-request-available
• Assignee: Jiayi Liao
• Reporter: Jiayi Liao
• Created: 19/Feb/21 03:13
• Updated: 28/Aug/21 11:11
• Resolved: 31/Mar/21 10:22

Description

Take the #TtlMapState as an example,

 

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;
}

 

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.

 

Issue Links

• links to: GitHub Pull Request #15016