This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new f63fac7fed3 [FLINK-36146][connector-base] Fix
SingleThreadFetcherManager race condition
f63fac7fed3 is described below
commit f63fac7fed3ca011ee0c58958695a0af7697f832
Author: Kim Gräsman <[email protected]>
AuthorDate: Mon Sep 16 08:50:16 2024 +0200
[FLINK-36146][connector-base] Fix SingleThreadFetcherManager race condition
This issue triggered when running a highly parallel batch job with
FileSource pointed at a large S3 bucket. The visible symptom was:
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)
...
While the exact sequence of events is unclear, there's an obvious
time-of-check/time-of-use bug in getRunningFetcher where fetchers may be
cleared between the time it's checked for empty and the time we try to
take the first element.
Take the fetcher from a single iterator instead. ConcurrentHashMap's
iterator reads the next entry ahead, so next() cannot throw once
hasNext() returned true, even if the fetcher is removed concurrently.
Add a test.
Co-authored-by: Sergey Nuyanzin <[email protected]>
Co-authored-by: Martijn Visser
<[email protected]>
Generated-by: Claude Code (Claude Opus 5.5)
---
.../reader/fetcher/SingleThreadFetcherManager.java | 4 ++-
.../reader/fetcher/SplitFetcherManagerTest.java | 36 ++++++++++++++++++++++
2 files changed, 39 insertions(+), 1 deletion(-)
diff --git
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java
index adb903cbcd6..f13c523686f 100644
---
a/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java
+++
b/flink-connectors/flink-connector-base/src/main/java/org/apache/flink/connector/base/source/reader/fetcher/SingleThreadFetcherManager.java
@@ -27,6 +27,7 @@ import
org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import javax.annotation.Nullable;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -122,6 +123,7 @@ public class SingleThreadFetcherManager<E, SplitT extends
SourceSplit>
}
protected SplitFetcher<E, SplitT> getRunningFetcher() {
- return fetchers.isEmpty() ? null : fetchers.values().iterator().next();
+ Iterator<SplitFetcher<E, SplitT>> iter = fetchers.values().iterator();
+ return iter.hasNext() ? iter.next() : null;
}
}
diff --git
a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
index bcf8bb934de..4a603dd0411 100644
---
a/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
+++
b/flink-connectors/flink-connector-base/src/test/java/org/apache/flink/connector/base/source/reader/fetcher/SplitFetcherManagerTest.java
@@ -44,14 +44,18 @@ import org.slf4j.MDC;
import javax.annotation.Nullable;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.time.Duration;
import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import static org.apache.flink.test.util.TestUtils.waitUntil;
@@ -309,6 +313,38 @@ class SplitFetcherManagerTest {
}
}
+ @Test
+ void testGetRunningFetcherSurvivesFetcherRemovalInToctouWindow() throws
Exception {
+ final SingleThreadFetcherManager<Object, TestingSourceSplit>
fetcherManager =
+ new SingleThreadFetcherManager<>(TestingSplitReader::new, new
Configuration());
+
+ final SplitFetcher<Object, TestingSourceSplit> fetcher =
+ fetcherManager.createSplitFetcher();
+
+ final Map<Integer, SplitFetcher<Object, TestingSourceSplit>>
racyFetchers =
+ new ConcurrentHashMap<Integer, SplitFetcher<Object,
TestingSourceSplit>>() {
+ private boolean removed;
+
+ @Override
+ public Collection<SplitFetcher<Object,
TestingSourceSplit>> values() {
+ // The shutdown callback (fetchers.remove(...)) fires
here: after
+ // getRunningFetcher()'s isEmpty() check saw the
entry, before it is read.
+ if (!removed) {
+ removed = true;
+ clear();
+ }
+ return super.values();
+ }
+ };
+ racyFetchers.put(0, fetcher);
+
+ final Field fetchersField =
SplitFetcherManager.class.getDeclaredField("fetchers");
+ fetchersField.setAccessible(true);
+ fetchersField.set(fetcherManager, racyFetchers);
+
+ assertThat(fetcherManager.getRunningFetcher()).isNull();
+ }
+
// the final modifier is important so that '@SafeVarargs' is accepted on
Java 8
@SuppressWarnings("FinalPrivateMethod")
@SafeVarargs
