hi all, i'm getting a problem with my session cluster attempting to fetch a non-existent file over and over and is causing my jobs to crash. It then reattempt the same file after restart. Im using the FLIP-27 file system connector library with the google storage plugin included in Flink. Could someone give me some ideas on how to best fix this situation?

It could have been there when you started the job and later it was moved. Flink wouldn’t remove a file from its state in this case. Without using something like the State Processor API you have to start from scratch.

thanks for the response. I've not implemented my own custom source before, would you know which components of the FileSystem datastream connector components would need to be rewritten to achieve the desired outcome?? would it be the splitenumerator and the source readers only, or something else i should target? I don't believe the state processor api is capable interacting with the FlieSource state within my application... unless there's something i'm missing?

state processor api is generic enough to interact with any state, but it's non-trivial to use