I have a simple Flink job (flink version 1.8.3) that has a sink which write to a s3 bucket every 10 minutes.
When I stop the job and restart it from the previous snap shot I often get this error which stopps me to resume the job properly
"throwableInformation": [
"java.io.FileNotFoundException: No such file or directory: s3://my-bucket/2020-09-22--18/_part-0-0_tmp_91f865fe-4774-4699-80ca-483aa36e3fd9",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2231)",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2149)",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2088)",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.open(S3AFileSystem.java:699)",
"\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.FileSystem.open(FileSystem.java:950)",
"\tat org.apache.flink.fs.s3hadoop.HadoopS3AccessHelper.getObject(HadoopS3AccessHelper.java:98)",
"\tat org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverInProgressPart(S3RecoverableMultipartUploadFactory.java:97)",
"\tat org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverRecoverableUpload(S3RecoverableMultipartUploadFactory.java:75)",
"\tat org.apache.flink.fs.s3.common.writer.S3RecoverableWriter.recover(S3RecoverableWriter.java:95)",
"\tat org.apache.flink.fs.s3.common.writer.S3RecoverableWriter.recover(S3RecoverableWriter.java:50)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.restoreInProgressFile(Bucket.java:140)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.<init>(Bucket.java:127)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.restore(Bucket.java:399)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.DefaultBucketFactoryImpl.restoreBucket(DefaultBucketFactoryImpl.java:64)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.handleRestoredBucketState(Buckets.java:177)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.initializeActiveBuckets(Buckets.java:165)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.initializeState(Buckets.java:149)",
"\tat org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState(StreamingFileSink.java:334)",
"\tat org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:178)",
"\tat org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:160)",
"\tat org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96)",
"\tat org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:278)",
"\tat org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:763)",
"\tat org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:297)",
"\tat org.apache.flink.runtime.taskmanager.Task.run(Task.java:714)",
"\tat java.lang.Thread.run(Thread.java:748)"
],
......
......) -> to: Result -> (Sink: Print to Std. Out, Sink: Unnamed, Sink: Unnamed, Sink: Unnamed) (1/1) (7dc2e0b7013bbfddf507531ebc184c46) switched from RUNNING to FAILED.","throwableInformation":["java.io.FileNotFoundException: No such file or directory: s3://my-bucket/2020-09-22--18/_part-0-0_tmp_91f865fe-4774-4699-80ca-483aa36e3fd9","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.s3GetFileStatus(S3AFileSystem.java:2231)","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.innerGetFileStatus(S3AFileSystem.java:2149)","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.getFileStatus(S3AFileSystem.java:2088)","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.s3a.S3AFileSystem.open(S3AFileSystem.java:699)","\tat org.apache.flink.fs.shaded.hadoop3.org.apache.hadoop.fs.FileSystem.open(FileSystem.java:950)","\tat org.apache.flink.fs.s3hadoop.HadoopS3AccessHelper.getObject(HadoopS3AccessHelper.java:98)","\tat org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverInProgressPart(S3RecoverableMultipartUploadFactory.java:97)","\tat org.apache.flink.fs.s3.common.writer.S3RecoverableMultipartUploadFactory.recoverRecoverableUpload(S3RecoverableMultipartUploadFactory.java:75)","\tat org.apache.flink.fs.s3.common.writer.S3RecoverableWriter.recover(S3RecoverableWriter.java:95)","\tat org.apache.flink.fs.s3.common.writer.S3RecoverableWriter.recover(S3RecoverableWriter.java:50)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.restoreInProgressFile(Bucket.java:140)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.<init>(Bucket.java:127)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Bucket.restore(Bucket.java:399)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.DefaultBucketFactoryImpl.restoreBucket(DefaultBucketFactoryImpl.java:64)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.handleRestoredBucketState(Buckets.java:177)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.initializeActiveBuckets(Buckets.java:165)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.Buckets.initializeState(Buckets.java:149)","\tat org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.initializeState(StreamingFileSink.java:334)","\tat org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tryRestoreFunction(StreamingFunctionUtils.java:178)","\tat org.apache.flink.streaming.util.functions.StreamingFunctionUtils.restoreFunctionState(StreamingFunctionUtils.java:160)","\tat org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.initializeState(AbstractUdfStreamOperator.java:96)","\tat org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.java:278)","\tat org.apache.flink.streaming.runtime.tasks.StreamTask.initializeState(StreamTask.java:763)","\tat
I have checked in the s3 bucket and that temp file is not there. it looks like somehow the when the job stopped there was some kind of error while writing the file and when I restart the job it somehow tries again to find that file but it's not there?
How can I avoid having this issue? Is there a way to ignore missing files?