[
https://issues.apache.org/jira/browse/FLINK-19437?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17204012#comment-17204012
]
Stephan Ewen commented on FLINK-19437:
--------------------------------------
Fixing this. As suggested by [~lzljs3620320] in FLINK-19370, the non-atomic
file writing can make the streaming file source pick up partial files in some
cases.
This can be fixed by writing to a hidden temp file first, and then renaming
this to the final file.
> FileSourceTextLinesITCase.testContinuousTextFileSource failed with
> "SimpleStreamFormat is not splittable, but found split end (0) different from
> file length (198)"
> -------------------------------------------------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-19437
> URL: https://issues.apache.org/jira/browse/FLINK-19437
> Project: Flink
> Issue Type: Bug
> Components: Connectors / FileSystem, Tests
> Affects Versions: 1.12.0
> Reporter: Dian Fu
> Assignee: Stephan Ewen
> Priority: Major
> Labels: test-stability
>
> https://dev.azure.com/apache-flink/apache-flink/_build/results?buildId=7008&view=logs&j=fc5181b0-e452-5c8f-68de-1097947f6483&t=62110053-334f-5295-a0ab-80dd7e2babbf
> {code}
> 2020-09-27T21:58:38.9199090Z [ERROR]
> testContinuousTextFileSource(org.apache.flink.connector.file.src.FileSourceTextLinesITCase)
> Time elapsed: 0.517 s <<< ERROR!
> 2020-09-27T21:58:38.9199619Z java.lang.RuntimeException: Failed to fetch next
> result
> 2020-09-27T21:58:38.9200118Z at
> org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:106)
> 2020-09-27T21:58:38.9200722Z at
> org.apache.flink.streaming.api.operators.collect.CollectResultIterator.hasNext(CollectResultIterator.java:77)
> 2020-09-27T21:58:38.9201290Z at
> org.apache.flink.streaming.api.datastream.DataStreamUtils.collectRecordsFromUnboundedStream(DataStreamUtils.java:150)
> 2020-09-27T21:58:38.9201920Z at
> org.apache.flink.connector.file.src.FileSourceTextLinesITCase.testContinuousTextFileSource(FileSourceTextLinesITCase.java:136)
> 2020-09-27T21:58:38.9202570Z at
> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> 2020-09-27T21:58:38.9203054Z at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> 2020-09-27T21:58:38.9203539Z at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> 2020-09-27T21:58:38.9203968Z at
> java.lang.reflect.Method.invoke(Method.java:498)
> 2020-09-27T21:58:38.9204369Z at
> org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
> 2020-09-27T21:58:38.9204844Z at
> org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
> 2020-09-27T21:58:38.9205359Z at
> org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
> 2020-09-27T21:58:38.9205814Z at
> org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
> 2020-09-27T21:58:38.9206240Z at
> org.junit.rules.TestWatcher$1.evaluate(TestWatcher.java:55)
> 2020-09-27T21:58:38.9206611Z at
> org.junit.rules.RunRules.evaluate(RunRules.java:20)
> 2020-09-27T21:58:38.9206971Z at
> org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
> 2020-09-27T21:58:38.9207404Z at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
> 2020-09-27T21:58:38.9207971Z at
> org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
> 2020-09-27T21:58:38.9208404Z at
> org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
> 2020-09-27T21:58:38.9208877Z at
> org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
> 2020-09-27T21:58:38.9209279Z at
> org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
> 2020-09-27T21:58:38.9209680Z at
> org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
> 2020-09-27T21:58:38.9210064Z at
> org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
> 2020-09-27T21:58:38.9210476Z at
> org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> 2020-09-27T21:58:38.9210881Z at
> org.junit.rules.ExternalResource$1.evaluate(ExternalResource.java:48)
> 2020-09-27T21:58:38.9211272Z at
> org.junit.rules.RunRules.evaluate(RunRules.java:20)
> 2020-09-27T21:58:38.9211638Z at
> org.junit.runners.ParentRunner.run(ParentRunner.java:363)
> 2020-09-27T21:58:38.9212305Z at
> org.apache.maven.surefire.junit4.JUnit4Provider.execute(JUnit4Provider.java:365)
> 2020-09-27T21:58:38.9213157Z at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeWithRerun(JUnit4Provider.java:273)
> 2020-09-27T21:58:38.9213663Z at
> org.apache.maven.surefire.junit4.JUnit4Provider.executeTestSet(JUnit4Provider.java:238)
> 2020-09-27T21:58:38.9214123Z at
> org.apache.maven.surefire.junit4.JUnit4Provider.invoke(JUnit4Provider.java:159)
> 2020-09-27T21:58:38.9214620Z at
> org.apache.maven.surefire.booter.ForkedBooter.invokeProviderInSameClassLoader(ForkedBooter.java:384)
> 2020-09-27T21:58:38.9215148Z at
> org.apache.maven.surefire.booter.ForkedBooter.runSuitesInProcess(ForkedBooter.java:345)
> 2020-09-27T21:58:38.9215650Z at
> org.apache.maven.surefire.booter.ForkedBooter.execute(ForkedBooter.java:126)
> 2020-09-27T21:58:38.9216095Z at
> org.apache.maven.surefire.booter.ForkedBooter.main(ForkedBooter.java:418)
> 2020-09-27T21:58:38.9216516Z Caused by: java.io.IOException: Failed to fetch
> job execution result
> 2020-09-27T21:58:38.9217004Z at
> org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:175)
> 2020-09-27T21:58:38.9217595Z at
> org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.next(CollectResultFetcher.java:126)
> 2020-09-27T21:58:38.9218182Z at
> org.apache.flink.streaming.api.operators.collect.CollectResultIterator.nextResultFromFetcher(CollectResultIterator.java:103)
> 2020-09-27T21:58:38.9218585Z ... 33 more
> 2020-09-27T21:58:38.9219037Z Caused by:
> java.util.concurrent.ExecutionException:
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> 2020-09-27T21:58:38.9219563Z at
> java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
> 2020-09-27T21:58:38.9219987Z at
> java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1928)
> 2020-09-27T21:58:38.9220511Z at
> org.apache.flink.streaming.api.operators.collect.CollectResultFetcher.getAccumulatorResults(CollectResultFetcher.java:172)
> 2020-09-27T21:58:38.9220915Z ... 35 more
> 2020-09-27T21:58:38.9221225Z Caused by:
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> 2020-09-27T21:58:38.9221680Z at
> org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:147)
> 2020-09-27T21:58:38.9222277Z at
> org.apache.flink.client.program.PerJobMiniClusterFactory$PerJobMiniClusterJobClient.lambda$getJobExecutionResult$2(PerJobMiniClusterFactory.java:196)
> 2020-09-27T21:58:38.9223032Z at
> java.util.concurrent.CompletableFuture.uniApply(CompletableFuture.java:616)
> 2020-09-27T21:58:38.9223479Z at
> java.util.concurrent.CompletableFuture.uniApplyStage(CompletableFuture.java:628)
> 2020-09-27T21:58:38.9223938Z at
> java.util.concurrent.CompletableFuture.thenApply(CompletableFuture.java:1996)
> 2020-09-27T21:58:38.9224503Z at
> org.apache.flink.client.program.PerJobMiniClusterFactory$PerJobMiniClusterJobClient.getJobExecutionResult(PerJobMiniClusterFactory.java:194)
> 2020-09-27T21:58:38.9224926Z ... 36 more
> 2020-09-27T21:58:38.9225315Z Caused by:
> org.apache.flink.runtime.JobException: Recovery is suppressed by
> NoRestartBackoffTimeStrategy
> 2020-09-27T21:58:38.9225889Z at
> org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:116)
> 2020-09-27T21:58:38.9226529Z at
> org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:78)
> 2020-09-27T21:58:38.9227128Z at
> org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:217)
> 2020-09-27T21:58:38.9227658Z at
> org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:210)
> 2020-09-27T21:58:38.9228200Z at
> org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:204)
> 2020-09-27T21:58:38.9228806Z at
> org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:527)
> 2020-09-27T21:58:38.9229416Z at
> org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:421)
> 2020-09-27T21:58:38.9229885Z at
> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
> 2020-09-27T21:58:38.9230283Z at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> 2020-09-27T21:58:38.9230755Z at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> 2020-09-27T21:58:38.9231160Z at
> java.lang.reflect.Method.invoke(Method.java:498)
> 2020-09-27T21:58:38.9231589Z at
> org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:284)
> 2020-09-27T21:58:38.9232086Z at
> org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:199)
> 2020-09-27T21:58:38.9232792Z at
> org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:74)
> 2020-09-27T21:58:38.9233305Z at
> org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:152)
> 2020-09-27T21:58:38.9233750Z at
> akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:26)
> 2020-09-27T21:58:38.9234127Z at
> akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:21)
> 2020-09-27T21:58:38.9234533Z at
> scala.PartialFunction$class.applyOrElse(PartialFunction.scala:123)
> 2020-09-27T21:58:38.9234949Z at
> akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:21)
> 2020-09-27T21:58:38.9235391Z at
> scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:170)
> 2020-09-27T21:58:38.9235808Z at
> scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
> 2020-09-27T21:58:38.9236218Z at
> scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
> 2020-09-27T21:58:38.9236590Z at
> akka.actor.Actor$class.aroundReceive(Actor.scala:517)
> 2020-09-27T21:58:38.9236981Z at
> akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:225)
> 2020-09-27T21:58:38.9237371Z at
> akka.actor.ActorCell.receiveMessage(ActorCell.scala:592)
> 2020-09-27T21:58:38.9237728Z at
> akka.actor.ActorCell.invoke(ActorCell.scala:561)
> 2020-09-27T21:58:38.9238087Z at
> akka.dispatch.Mailbox.processMailbox(Mailbox.scala:258)
> 2020-09-27T21:58:38.9238421Z at akka.dispatch.Mailbox.run(Mailbox.scala:225)
> 2020-09-27T21:58:38.9238819Z at akka.dispatch.Mailbox.exec(Mailbox.scala:235)
> 2020-09-27T21:58:38.9239191Z at
> akka.dispatch.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260)
> 2020-09-27T21:58:38.9239605Z at
> akka.dispatch.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339)
> 2020-09-27T21:58:38.9240040Z at
> akka.dispatch.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979)
> 2020-09-27T21:58:38.9240480Z at
> akka.dispatch.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107)
> 2020-09-27T21:58:38.9240913Z Caused by: java.lang.RuntimeException: One or
> more fetchers have encountered exception
> 2020-09-27T21:58:38.9241431Z at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:178)
> 2020-09-27T21:58:38.9242009Z at
> org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:155)
> 2020-09-27T21:58:38.9242688Z at
> org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:116)
> 2020-09-27T21:58:38.9243221Z at
> org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:199)
> 2020-09-27T21:58:38.9243738Z at
> org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:64)
> 2020-09-27T21:58:38.9244271Z at
> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:67)
> 2020-09-27T21:58:38.9244798Z at
> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:368)
> 2020-09-27T21:58:38.9245369Z at
> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:185)
> 2020-09-27T21:58:38.9245853Z at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:594)
> 2020-09-27T21:58:38.9246417Z at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:558)
> 2020-09-27T21:58:38.9246841Z at
> org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:722)
> 2020-09-27T21:58:38.9247288Z at
> org.apache.flink.runtime.taskmanager.Task.run(Task.java:547)
> 2020-09-27T21:58:38.9247629Z at java.lang.Thread.run(Thread.java:748)
> 2020-09-27T21:58:38.9248047Z Caused by: java.lang.RuntimeException:
> SplitFetcher thread 1 received unexpected exception while polling the records
> 2020-09-27T21:58:38.9248558Z at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:123)
> 2020-09-27T21:58:38.9249151Z at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:91)
> 2020-09-27T21:58:38.9249647Z at
> org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:42)
> 2020-09-27T21:58:38.9250090Z at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> 2020-09-27T21:58:38.9250506Z at
> java.util.concurrent.FutureTask.run(FutureTask.java:266)
> 2020-09-27T21:58:38.9250920Z at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> 2020-09-27T21:58:38.9251369Z at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> 2020-09-27T21:58:38.9251694Z ... 1 more
> 2020-09-27T21:58:38.9252087Z Caused by: java.lang.IllegalArgumentException:
> SimpleStreamFormat is not splittable, but found split end (0) different from
> file length (198)
> 2020-09-27T21:58:38.9252875Z at
> org.apache.flink.connector.file.src.reader.SimpleStreamFormat.checkNotSplit(SimpleStreamFormat.java:110)
> 2020-09-27T21:58:38.9253430Z at
> org.apache.flink.connector.file.src.reader.SimpleStreamFormat.createReader(SimpleStreamFormat.java:85)
> 2020-09-27T21:58:38.9253999Z at
> org.apache.flink.connector.file.src.impl.StreamFormatAdapter.lambda$createReader$0(StreamFormatAdapter.java:68)
> 2020-09-27T21:58:38.9254516Z at
> org.apache.flink.connector.file.src.util.Utils.doWithCleanupOnException(Utils.java:45)
> 2020-09-27T21:58:38.9255034Z at
> org.apache.flink.connector.file.src.impl.StreamFormatAdapter.createReader(StreamFormatAdapter.java:67)
> 2020-09-27T21:58:38.9255655Z at
> org.apache.flink.connector.file.src.impl.FileSourceSplitReader.checkSplitOrStartNext(FileSourceSplitReader.java:103)
> 2020-09-27T21:58:38.9256225Z at
> org.apache.flink.connector.file.src.impl.FileSourceSplitReader.fetch(FileSourceSplitReader.java:68)
> 2020-09-27T21:58:38.9256723Z at
> org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)
> 2020-09-27T21:58:38.9257235Z at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:117)
> 2020-09-27T21:58:38.9257586Z ... 7 more
> {code}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
