You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Kim Gräsman (Jira)" <ji...@apache.org> on 2024/09/16 07:03:00 UTC
[jira] [Comment Edited] (FLINK-36146) NoSuchElement exception from SingleThreadFetcherManager
[ https://issues.apache.org/jira/browse/FLINK-36146?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17881867#comment-17881867 ]
Kim Gräsman edited comment on FLINK-36146 at 9/16/24 7:02 AM:
--------------------------------------------------------------
Thank you!
Hmm, yeah, I don't understand the higher-level flow enough to say why this happens.
A few things that are maybe unusual about the runs where we saw this:
* Flink 1.14.2 (i.e. bug could have been masked in newer releases by behavior changes "higher up")
* Batch mode (I guess there's more distinct shutdown in batch mode when a task completes, but it's not really clear to me)
* High parallelism (I think these runs had -p 1600. Not sure if that affects something?)
* Many input files (I think I counted 400+ billion records, and I believe each record is a file in this archive)
So it's a big, old batch job.
was (Author: JIRAUSER306734):
Thank you!
Hmm, yeah, I don't understand the higher-level flow enough to say why this happens.
A few things that are maybe unusual about the runs where we saw this:
* Flink 1.14.2 (i.e. bug could have been masked in newer releases by behavior changes "higher up")
* Batch mode (I guess there's more distinct shutdown in batch mode when a task completes, but it's not really clear to me)
* High parallelism (I think these runs had -p 400. Not sure if that affects something?)
* Many input files (I think I counted 400+ billion records, and I believe each record is a file in this archive)
So it's a big, old batch job.
> NoSuchElement exception from SingleThreadFetcherManager
> -------------------------------------------------------
>
> Key: FLINK-36146
> URL: https://issues.apache.org/jira/browse/FLINK-36146
> Project: Flink
> Issue Type: Bug
> Components: API / Core
> Environment: AWS EMR/Yarn
> Reporter: Kim Gräsman
> Priority: Minor
>
> We're running Flink 1.14.2, but this appears to be an issue still on mainline, so I thought I'd report it.
> When running with high parallelism we've noticed a spurious error triggered by a FileSource reader from S3;
> {code:java}
> 2024-08-19 15:23:07,044 INFO org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Finished reading split(s) [0000543131]
> 2024-08-19 15:23:07,044 INFO org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Finished reading from splits [0000543131]
> 2024-08-19 15:23:07,044 INFO org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager [] - Closing splitFetcher 157 because it is idle.
> 2024-08-19 15:23:07,045 INFO org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Shutting down split fetcher 157
> 2024-08-19 15:23:07,045 INFO org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher [] - Split fetcher 157 exited.
> 2024-08-19 15:23:07,048 INFO org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Adding split(s) to reader: [FileSourceSplit: ... [0, 21679984) hosts=[localhost] ID=0000201373 position=null]
> 2024-08-19 15:23:07,064 INFO org.apache.flink.connector.base.source.reader.SourceReaderBase [] - Closing Source Reader.
> 2024-08-19 15:23:07,069 WARN org.apache.flink.runtime.taskmanager.Task [] - Source: ... -> ... (114/1602)#0 (...) switched from RUNNING to FAILED with failure cause: java.util.NoSuchElementException
> at java.base/java.util.concurrent.ConcurrentHashMap$ValueIterator.next(ConcurrentHashMap.java:3471)
> at org.apache.flink.connector.base.source.reader.fetcher.SingleThreadFetcherManager.getRunningFetcher(SingleThreadFetcherManager.java:94)
> at org.apache.flink.connector.base.source.reader.fetcher.SingleThreadFetcherManager.addSplits(SingleThreadFetcherManager.java:82)
> at org.apache.flink.connector.base.source.reader.SourceReaderBase.addSplits(SourceReaderBase.java:242)
> at org.apache.flink.streaming.api.operators.SourceOperator.handleOperatorEvent(SourceOperator.java:428)
> at org.apache.flink.streaming.runtime.tasks.OperatorEventDispatcherImpl.dispatchEventToHandlers(OperatorEventDispatcherImpl.java:70)
> at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.dispatchOperatorEvent(RegularOperatorChain.java:83)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$dispatchOperatorEvent$19(StreamTask.java:1473)
> at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
> at org.apache.flink.streaming.runtime.tasks.mailbox.Mail.run(Mail.java:90)
> at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMailsNonBlocking(MailboxProcessor.java:353)
> at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.processMail(MailboxProcessor.java:317)
> at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:201)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:809)
> at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:761)
> at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
> at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
> at java.base/java.lang.Thread.run(Thread.java:829) {code}
> I believe this may be caused by a tiny TOCTOU race in {{{}SingleThreadedFetcherManager{}}}. I'll admit that I don't fully understand what the execution flows through that code look like, but the use of atomic and synchronized indicate that it's used by multiple threads. If that's not the case, this report can be safely ignored.
> The backtrace points to [https://github.com/apache/flink/blob/4faf0966766e3734792f80ed66e512aa3033cacd/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java#L165|https://github.com/apache/flink/blob/4faf0966766e3734792f80ed66e512aa3033cacd/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java#L165.]
> And it looks like the concurrent hash map might be modified between the check for {{isEmpty()}} and the call to {{{}fetchers.values().iterator().next(){}}}.
> I would suggest Python-style:
> {code:java}
> try {
> return fetchers.values().iterator().next();
> } catch (NoSuchElementException) {
> return null;
> }{code}
> here instead, which should let {{ConcurrentHashMap}} handle its synchronization internally.
> For some reason we were able to reproduce consistently with 2 Task Managers and 2 slots per node, but not with 1 Task Manager and 4 slots, if that helps construct a repro test case (presumably more interlocking from {{synchronized}} in a single-TM environment, but not sure).
--
This message was sent by Atlassian Jira
(v8.20.10#820010)