[
https://issues.apache.org/jira/browse/FLINK-36146?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17881864#comment-17881864
]
Jiangjie Qin commented on FLINK-36146:
--------------------------------------
Hi Kim, yes, the ConcurrentHashMap iterator works on top of a state at the
point when the iterator was created. You may refer to the [code contribution
guide|https://flink.apache.org/how-to-contribute/contribute-code/] for the PR
creation.
BTW, just want to clarify a little bit on the issue you saw. When the source is
in the normal running state, we should not see this race condition, because all
the fetcher manipulations are done synchronously in a single main thread. The
issue you saw here happens when the SourceReader is closing. As a part of the
shutdown sequence, the SplitFetchers are shutting down and removed from the
fetcher map asynchronously. At this point we do not expect new splits to be
added to the source reader. So, it is a little weird that the main thread is
still adding splits to the SourceReader when the source reader is closing.
So apart from this specific race condition, it would be good understand why
there are still more splits added to the SourceReader while it is closing.
> 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)
