This is an automated email from the ASF dual-hosted git repository.
zhuzh 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 b1105813cd7 [FLINK-35522][runtime] Fix the issue that the source task
may get stuck in speculative execution mode.
b1105813cd7 is described below
commit b1105813cd713201d8ba031d6464d0e9ee1489bf
Author: sunxia <[email protected]>
AuthorDate: Wed Jun 5 19:17:40 2024 +0800
[FLINK-35522][runtime] Fix the issue that the source task may get stuck in
speculative execution mode.
---
.../coordinator/SourceCoordinatorContext.java | 12 ++---
.../scheduling/SpeculativeExecutionITCase.java | 53 +++++++++++++++++++++-
2 files changed, 57 insertions(+), 8 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
index fa49577caad..ab303397e03 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java
@@ -663,14 +663,14 @@ public class SourceCoordinatorContext<SplitT extends
SourceSplit>
assignmentTracker.uncheckpointedAssignments().get(subtaskIndex);
if (cachedSplits != null) {
- if (supportsConcurrentExecutionAttempts) {
- assignSplitsToAttempt(subtaskIndex, attemptNumber, new
ArrayList<>(cachedSplits));
- if (hasNoMoreSplits(subtaskIndex)) {
- signalNoMoreSplitsToAttempt(subtaskIndex, attemptNumber);
- }
- } else {
+ if (!supportsConcurrentExecutionAttempts) {
throw new IllegalStateException("No cached split is
expected.");
}
+ assignSplitsToAttempt(subtaskIndex, attemptNumber, new
ArrayList<>(cachedSplits));
+ }
+
+ if (supportsConcurrentExecutionAttempts &&
hasNoMoreSplits(subtaskIndex)) {
+ signalNoMoreSplitsToAttempt(subtaskIndex, attemptNumber);
}
}
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeExecutionITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeExecutionITCase.java
index dc1205cef2b..3e6a1ac7aa3 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeExecutionITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/scheduling/SpeculativeExecutionITCase.java
@@ -40,7 +40,10 @@ import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.ReaderOutput;
import org.apache.flink.api.connector.source.SourceReader;
import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
import org.apache.flink.api.connector.source.lib.NumberSequenceSource;
+import org.apache.flink.api.connector.source.lib.util.IteratorSourceEnumerator;
import org.apache.flink.api.connector.source.lib.util.IteratorSourceReader;
import org.apache.flink.api.connector.source.lib.util.IteratorSourceSplit;
import org.apache.flink.api.java.tuple.Tuple3;
@@ -76,6 +79,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -183,6 +187,13 @@ class SpeculativeExecutionITCase {
checkResults();
}
+ @Test
+ void testSpeculativeExecutionOfNewSourceWithFailure() throws Exception {
+ executeJob(env -> setupJobWithSlowNewSource(env, true));
+ waitUntilJobArchived();
+ checkResults();
+ }
+
@Test
public void testSpeculativeSlowSink() throws Exception {
executeJob(this::setupSpeculativeSlowSink);
@@ -357,9 +368,14 @@ class SpeculativeExecutionITCase {
}
private void setupJobWithSlowNewSource(StreamExecutionEnvironment env) {
+ setupJobWithSlowNewSource(env, false);
+ }
+
+ private void setupJobWithSlowNewSource(
+ StreamExecutionEnvironment env, boolean forceFailureFlag) {
final DataStream<Long> source =
env.fromSource(
- new TestingNumberSequenceSource(),
+ new TestingNumberSequenceSource(forceFailureFlag),
WatermarkStrategy.noWatermarks(),
"source");
addSink(source);
@@ -524,8 +540,18 @@ class SpeculativeExecutionITCase {
}
private static class TestingNumberSequenceSource extends
NumberSequenceSource {
- private TestingNumberSequenceSource() {
+
+ private final boolean forceFailureFlag;
+ // When forceFailureCounter > 0, the source task will throw an
exception on reader close
+ // until forceFailureCounter reaches 0.
+ public static AtomicInteger forceFailureCounter = new AtomicInteger(0);
+
+ private TestingNumberSequenceSource(boolean forceFailureFlag) {
super(0, NUMBERS_TO_PRODUCE - 1);
+ this.forceFailureFlag = forceFailureFlag;
+ if (forceFailureFlag) {
+ forceFailureCounter = new AtomicInteger(1);
+ }
}
@Override
@@ -533,6 +559,21 @@ class SpeculativeExecutionITCase {
SourceReaderContext readerContext) {
return new TestingIteratorSourceReader(readerContext);
}
+
+ @Override
+ public SplitEnumerator<NumberSequenceSplit,
Collection<NumberSequenceSplit>>
+ createEnumerator(final
SplitEnumeratorContext<NumberSequenceSplit> enumContext) {
+
+ int splitSize = enumContext.currentParallelism();
+ // Simulating the case that the splits number less than the
parallelism to verify
+ // unassigned source tasks and failover.
+ if (forceFailureFlag) {
+ splitSize = 1;
+ }
+ final List<NumberSequenceSplit> splits =
+ splitNumberRange(0, NUMBERS_TO_PRODUCE - 1, splitSize);
+ return new IteratorSourceEnumerator<>(enumContext, splits);
+ }
}
private static class TestingIteratorSourceReader<
@@ -548,6 +589,14 @@ class SpeculativeExecutionITCase {
maybeSleep();
return super.pollNext(output);
}
+
+ @Override
+ public void close() throws Exception {
+ if (TestingNumberSequenceSource.forceFailureCounter.get() > 0) {
+
TestingNumberSequenceSource.forceFailureCounter.decrementAndGet();
+ throw new RuntimeException("Forced failure for testing");
+ }
+ }
}
private static class NumberCounterSink extends RichSinkFunction<Long> {
