[FLINK-35737] Prevent Memory Leak by Closing MemoryExecutionGraphInfoStore on MiniCluster Shutdown

• Type: Bug
• Status: Open
• Priority: Critical
• Resolution: Unresolved
• Affects Version/s: 1.18.1
• Fix Version/s: 2.0.0
• Component/s: Runtime / Task
• Labels: pull-request-available
• Assignee: Unassigned
• Reporter: Feng Jiajie
• Created: 02/Jul/24 03:38
• Updated: 23/Jul/24 08:59

Description

MemoryExecutionGraphInfoStore registers a ShutdownHook upon construction and deregisters it within its close() method.

public MemoryExecutionGraphInfoStore(...) {
    ...
    this.shutdownHook = ShutdownHookUtil.addShutdownHook(this, getClass().getSimpleName(), LOG);
}

@Override
public void close() throws IOException {
    ...
    // Remove shutdown hook to prevent resource leaks
    ShutdownHookUtil.removeShutdownHook(shutdownHook, getClass().getSimpleName(), LOG);
}

Currently, MiniCluster instantiates a MemoryExecutionGraphInfoStore object but doesn't retain a reference to it, nor does it call close() during its own shutdown process.

        final DispatcherResourceManagerComponent dispatcherResourceManagerComponent =
                dispatcherResourceManagerComponentFactory.create(
                        ...
                        new MemoryExecutionGraphInfoStore(),  // -> new
                        ...); 

This behavior leads to an accumulation of ShutdownHooks when running multiple Flink jobs within the same local JVM. These accumulating hooks, along with their associated references, contribute to a memory leak.

This patch addresses the issue by ensuring that MemoryExecutionGraphInfoStore's close() method is invoked during MiniCluster shutdown.

https://github.com/apache/flink/pull/25009

Issue Links

• links to: GitHub Pull Request #25009