Hi, I have a simple data ingestion job

FileSource

(S3) ->

KafkaSink

with a

JobListner.onJobExecuted

to "mv" the source file to a new folder/bucket (cleanup logic). When the job runs, one or more of the fetchers throw

FileNotFoundException

indicating the source file is missing; if I remove the

JobListener

code, the job completes successfully. It appears that the source file was moved to the new folder/bucket while the

SplitFetcher

is still polling for the records. Is there a way to perform clean-up logic after all streaming tasks have been completed? ps. this seems to occur when running in session mode (1 JobManager, 1 TaskManager); when I ran it in my IDE, the job completes successfully with the cleanup logic.