[jira] [Created] (FLINK-12529) Release buffers of the record deserializer timely to improve the efficiency of heap memory usage on taskmanager

["Haibo Sun (JIRA)" <ji...@apache.org>]

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)