[jira] [Assigned] (FLINK-38697) TaskInformation blobs accumulate without cleanup causing storage exhaustion

["Matthias Pohl (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-38697?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Matthias Pohl reassigned FLINK-38697:
-------------------------------------

    Assignee:     (was: Piotr Rudnicki)

> TaskInformation blobs accumulate without cleanup causing storage exhaustion
> ---------------------------------------------------------------------------
>
>                 Key: FLINK-38697
>                 URL: https://issues.apache.org/jira/browse/FLINK-38697
>             Project: Flink
>          Issue Type: Bug
>          Components: Runtime / Coordination
>    Affects Versions: 1.14.3, 1.15.4, 1.16.3, 1.17.2, 1.18.1, 1.19.3, 1.20.3
>            Reporter: ari
>            Priority: Major
>              Labels: pull-request-available
>
> h1. Summary
> When adaptive scheduler is enabled, TaskInformation blob files accumulate on persistent storage without proper cleanup, eventually leading to storage exhaustion and job stalling.
> h1. Problem Details
> On each job deployment and restart (including JM failures), {{ExecutionJobVertex.getTaskInformationOrBlobKey() }}reconstructs TaskInformation objects from the JobGraph. These TaskInformation objects can be extremely large (>200MB) as they can contain serialized UDFs within the task configuration.
> Specifically, the Configuration field in TaskInformation includes a {{SERIALIZED_UDF}} entry.
> When these objects exceed the {{blob.offload-minsize}} configuration, they are offloaded as permanent blobs to avoid hitting RPC framesize limits. However:
>  # Blob keys are not reused across failures - each restart creates a new blob with a different key (same content hash)
>  # No cleanup mechanism until global termination - Permanent blobs are only cleaned up when the job reaches a globally terminated state (a state that doesn’t get reached during internal restarts)
>  # JobJar blobs ARE reused - In contrast, job JAR blobs stored in the JobGraph have their keys persisted and are correctly reused
>  
> h1. Impact
>  * Storage Exhaustion in storage directory (specifically a problem for high-availability storage directory since there could be hard storage limits)
>  * Job Stalling when storage limit is reached as a restart occurs but because it cant offload the blob it sends it over RPC causing it to hit the framesize limit causing checkpoints to never trigger.
>  * Particularly severe with
>  * Complex streaming jobs with large/many serialized UDFs in task config
>  * Frequent TM failures requiring restarts
>  * High parallelism (each parallelized vertex creates its on TaskInformation blob)
> h1. Reproduction Steps
>  # Enable adaptive scheduler
>  # Se {{blob.offload-minsize: 0}} (forces all TaskInformation objects to be offloaded)
>  # Run {{kubectl delete pod \{task-manager-pod-name}}} to trigger job restart
>  # Wait for job to restart and process records
>  # {{kubectl exec -it \{job-manager-pod-name} -- bash}}
>  # cd to blobstore directory and run {{ls && stat *}}
>  # Observe: Every file except the job JAR blob is duplicated after each restart
> h1. Expected vs Actual Behavior
> Expected: On a restart if content hash is the same, use the previously created task information object. However, from Flink-7140 it seems that a random key was introduced to the blob key to prevent hash-collisions. Otherwise, delete the ones that are no longer needed and then generate the new one.
> Actual: New TaskInformation blobs are created on every restart, there is no cleanup until job reaches globally terminated state, and there is unbounded blob accumulation over time



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