Hi team! I have been experiencing an issue with task manager heap memory when running Flink on EMR cluster. I have a simple stateless job that consumes messages from Kafka, applies a transformation on the message using ProcessFunction and produces the output back to Kafka. When there is constant flow of messages, then there are no issues and all the metrics seem to be fine. However if there are no new messages to be consumed from Kafka and the job just sits idle, then the task manager heap usage starts to slowly creep up. There have been occasions where it creeps up so far that I get a OutOfMemoryError and the job restarts. Do anyone have any ideas on what could cause this heap memory issue when there are no messages going through the system? It does not seem to be a problem when Flink is actively processing messages and only happens when the job sits idle for a long period time. Any pointers or guesses on what could cause this would be very helpful.