This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new cf37da108a0 [FLINK-36146][connector-base] Fix
SingleThreadFetcherManager race condition
cf37da108a0 is described below
commit cf37da108a06ba5c99ebd7d5778279e36aaed7e5
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 391885a6fbe..d3feaf2e9d2 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
@@ -28,6 +28,7 @@ import
org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
import
org.apache.flink.connector.base.source.reader.synchronization.FutureCompletingBlockingQueue;
import java.util.Collection;
+import java.util.Iterator;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Supplier;
@@ -163,6 +164,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 08e7ac45df9..9f7d132cec4 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
@@ -33,10 +33,14 @@ import org.apache.flink.core.testutils.OneShotLatch;
import org.junit.Test;
import java.io.IOException;
+import java.lang.reflect.Field;
import java.util.ArrayDeque;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
+import java.util.Map;
import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -68,6 +72,38 @@ public class SplitFetcherManagerTest {
.hasRootCauseMessage("Artificial exception on closing the
split reader.");
}
+ @Test
+ public 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
