[FLINK-2004] Memory leak in presence of failed checkpoints in KafkaSource

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 0.9
• Fix Version/s: 0.9
• Assignee: Robert Metzger
• Reporter: Stephan Ewen
• Created: 12/May/15 19:56
• Updated: 28/Feb/19 11:15
• Resolved: 01/Jun/15 15:58

Description

Checkpoints that fail never send a commit message to the tasks.

Maintaining a map of all pending checkpoints introduces a memory leak, as entries for failed checkpoints will never be removed.

Approaches to fix this:

• The source cleans up entries from older checkpoints once a checkpoint is committed (simple implementation in a linked hash map)
• The commit message could include the optional state handle (source needs not maintain the map)
• The checkpoint coordinator could send messages for failed checkpoints?