Assuming it is flink batch processing. Check the size of the input and how the operator chained. Batch job stores intermediate state in the /tmp. Once the job is completed, Flink cleans up the blob.

There are Two types of blobs:

1. Transient blobs: Temporary files (cleaned up after last access)
2. Permanent blobs: Job-related files (cleaned up after job completion)

blob.service.cleanup.interval: 3600 ( by default 1hr )

In your case try to reduce it since the job completes within 5 min.

https://nightlies.apache.org/flink/flink-docs-release-1.4/ops/config.html