Lauri Suurväli created FLINK-33231: --------------------------------------
Summary: Memory leak in KafkaSourceReader if no data in consumed topic Key: FLINK-33231 URL: https://issues.apache.org/jira/browse/FLINK-33231 Project: Flink Issue Type: Bug Components: Connectors / Kafka Affects Versions: 1.17.1 Reporter: Lauri Suurväli Attachments: Screenshot 2023-10-09 at 13.02.34.png, Screenshot 2023-10-09 at 14.13.43.png, Screenshot 2023-10-10 at 12.49.37.png *Problem description* Our Flink streaming job TaskManager heap gets full when the job has nothing to consume and process. It's a simple streaming job with KafkaSource -> ProcessFunction -> KafkaSink. When there are no messages in the source topic the TaskManager heap usage starts increasing until the job exits after receiving a SIGTERM signal. We are running the job on AWS EMR with YARN. The problems with the TaskManager heap usage do not occur when there is data to process. It's also worth noting that sending a single message to the source topic of a streaming job that has been sitting idle and suffers from the memory leak will cause the heap to be cleared. However it does not resolve the problem since the heap usage will start increasing immediately after processing the message. !Screenshot 2023-10-10 at 12.49.37.png! TaskManager heap used percentage is calculated by {code:java} flink.taskmanager.Status.JVM.Memory.Heap.Used * 100 / flink.taskmanager.Status.JVM.Memory.Heap.Max{code} I was able to take heap dumps of the TaskManager processes during a high heap usage percentage. Heap dump analysis detected 912,355 instances of java.util.HashMap empty collections retaining >= 43,793,040 bytes. !Screenshot 2023-10-09 at 14.13.43.png! The retained heap seemed to be located at: {code:java} org.apache.flink.connector.kafka.source.reader.KafkaSourceReader#offsetsToCommit{code} !Screenshot 2023-10-09 at 13.02.34.png! *Possible hints:* An empty HashMap is added during the snapshotState method to offsetsToCommit map if it does not already exist for the given checkpoint. [KafkaSourceReader line 107|https://github.com/apache/flink-connector-kafka/blob/b09928d5ef290f2a046dc1fe40b4c5cebe76f997/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java#L107] {code:java} Map<TopicPartition, OffsetAndMetadata> offsetsMap = offsetsToCommit.computeIfAbsent(checkpointId, id -> new HashMap<>()); {code} If the startingOffset for the given split is >= 0 then a new entry would be added to the map from the previous step. [KafkaSourceReader line 113|https://github.com/apache/flink-connector-kafka/blob/b09928d5ef290f2a046dc1fe40b4c5cebe76f997/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java#L113] {code:java} if (split.getStartingOffset() >= 0) { offsetsMap.put( split.getTopicPartition(), new OffsetAndMetadata(split.getStartingOffset())); }{code} If the starting offset is smaller than 0 then this would leave the offsetMap created in step 1 empty. We can see from the logs that the startingOffset is -3 when the splits are added to the reader. {code:java} Adding split(s) to reader: [[Partition: source-events-20, StartingOffset: 1, StoppingOffset: -9223372036854775808], [Partition: source-events-44, StartingOffset: -3, StoppingOffset: -9223372036854775808], [Partition: source-events-12, StartingOffset: -3, StoppingOffset: -9223372036854775808], [Partition: source-events-36, StartingOffset: 1, StoppingOffset: -9223372036854775808], [Partition: source-events-4, StartingOffset: -3, StoppingOffset: -9223372036854775808], [Partition: source-events-28, StartingOffset: -3, StoppingOffset: -9223372036854775808]]{code} The offsetsToCommit map is cleaned from entries once they have been committed to Kafka which happens during the callback function that is passed to the KafkaSourceFetcherManager.commitOffsets method in KafkaSourceReader.notifyCheckpointComplete method. However if the committedPartitions is empty for the given checkpoint, then the KafkaSourceFetcherManager.commitOffsets method returns. [KafkaSourceFetcherManager line 78|https://github.com/apache/flink-connector-kafka/blob/b09928d5ef290f2a046dc1fe40b4c5cebe76f997/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/fetcher/KafkaSourceFetcherManager.java#L78] {code:java} if (offsetsToCommit.isEmpty()) { return; } {code} We can observe from the logs that indeed an empty map is encountered at this step: {code:java} Committing offsets {}{code} *Conclusion* It seems that an empty map gets added per each checkpoint to offsetsToCommit map. Since the startingOffset in our case is -3 then the empty map never gets filled. During the offset commit phase the offsets for these checkpoints are ignored, since there is nothing to commit, however there isn't any cleanup either so the empty maps keep accumulating. -- This message was sent by Atlassian Jira (v8.20.10#820010)