Is there a good way to remove "old" files from the Flink file source state? We are reading from S3 and move data out periodically to another archive bucket.

would it be easier to set TTL on S3 source side?

The source state would still have references to the file paths if we use an archive policy in S3 to move them.

good point. I haven't tried https://flink.apache.org/2019/05/17/state-ttl-in-flink-1.8.0-how-to-automatically-cleanup-application-state-in-apache-flink/ on the S3 file source, so not sure would it help you