[jira] [Created] (FLINK-33192) State memory leak in the Window Operator due to unregistered cleanup timer

["Vidya Sagar Kalvakunta (Jira)" <ji...@apache.org>]

Vidya Sagar Kalvakunta created FLINK-33192:
----------------------------------------------

             Summary: State memory leak in the Window Operator due to unregistered cleanup timer
                 Key: FLINK-33192
                 URL: https://issues.apache.org/jira/browse/FLINK-33192
             Project: Flink
          Issue Type: Bug
            Reporter: Vidya Sagar Kalvakunta


I have encountered a state memory leak issue in the default window operator. The cleanup timer for a window is not registered when it does not emit a result if it’s fired immediately after creation. The window is already added to the window state.



*Steps to Reproduce:*
 # Write a custom trigger that triggers for every element.
 # Write a custom aggregate function that never produces a result.
 # Use a default tumbling event time window with this custom trigger and aggregate function.
 # Publish events spanning multiple time windows.
 # The window state will contain all the windows even after their expiry/cleanup time.


[Window Operator bug|https://github.com/apache/flink/blob/cd95b560d0c11a64b42bf6b98107314d32a4de86/flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/windowing/WindowOperator.java#L399C18-L417]
{code:java}
windowState.add(element.getValue());

if (triggerResult.isFire()) {
    ACC contents = windowState.get();
    if (contents == null) {
        continue;
    }
    emitWindowContents(window, contents);
}

if (triggerResult.isPurge()) {
    windowState.clear();
}
registerCleanupTimer(window);{code}
 

*Expected Result:*

The cleanup timer should be registered for every window that's added to the window state even when it does not emit a result after it’s fired.

*Actual Result:*

The cleanup timer is not registered for a window when it does not emit a result after it’s fired, causing the window state that is already created to live on indefinitely.

*Impact:*

This issue led to a huge state memory leak in our applications and was very challenging to identify.

 

*Fix:*

There are two ways to fix this issue. I'm willing to create a PR with the fix if approved. 
 # Register the cleanup timer immediately after a window is added to the state.

{code:java}
windowState.add(element.getValue());
registerCleanupTimer(window);{code}

 # Emit the results when the contents are not null and remove the continue statement.

{code:java}
if (triggerResult.isFire()) {
    ACC contents = windowState.get();
    if (contents != null) {         
        emitWindowContents(window, contents);
    }
} {code}

 

 

 

 



--
This message was sent by Atlassian Jira
(v8.20.10#820010)