[
https://issues.apache.org/jira/browse/SAMZA-2044?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16729072#comment-16729072
]
Peter Georgantas commented on SAMZA-2044:
-----------------------------------------
Hi Prateek, Xinyu.
I've been thinking about this problem a lot over the holiday and hoped I might
be about to stoke some conversation.
The {{WindowOperatorImpl}} might have a similar problem coming from the
{{handleTimer}} method. Consider a stream of data from Kafka is sent into a
Window over 5 minutes. The amount of heap memory needed is determined by how
much is polled from Kafka before being written to RocksDB during each event
loop (plus whats required to represent the window's {{triggers}} data
structure). However, if the {{task.window.ms}} isn't tuned properly, that
entire 5 minutes worth of data might be pulled from RocksDB into the heap as
the results of the {{handleTimer}} method. That link between memory and
{{task.window.ms}} was not directly obvious from the documentation.
Taking another step forward, it seems like each of `OperatorImpl`'s
`handleMessage`, `handleTimer`, `handleEndOfStream`, and `handleWatermark`
could have similar memory issues. If an agreeable solution can be found, it
would be a good idea to keep these methods consistent.
I'm very curious to hear your thoughts.
Thanks for your consideration.
-Peter
Just to give some background: I have been working with Samza for about a year
to process ad tech data. Much of my success with the framework has come from
digging into the source to better understand it's nuances.
> EOSMessage causes out of memory Exceptions related to WindowOperatorImpl
> ------------------------------------------------------------------------
>
> Key: SAMZA-2044
> URL: https://issues.apache.org/jira/browse/SAMZA-2044
> Project: Samza
> Issue Type: Bug
> Components: container
> Affects Versions: 0.14.0, 1.0
> Reporter: Peter Georgantas
> Priority: Major
>
> The contract of the handleEndOfStream method dictates that a collection of
> results be returned. In the case of WindowOperatorImpl which has a backing
> RocksDB store, this effectively causes the entirety of the store to be pulled
> into memory. In many cases, this will cause out of memory exceptions
> (otherwise why not keep the store in memory in the first place).
> Since this is a protected api, I have a relatively simple change to propose
> which could allow data to be consumed downstream as it is brought out of the
> store:
> {{protected void handleEndOfStream(Consumer<WindowPane<K, Object>> consumer,
> MessageCollector collector, TaskCoordinator coordinator)}}
> [Pull Request |https://github.com/apache/samza/pull/862]
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
