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 2025/08/30 09:19:00 UTC
[jira] [Commented] (FLINK-36146) NoSuchElement exception from SingleThreadFetcherManager
[ https://issues.apache.org/jira/browse/FLINK-36146?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18017160#comment-18017160 ]
Kim Gräsman commented on FLINK-36146:
-------------------------------------
The PR closed due to inactivity, I would appreciate if it could be reopened. This is still an issue that hits us occasionally, and it would be nice to get it out of the way for future releases.
> 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
> Labels: pull-request-available
>
> 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)