Hi Aneesha,
the logs would show that Flink is going through a recovery cycle. Recovery
means to cancel running tasks and start them again.
If you don't see something like that in the logs, Flink continues to
processing.
I'm not familiar with the details of S3, so I can't tell if the exception
indicates data loss.
Best, Fabian
2017-08-01 20:38 GMT+02:00 Aneesha Kaushal <aneesha.kaus...@reflektion.com>:
> Hello,
>
> I am using flink 1.2 and writing records to S3 using rolling sink.
>
> I am encountering this S3 write error quite frequently :
>
> TimerException{com.amazonaws.services.s3.model.AmazonS3Exception: Status
> Code: 404, AWS Service: Amazon S3, AWS Request ID: B573887B1850BF28, AWS
> Error Code: null, AWS Error Message: Not Found, S3 Extended Request ID:
> JKAW8uhr/kZixw2ltGUqYdp28ssVw3zXpO/hkNvOHATOvz8bYbEE2EVxkg/vgZdr}
> at
> org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$TriggerTask.run(SystemProcessingTimeService.java:220)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
> at
> java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: Status Code:
> 404, AWS Service: Amazon S3, AWS Request ID: B573887B1850BF28, AWS Error
> Code: null, AWS Error Message: Not Found, S3 Extended Request ID:
> JKAW8uhr/kZixw2ltGUqYdp28ssVw3zXpO/hkNvOHATOvz8bYbEE2EVxkg/vgZdr
> at
> com.amazonaws.http.AmazonHttpClient.handleErrorResponse(AmazonHttpClient.java:798)
> at
> com.amazonaws.http.AmazonHttpClient.executeHelper(AmazonHttpClient.java:421)
> at
> com.amazonaws.http.AmazonHttpClient.execute(AmazonHttpClient.java:232)
> at
> com.amazonaws.services.s3.AmazonS3Client.invoke(AmazonS3Client.java:3528)
> at
> com.amazonaws.services.s3.AmazonS3Client.getObjectMetadata(AmazonS3Client.java:976)
> at
> com.amazonaws.services.s3.AmazonS3Client.getObjectMetadata(AmazonS3Client.java:956)
> at
> org.apache.hadoop.fs.s3a.S3AFileSystem.copyFile(S3AFileSystem.java:1088)
> at org.apache.hadoop.fs.s3a.S3AFileSystem.rename(S3AFileSystem.java:521)
> at
> org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.closeCurrentPartFile(BucketingSink.java:563)
> at
> org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.checkForInactiveBuckets(BucketingSink.java:496)
> at
> org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink.onProcessingTime(BucketingSink.java:479)
> at
> org.apache.flink.streaming.runtime.tasks.SystemProcessingTimeService$TriggerTask.run(SystemProcessingTimeService.java:218)
> ... 7 more
>
>
> I am unable to find the cause of this error. Also, I have the following
> questions regarding this error :
>
> 1) Do we loose any data or flink will go to last checkpoint and write
> again?
> 2) how can we prevent this error?
>
> Thanks,
> Aneesha
>
>
>
