Haibo Sun created FLINK-12529:
---------------------------------
Summary: Release buffers of the record deserializer timely to improve the efficiency of heap memory usage on taskmanager
Key: FLINK-12529
URL:
https://issues.apache.org/jira/browse/FLINK-12529 Project: Flink
Issue Type: Improvement
Affects Versions: 1.8.0
Reporter: Haibo Sun
Assignee: Haibo Sun
In input processors (`StreamInputProcessor` and `StreamTwoInputProcessor`), each input channel has a corresponding record deserializer. Currently, these record deserializers are cleaned up at the end of the task (look at `StreamInputProcessor#cleanup()` and `StreamTwoInputProcessor#cleanup()`). This is not a problem for unbounded streams, but it may reduce the efficiency of heap memory usage on taskmanger when input is bounded stream.
For example, in case that all inputs are bounded streams, some of them end very early because of the small amount of data, and the other end very late because of the large amount of data, then the buffers of the record deserializers corresponding to the input channels finished early is idle for a long time and no longer used.
In another case, when both unbounded and bounded streams exist in the inputs, the buffers of the record deserializers corresponding to the bounded stream are idle for ever (no longer used) after the bounded streams are finished. Especially when the record and the parallelism of upstream are large, the total size of `SpanningWrapper#buffer` are very large. The size of `SpanningWrapper#buffer` is allowed to reach up to 5 MB, and if the parallelism of upstream is 100, the maximum total size will reach 500 MB (in our production, there are jobs with the record size up to hundreds of KB and the parallelism of upstream up to 1000).
Overall, after receiving `EndOfPartitionEvent` from the input channel, the corresponding record deserializer should be cleared immediately to improve the efficiency of heap memory usage on taskmanager.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
Locked
