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 533e5fc321c [FLINK-40098][test-utils] Add FailingCheckpointedSource
for exactly-once source tests (#28686)
533e5fc321c is described below
commit 533e5fc321cea2b13807e5006f3efbda8b4b0f24
Author: Martijn Visser <[email protected]>
AuthorDate: Wed Oct 21 19:56:14 2026 +0200
[FLINK-40098][test-utils] Add FailingCheckpointedSource for exactly-once
source tests (#28686)
* [FLINK-40098][test-utils] Add FailingCheckpointedSource for exactly-once
source tests
Replaces the fail-once-and-recover semantics of the legacy
SourceFunction-based FailingSource with a Source V2 utility
* [FLINK-40098][tests] Migrate EventTimeAllWindowCheckpointingITCase to
Source V2
Replaces the legacy FailingSource with FailingCheckpointedSource,
pinned to parallelism 1 like the implicitly non-parallel legacy source.
KeyedEventTimeGenerator implements both generator interfaces until the
remaining FailingSource consumers are migrated.
* [FLINK-40098][tests] Migrate ProcessingTimeWindowCheckpointingITCase to
Source V2
Exercises the idle-after-emission mode of FailingCheckpointedSource,
which mirrors the usingProcessingTime flag of the legacy FailingSource
Generated-by: Claude Code (Fable 5.1)
---
.../flink-test-utils-connector/pom.xml | 9 +
.../util/source/FailingCheckpointedSource.java | 458 +++++++++++++++++++++
.../util/source/FailingCheckpointedSourceTest.java | 329 +++++++++++++++
.../EventTimeAllWindowCheckpointingITCase.java | 46 ++-
.../EventTimeWindowCheckpointingITCase.java | 20 +-
.../ProcessingTimeWindowCheckpointingITCase.java | 50 ++-
6 files changed, 880 insertions(+), 32 deletions(-)
diff --git a/flink-test-utils-parent/flink-test-utils-connector/pom.xml
b/flink-test-utils-parent/flink-test-utils-connector/pom.xml
index 250f6252a38..7b37b2072ab 100644
--- a/flink-test-utils-parent/flink-test-utils-connector/pom.xml
+++ b/flink-test-utils-parent/flink-test-utils-connector/pom.xml
@@ -49,6 +49,15 @@ under the License.
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+
+ <!-- For the source mocks (MockSplitEnumeratorContext) -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/flink-test-utils-parent/flink-test-utils-connector/src/main/java/org/apache/flink/test/util/source/FailingCheckpointedSource.java
b/flink-test-utils-parent/flink-test-utils-connector/src/main/java/org/apache/flink/test/util/source/FailingCheckpointedSource.java
new file mode 100644
index 00000000000..8c887e44f85
--- /dev/null
+++
b/flink-test-utils-parent/flink-test-utils-connector/src/main/java/org/apache/flink/test/util/source/FailingCheckpointedSource.java
@@ -0,0 +1,458 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.util.source;
+
+import org.apache.flink.annotation.Experimental;
+import org.apache.flink.api.common.eventtime.Watermark;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.connector.source.Boundedness;
+import org.apache.flink.api.connector.source.ReaderOutput;
+import org.apache.flink.api.connector.source.Source;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SourceReaderContext;
+import org.apache.flink.api.connector.source.SourceSplit;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.SplitEnumeratorContext;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.core.io.InputStatus;
+import org.apache.flink.core.io.SimpleVersionedSerializer;
+import org.apache.flink.core.memory.DataInputDeserializer;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/**
+ * A test source with a checkpointed emission position that fails subtask 0
exactly once after a
+ * completed checkpoint and resumes from the restored position, for verifying
exactly-once behavior
+ * across recovery.
+ *
+ * <p>Each subtask invokes the {@link EventGenerator} once per sequence
number. Subtask 0 fails
+ * after half of its emit calls, like the legacy {@code FailingSource}: a
snapshot taken past a
+ * quarter of the emit calls arms the failure, emission continues to the
failure position, and the
+ * reader throws once the armed checkpoint completes. Emission holds just past
the arming threshold
+ * until a snapshot arms, so recovery always replays the records between the
arming and failure
+ * positions, independent of emission speed. Checkpointing must be enabled.
All records of one
+ * {@link EventGenerator#emit} call are emitted in a single {@code pollNext}
and cannot be split by
+ * a checkpoint. Rescaling is not supported; use a fixed source parallelism.
+ *
+ * <p>{@link #withIdleAfterEmission()} keeps the source idle instead of
finishing, for
+ * processing-time tests whose job is terminated externally.
+ */
+@Experimental
+public class FailingCheckpointedSource<T>
+ implements Source<T, FailingCheckpointedSource.SequenceSplit, Void>,
+ ResultTypeQueryable<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Generates and emits the records (and optionally timestamps and
watermarks) for one sequence
+ * number of one subtask.
+ */
+ @FunctionalInterface
+ public interface EventGenerator<T> extends Serializable {
+ void emit(int subtaskIndex, long sequenceNo, GeneratorOutput<T>
output);
+ }
+
+ /** Output for the {@link EventGenerator}, bridging to the reader output.
*/
+ public interface GeneratorOutput<T> {
+ void collect(T record);
+
+ void collect(T record, long timestamp);
+
+ void emitWatermark(long timestamp);
+ }
+
+ private final EventGenerator<T> generator;
+ private final long emitCallsPerSubtask;
+ private final long failAfterEmitCalls;
+ private final long armAfterEmitCalls;
+ private final TypeInformation<T> producedType;
+ private final boolean idleAfterEmission;
+
+ private FailingCheckpointedSource(
+ EventGenerator<T> generator,
+ long emitCallsPerSubtask,
+ TypeInformation<T> producedType,
+ boolean idleAfterEmission) {
+ // fewer than two emit calls leave no checkpointable position before
the failure
+ checkArgument(emitCallsPerSubtask >= 2, "emitCallsPerSubtask must be
>= 2");
+ this.generator = checkNotNull(generator);
+ this.emitCallsPerSubtask = emitCallsPerSubtask;
+ this.failAfterEmitCalls = emitCallsPerSubtask / 2;
+ this.armAfterEmitCalls = failAfterEmitCalls / 2;
+ this.producedType = checkNotNull(producedType);
+ this.idleAfterEmission = idleAfterEmission;
+ }
+
+ /**
+ * Creates a source whose subtasks each invoke the generator {@code
emitCallsPerSubtask} times
+ * and whose subtask 0 fails once after half of them.
+ */
+ public static <T> FailingCheckpointedSource<T> of(
+ EventGenerator<T> generator,
+ long emitCallsPerSubtask,
+ TypeInformation<T> producedType) {
+ return new FailingCheckpointedSource<>(generator, emitCallsPerSubtask,
producedType, false);
+ }
+
+ /**
+ * Returns a copy of this source that stays idle after emitting all
sequence numbers instead of
+ * finishing, for processing-time tests whose job is terminated externally.
+ */
+ public FailingCheckpointedSource<T> withIdleAfterEmission() {
+ return new FailingCheckpointedSource<>(generator, emitCallsPerSubtask,
producedType, true);
+ }
+
+ @Override
+ public Boundedness getBoundedness() {
+ return Boundedness.CONTINUOUS_UNBOUNDED;
+ }
+
+ @Override
+ public SourceReader<T, SequenceSplit> createReader(SourceReaderContext
readerContext) {
+ return new FailingCheckpointedSourceReader<>(
+ generator,
+ emitCallsPerSubtask,
+ failAfterEmitCalls,
+ armAfterEmitCalls,
+ idleAfterEmission);
+ }
+
+ @Override
+ public SplitEnumerator<SequenceSplit, Void> createEnumerator(
+ SplitEnumeratorContext<SequenceSplit> enumContext) {
+ return new SequenceSplitEnumerator(enumContext, true);
+ }
+
+ @Override
+ public SplitEnumerator<SequenceSplit, Void> restoreEnumerator(
+ SplitEnumeratorContext<SequenceSplit> enumContext, Void
checkpoint) {
+ // After restore the readers recover their splits from their own
state, so the enumerator
+ // must not assign fresh splits to re-registering readers.
+ return new SequenceSplitEnumerator(enumContext, false);
+ }
+
+ @Override
+ public SimpleVersionedSerializer<SequenceSplit> getSplitSerializer() {
+ return SequenceSplit.SERIALIZER;
+ }
+
+ @Override
+ public SimpleVersionedSerializer<Void> getEnumeratorCheckpointSerializer()
{
+ return VoidSerializer.INSTANCE;
+ }
+
+ @Override
+ public TypeInformation<T> getProducedType() {
+ return producedType;
+ }
+
+ /**
+ * A split carrying the emission position of one source subtask, so that
the reader can resume
+ * from the checkpointed position after recovery. The subtask index
assumes a fixed source
+ * parallelism.
+ */
+ static final class SequenceSplit implements SourceSplit {
+
+ static final SimpleVersionedSerializer<SequenceSplit> SERIALIZER =
+ new SequenceSplitSerializer();
+
+ private final int subtaskIndex;
+ private final long position;
+
+ SequenceSplit(int subtaskIndex, long position) {
+ this.subtaskIndex = subtaskIndex;
+ this.position = position;
+ }
+
+ int getSubtaskIndex() {
+ return subtaskIndex;
+ }
+
+ long getPosition() {
+ return position;
+ }
+
+ @Override
+ public String splitId() {
+ return "sequence-split-" + subtaskIndex;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (!(o instanceof SequenceSplit)) {
+ return false;
+ }
+ final SequenceSplit that = (SequenceSplit) o;
+ return subtaskIndex == that.subtaskIndex && position ==
that.position;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(subtaskIndex, position);
+ }
+
+ @Override
+ public String toString() {
+ return "SequenceSplit{subtaskIndex=" + subtaskIndex + ",
position=" + position + "}";
+ }
+ }
+
+ private static final class SequenceSplitSerializer
+ implements SimpleVersionedSerializer<SequenceSplit> {
+
+ @Override
+ public int getVersion() {
+ return 1;
+ }
+
+ @Override
+ public byte[] serialize(SequenceSplit split) throws IOException {
+ final DataOutputSerializer out = new DataOutputSerializer(12);
+ out.writeInt(split.subtaskIndex);
+ out.writeLong(split.position);
+ return out.getCopyOfBuffer();
+ }
+
+ @Override
+ public SequenceSplit deserialize(int version, byte[] serialized)
throws IOException {
+ final DataInputDeserializer in = new
DataInputDeserializer(serialized);
+ return new SequenceSplit(in.readInt(), in.readLong());
+ }
+ }
+
+ /** Assigns one position split per subtask on fresh starts. */
+ private static final class SequenceSplitEnumerator
+ implements SplitEnumerator<SequenceSplit, Void> {
+
+ private final SplitEnumeratorContext<SequenceSplit> context;
+ private final boolean assignSplitsOnRegistration;
+ private final Map<Integer, SequenceSplit> splitsToReassign = new
HashMap<>();
+ private final Set<Integer> assignedSubtasks = new HashSet<>();
+
+ private SequenceSplitEnumerator(
+ SplitEnumeratorContext<SequenceSplit> context, boolean
assignSplitsOnRegistration) {
+ this.context = context;
+ this.assignSplitsOnRegistration = assignSplitsOnRegistration;
+ }
+
+ @Override
+ public void start() {}
+
+ @Override
+ public void handleSplitRequest(int subtaskId, @Nullable String
requesterHostname) {}
+
+ @Override
+ public void addSplitsBack(List<SequenceSplit> splits, int subtaskId) {
+ for (SequenceSplit split : splits) {
+ splitsToReassign.put(split.getSubtaskIndex(), split);
+ }
+ }
+
+ @Override
+ public void addReader(int subtaskId) {
+ final SequenceSplit returnedSplit =
splitsToReassign.remove(subtaskId);
+ if (returnedSplit != null) {
+ context.assignSplit(returnedSplit, subtaskId);
+ } else if (assignSplitsOnRegistration &&
assignedSubtasks.add(subtaskId)) {
+ // only assign on the very first registration; on partial
failover this
+ // enumerator survives and the reader recovers its split from
its own state
+ context.assignSplit(new SequenceSplit(subtaskId, 0),
subtaskId);
+ }
+ }
+
+ @Override
+ public Void snapshotState(long checkpointId) {
+ return null;
+ }
+
+ @Override
+ public void close() {}
+ }
+
+ /**
+ * The reader. All methods are invoked in the task's mailbox thread, so no
synchronization is
+ * required between emission, snapshots, and checkpoint notifications.
+ */
+ private static final class FailingCheckpointedSourceReader<T>
+ implements SourceReader<T, SequenceSplit> {
+
+ private final EventGenerator<T> generator;
+ private final long emitCallsPerSubtask;
+ private final long failAfterEmitCalls;
+ private final long armAfterEmitCalls;
+ private final boolean idleAfterEmission;
+ private final BridgingGeneratorOutput generatorOutput = new
BridgingGeneratorOutput();
+
+ private boolean hasSplit;
+ private int subtaskIndex;
+ private long position;
+ private boolean eligibleToFail;
+ @Nullable private Long armedCheckpointId;
+ private boolean failureRequested;
+ private CompletableFuture<Void> availability =
CompletableFuture.completedFuture(null);
+
+ private FailingCheckpointedSourceReader(
+ EventGenerator<T> generator,
+ long emitCallsPerSubtask,
+ long failAfterEmitCalls,
+ long armAfterEmitCalls,
+ boolean idleAfterEmission) {
+ this.generator = generator;
+ this.emitCallsPerSubtask = emitCallsPerSubtask;
+ this.failAfterEmitCalls = failAfterEmitCalls;
+ this.armAfterEmitCalls = armAfterEmitCalls;
+ this.idleAfterEmission = idleAfterEmission;
+ }
+
+ @Override
+ public void start() {}
+
+ @Override
+ public InputStatus pollNext(ReaderOutput<T> output) {
+ if (failureRequested && position >= failAfterEmitCalls) {
+ throw new FlinkRuntimeException("Artificial Failure");
+ }
+ if (!hasSplit) {
+ return pause();
+ }
+ if (eligibleToFail) {
+ if (armedCheckpointId == null && position > armAfterEmitCalls)
{
+ // hold until a snapshot arms, so the armed checkpoint
provably captures a
+ // pre-failure position and recovery replays the records
in between
+ return pause();
+ }
+ if (position >= failAfterEmitCalls) {
+ // hold at the failure position until the armed checkpoint
completes
+ return pause();
+ }
+ }
+ if (position >= emitCallsPerSubtask) {
+ return idleAfterEmission ? pause() : InputStatus.END_OF_INPUT;
+ }
+ generatorOutput.delegate = output;
+ generator.emit(subtaskIndex, position++, generatorOutput);
+ return InputStatus.MORE_AVAILABLE;
+ }
+
+ private InputStatus pause() {
+ if (availability.isDone()) {
+ availability = new CompletableFuture<>();
+ }
+ return InputStatus.NOTHING_AVAILABLE;
+ }
+
+ @Override
+ public List<SequenceSplit> snapshotState(long checkpointId) {
+ if (!hasSplit) {
+ return Collections.emptyList();
+ }
+ // only arm past the threshold, so recovery demonstrably restores
progress
+ if (eligibleToFail && armedCheckpointId == null && position >
armAfterEmitCalls) {
+ armedCheckpointId = checkpointId;
+ // resume emission toward the failure position
+ availability.complete(null);
+ }
+ return Collections.singletonList(new SequenceSplit(subtaskIndex,
position));
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) {
+ if (eligibleToFail && armedCheckpointId != null &&
armedCheckpointId == checkpointId) {
+ failureRequested = true;
+ availability.complete(null);
+ }
+ }
+
+ @Override
+ public void notifyCheckpointAborted(long checkpointId) {
+ // let a later snapshot re-arm after an abort
+ if (armedCheckpointId != null && armedCheckpointId ==
checkpointId) {
+ armedCheckpointId = null;
+ }
+ }
+
+ @Override
+ public void addSplits(List<SequenceSplit> splits) {
+ checkState(
+ !hasSplit && splits.size() == 1,
+ "expecting exactly one split per subtask, got %s (split
already present: %s)",
+ splits,
+ hasSplit);
+ final SequenceSplit split = splits.get(0);
+ hasSplit = true;
+ subtaskIndex = split.getSubtaskIndex();
+ position = split.getPosition();
+ // restored position > 0 identifies the recovery attempt (no
attempt number in
+ // SourceReaderContext)
+ eligibleToFail = split.getSubtaskIndex() == 0 &&
split.getPosition() == 0;
+ availability.complete(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> isAvailable() {
+ return availability;
+ }
+
+ @Override
+ public void notifyNoMoreSplits() {}
+
+ @Override
+ public void close() {
+ availability.complete(null);
+ }
+
+ private final class BridgingGeneratorOutput implements
GeneratorOutput<T> {
+
+ @Nullable private ReaderOutput<T> delegate;
+
+ @Override
+ public void collect(T record) {
+ delegate.collect(record);
+ }
+
+ @Override
+ public void collect(T record, long timestamp) {
+ delegate.collect(record, timestamp);
+ }
+
+ @Override
+ public void emitWatermark(long timestamp) {
+ delegate.emitWatermark(new Watermark(timestamp));
+ }
+ }
+ }
+}
diff --git
a/flink-test-utils-parent/flink-test-utils-connector/src/test/java/org/apache/flink/test/util/source/FailingCheckpointedSourceTest.java
b/flink-test-utils-parent/flink-test-utils-connector/src/test/java/org/apache/flink/test/util/source/FailingCheckpointedSourceTest.java
new file mode 100644
index 00000000000..9573bfcdb3f
--- /dev/null
+++
b/flink-test-utils-parent/flink-test-utils-connector/src/test/java/org/apache/flink/test/util/source/FailingCheckpointedSourceTest.java
@@ -0,0 +1,329 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.util.source;
+
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.connector.source.SourceReader;
+import org.apache.flink.api.connector.source.SplitEnumerator;
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.core.io.InputStatus;
+import
org.apache.flink.test.util.source.FailingCheckpointedSource.SequenceSplit;
+import org.apache.flink.util.FlinkRuntimeException;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/**
+ * Tests for the failure protocol of {@link FailingCheckpointedSource}: arm on
a snapshot past the
+ * threshold, pause at the failure position, fail once the armed checkpoint
completes, and resume
+ * from the restored position without failing again.
+ */
+class FailingCheckpointedSourceTest {
+
+ private static final long EMIT_CALLS = 20;
+ private static final long FAILURE_POSITION = EMIT_CALLS / 2;
+
+ private static FailingCheckpointedSource<Long> failingSource() {
+ return FailingCheckpointedSource.of(
+ (subtaskIndex, sequenceNo, output) ->
output.collect(sequenceNo),
+ EMIT_CALLS,
+ Types.LONG);
+ }
+
+ private static SourceReader<Long, SequenceSplit> readerWithSplit(
+ FailingCheckpointedSource<Long> source, SequenceSplit split) {
+ final SourceReader<Long, SequenceSplit> reader =
source.createReader(null);
+ reader.addSplits(Collections.singletonList(split));
+ return reader;
+ }
+
+ private static List<Long> range(long startInclusive, long endExclusive) {
+ return LongStream.range(startInclusive,
endExclusive).boxed().collect(Collectors.toList());
+ }
+
+ @Test
+ void testFailsOnceAfterArmedCheckpointCompletes() throws Exception {
+ final SourceReader<Long, SequenceSplit> reader =
+ readerWithSplit(failingSource(), new SequenceSplit(0, 0));
+ final TestReaderOutput<Long> output = new TestReaderOutput<>();
+
+ // Emission holds just past the arming threshold until a snapshot arms
the failure.
+ final long armHoldPosition = FAILURE_POSITION / 2 + 1;
+ for (long i = 0; i < armHoldPosition; i++) {
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ }
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(reader.isAvailable()).isNotDone();
+ assertThat(output.getCollected()).isEqualTo(range(0, armHoldPosition));
+
+ // The arming snapshot captures a position strictly before the failure
position, so
+ // recovery is guaranteed to replay the records in between.
+ final List<SequenceSplit> snapshot = reader.snapshotState(1);
+ assertThat(snapshot).hasSize(1);
+ assertThat(snapshot.get(0).getPosition())
+ .isEqualTo(armHoldPosition)
+ .isLessThan(FAILURE_POSITION);
+ assertThat(reader.isAvailable()).isDone();
+
+ // Emission resumes to the failure position and holds for the armed
checkpoint.
+ for (long i = armHoldPosition; i < FAILURE_POSITION; i++) {
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ }
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(reader.isAvailable()).isNotDone();
+
+ reader.notifyCheckpointComplete(1);
+ assertThat(reader.isAvailable()).isDone();
+ assertThatThrownBy(() -> reader.pollNext(output))
+ .isInstanceOf(FlinkRuntimeException.class)
+ .hasMessage("Artificial Failure");
+ assertThat(output.getCollected()).isEqualTo(range(0,
FAILURE_POSITION));
+ }
+
+ @Test
+ void testDoesNotArmOnCheckpointBeforeThreshold() throws Exception {
+ final SourceReader<Long, SequenceSplit> reader =
+ readerWithSplit(failingSource(), new SequenceSplit(0, 0));
+ final TestReaderOutput<Long> output = new TestReaderOutput<>();
+
+ // Only 3 emit calls: below the arming threshold of FAILURE_POSITION /
2.
+ for (long i = 0; i < 3; i++) {
+ reader.pollNext(output);
+ }
+ reader.snapshotState(1);
+ reader.notifyCheckpointComplete(1);
+
+ // The early checkpoint must not trigger the failure; emission
continues to the arm hold.
+ final long armHoldPosition = FAILURE_POSITION / 2 + 1;
+ for (long i = 3; i < armHoldPosition; i++) {
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ }
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
+ reader.snapshotState(2);
+ for (long i = armHoldPosition; i < FAILURE_POSITION; i++) {
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ }
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ reader.notifyCheckpointComplete(2);
+ assertThatThrownBy(() -> reader.pollNext(output))
+ .isInstanceOf(FlinkRuntimeException.class)
+ .hasMessage("Artificial Failure");
+ }
+
+ @Test
+ void testRestoredReaderResumesWithoutFailingAgain() throws Exception {
+ final SourceReader<Long, SequenceSplit> reader =
+ readerWithSplit(failingSource(), new SequenceSplit(0,
FAILURE_POSITION));
+ final TestReaderOutput<Long> output = new TestReaderOutput<>();
+
+ InputStatus status = InputStatus.MORE_AVAILABLE;
+ while (status == InputStatus.MORE_AVAILABLE) {
+ status = reader.pollNext(output);
+ }
+ assertThat(status).isEqualTo(InputStatus.END_OF_INPUT);
+ assertThat(output.getCollected()).isEqualTo(range(FAILURE_POSITION,
EMIT_CALLS));
+ }
+
+ @Test
+ void testNonZeroSubtaskNeverFails() throws Exception {
+ final SourceReader<Long, SequenceSplit> reader =
+ readerWithSplit(failingSource(), new SequenceSplit(1, 0));
+ final TestReaderOutput<Long> output = new TestReaderOutput<>();
+
+ InputStatus status = InputStatus.MORE_AVAILABLE;
+ while (status == InputStatus.MORE_AVAILABLE) {
+ reader.snapshotState(output.getCollectedCount());
+ reader.notifyCheckpointComplete(output.getCollectedCount());
+ status = reader.pollNext(output);
+ }
+ assertThat(status).isEqualTo(InputStatus.END_OF_INPUT);
+ assertThat(output.getCollected()).isEqualTo(range(0, EMIT_CALLS));
+ }
+
+ @Test
+ void testIdleAfterEmissionKeepsReaderAlive() throws Exception {
+ final SourceReader<Long, SequenceSplit> reader =
+ readerWithSplit(
+ failingSource().withIdleAfterEmission(),
+ new SequenceSplit(0, FAILURE_POSITION));
+ final TestReaderOutput<Long> output = new TestReaderOutput<>();
+
+ InputStatus status = InputStatus.MORE_AVAILABLE;
+ while (status == InputStatus.MORE_AVAILABLE) {
+ status = reader.pollNext(output);
+ }
+ assertThat(status).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ assertThat(reader.isAvailable()).isNotDone();
+ assertThat(output.getCollected()).isEqualTo(range(FAILURE_POSITION,
EMIT_CALLS));
+ }
+
+ @Test
+ void testGeneratorEmitsMultipleRecordsWithTimestampsAtomically() throws
Exception {
+ final FailingCheckpointedSource<Long> source =
+ FailingCheckpointedSource.of(
+ (subtaskIndex, sequenceNo, output) -> {
+ output.collect(sequenceNo, sequenceNo);
+ output.collect(-sequenceNo, sequenceNo);
+ output.emitWatermark(sequenceNo);
+ },
+ 3,
+ Types.LONG);
+ // subtask 1 never fails, so the whole sequence is observable without
a checkpoint
+ final SourceReader<Long, SequenceSplit> reader =
+ readerWithSplit(source, new SequenceSplit(1, 0));
+ final TestReaderOutput<Long> output = new TestReaderOutput<>();
+
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ assertThat(output.getCollected()).containsExactly(0L, 0L);
+
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ assertThat(output.getCollected()).containsExactly(0L, 0L, 1L, -1L);
+ }
+
+ @Test
+ void testRejectsFewerThanTwoEmitCalls() {
+ assertThatThrownBy(
+ () ->
+ FailingCheckpointedSource.of(
+ (subtaskIndex, sequenceNo, output) ->
+ output.collect(sequenceNo),
+ 1,
+ Types.LONG))
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test
+ void testAbortedArmedCheckpointRearmsOnNextSnapshot() throws Exception {
+ final SourceReader<Long, SequenceSplit> reader =
+ readerWithSplit(failingSource(), new SequenceSplit(0, 0));
+ final TestReaderOutput<Long> output = new TestReaderOutput<>();
+
+ final long armHoldPosition = FAILURE_POSITION / 2 + 1;
+ for (long i = 0; i < armHoldPosition; i++) {
+ reader.pollNext(output);
+ }
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
+ reader.snapshotState(1);
+ reader.notifyCheckpointAborted(1);
+ // the aborted checkpoint must not trigger the failure and must not
stay armed; the
+ // reader holds at the arming position again
+ reader.notifyCheckpointComplete(1);
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+
+ reader.snapshotState(2);
+ for (long i = armHoldPosition; i < FAILURE_POSITION; i++) {
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.MORE_AVAILABLE);
+ }
+
assertThat(reader.pollNext(output)).isEqualTo(InputStatus.NOTHING_AVAILABLE);
+ reader.notifyCheckpointComplete(2);
+ assertThatThrownBy(() -> reader.pollNext(output))
+ .isInstanceOf(FlinkRuntimeException.class)
+ .hasMessage("Artificial Failure");
+ }
+
+ @Test
+ void testEnumeratorAssignsOneSplitPerSubtaskOnFreshStart() throws
Exception {
+ final MockSplitEnumeratorContext<SequenceSplit> context =
+ new MockSplitEnumeratorContext<>(2);
+ final SplitEnumerator<SequenceSplit, Void> enumerator =
+ failingSource().createEnumerator(context);
+
+ enumerator.addReader(0);
+ enumerator.addReader(1);
+
+ assertThat(assignedSplits(context, 0)).containsExactly(new
SequenceSplit(0, 0));
+ assertThat(assignedSplits(context, 1)).containsExactly(new
SequenceSplit(1, 0));
+ }
+
+ @Test
+ void testEnumeratorDoesNotReassignOnReRegistration() throws Exception {
+ // Regression test: on a partial failover the enumerator instance
survives and the
+ // restarted reader re-registers; it recovers its split from its own
checkpointed state,
+ // so the enumerator must not assign a second, fresh split.
+ final MockSplitEnumeratorContext<SequenceSplit> context =
+ new MockSplitEnumeratorContext<>(1);
+ final SplitEnumerator<SequenceSplit, Void> enumerator =
+ failingSource().createEnumerator(context);
+
+ enumerator.addReader(0);
+ enumerator.addReader(0);
+
+ assertThat(assignedSplits(context, 0)).containsExactly(new
SequenceSplit(0, 0));
+ }
+
+ @Test
+ void testEnumeratorReassignsSplitsReturnedAfterFailure() throws Exception {
+ final MockSplitEnumeratorContext<SequenceSplit> context =
+ new MockSplitEnumeratorContext<>(1);
+ final SplitEnumerator<SequenceSplit, Void> enumerator =
+ failingSource().createEnumerator(context);
+
+ enumerator.addReader(0);
+ enumerator.addSplitsBack(
+ Collections.singletonList(new SequenceSplit(0,
FAILURE_POSITION)), 0);
+ enumerator.addReader(0);
+
+ assertThat(assignedSplits(context, 0))
+ .containsExactly(new SequenceSplit(0, 0), new SequenceSplit(0,
FAILURE_POSITION));
+ }
+
+ @Test
+ void testRestoredEnumeratorAssignsNothingOnRegistration() throws Exception
{
+ final MockSplitEnumeratorContext<SequenceSplit> context =
+ new MockSplitEnumeratorContext<>(1);
+ final SplitEnumerator<SequenceSplit, Void> enumerator =
+ failingSource().restoreEnumerator(context, null);
+
+ enumerator.addReader(0);
+
+ assertThat(context.getSplitsAssignmentSequence()).isEmpty();
+ }
+
+ private static List<SequenceSplit> assignedSplits(
+ MockSplitEnumeratorContext<SequenceSplit> context, int subtask) {
+ return context.getSplitsAssignmentSequence().stream()
+ .flatMap(
+ assignment ->
+ assignment
+ .assignment()
+ .getOrDefault(subtask,
Collections.emptyList())
+ .stream())
+ .collect(Collectors.toList());
+ }
+
+ @Test
+ void testSplitSerializerRoundTrip() throws Exception {
+ final SequenceSplit split = new SequenceSplit(3, 42);
+ final byte[] bytes = SequenceSplit.SERIALIZER.serialize(split);
+ final SequenceSplit restored =
+
SequenceSplit.SERIALIZER.deserialize(SequenceSplit.SERIALIZER.getVersion(),
bytes);
+ assertThat(restored.getSubtaskIndex()).isEqualTo(3);
+ assertThat(restored.getPosition()).isEqualTo(42);
+ assertThat(restored.splitId()).isEqualTo(split.splitId());
+ }
+}
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java
index 2a1f82028dd..06da892f2d1 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeAllWindowCheckpointingITCase.java
@@ -18,8 +18,11 @@
package org.apache.flink.test.checkpointing;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.typeinfo.TypeHint;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.configuration.Configuration;
@@ -27,16 +30,17 @@ import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.RpcOptions;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import
org.apache.flink.streaming.api.functions.windowing.RichAllWindowFunction;
import
org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import
org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.util.RestartStrategyUtils;
-import org.apache.flink.test.checkpointing.utils.FailingSource;
import org.apache.flink.test.checkpointing.utils.IntType;
import org.apache.flink.test.checkpointing.utils.ValidatingSink;
import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.test.util.source.FailingCheckpointedSource;
import org.apache.flink.util.Collector;
import org.apache.flink.util.TestLoggerExtension;
@@ -75,6 +79,22 @@ class EventTimeAllWindowCheckpointingITCase {
return config;
}
+ private static DataStream<Tuple2<Long, IntType>> createFailingSourceStream(
+ StreamExecutionEnvironment env,
+ int numKeys,
+ int numElementsPerWindow,
+ int numElementsPerKey) {
+ final FailingCheckpointedSource<Tuple2<Long, IntType>> source =
+ FailingCheckpointedSource.of(
+ new
EventTimeWindowCheckpointingITCase.KeyedEventTimeGenerator(
+ numKeys, numElementsPerWindow),
+ numElementsPerKey,
+ TypeInformation.of(new TypeHint<Tuple2<Long,
IntType>>() {}));
+ // like the legacy FailingSource, the source is non-parallel
+ return env.fromSource(source, WatermarkStrategy.noWatermarks(),
"Failing Source")
+ .setParallelism(1);
+ }
+
// ------------------------------------------------------------------------
@Test
@@ -88,11 +108,7 @@ class EventTimeAllWindowCheckpointingITCase {
env.enableCheckpointing(100);
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 1, 0L);
- env.addSource(
- new FailingSource(
- new
EventTimeWindowCheckpointingITCase.KeyedEventTimeGenerator(
- numKeys, windowSize),
- numElementsPerKey))
+ createFailingSourceStream(env, numKeys, windowSize, numElementsPerKey)
.rebalance()
.windowAll(TumblingEventTimeWindows.of(Duration.ofMillis(windowSize)))
.apply(
@@ -160,11 +176,7 @@ class EventTimeAllWindowCheckpointingITCase {
env.enableCheckpointing(100);
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 1, 0L);
- env.addSource(
- new FailingSource(
- new
EventTimeWindowCheckpointingITCase.KeyedEventTimeGenerator(
- numKeys, windowSlide),
- numElementsPerKey))
+ createFailingSourceStream(env, numKeys, windowSlide, numElementsPerKey)
.rebalance()
.windowAll(
SlidingEventTimeWindows.of(
@@ -233,11 +245,7 @@ class EventTimeAllWindowCheckpointingITCase {
env.enableCheckpointing(100);
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 1, 0L);
- env.addSource(
- new FailingSource(
- new
EventTimeWindowCheckpointingITCase.KeyedEventTimeGenerator(
- numKeys, windowSize),
- numElementsPerKey))
+ createFailingSourceStream(env, numKeys, windowSize, numElementsPerKey)
.rebalance()
.windowAll(TumblingEventTimeWindows.of(Duration.ofMillis(windowSize)))
.reduce(
@@ -309,11 +317,7 @@ class EventTimeAllWindowCheckpointingITCase {
env.enableCheckpointing(100);
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 1, 0L);
- env.addSource(
- new FailingSource(
- new
EventTimeWindowCheckpointingITCase.KeyedEventTimeGenerator(
- numKeys, windowSlide),
- numElementsPerKey))
+ createFailingSourceStream(env, numKeys, windowSlide, numElementsPerKey)
.rebalance()
.windowAll(
SlidingEventTimeWindows.of(
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
index c062a69de19..a4b990a55f5 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/EventTimeWindowCheckpointingITCase.java
@@ -52,6 +52,7 @@ import
org.apache.flink.test.checkpointing.utils.FailingSource;
import org.apache.flink.test.checkpointing.utils.IntType;
import org.apache.flink.test.checkpointing.utils.ValidatingSink;
import org.apache.flink.test.util.MiniClusterWithClientResource;
+import org.apache.flink.test.util.source.FailingCheckpointedSource;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameter;
import
org.apache.flink.testutils.junit.extensions.parameterized.ParameterizedTestExtension;
import org.apache.flink.testutils.junit.extensions.parameterized.Parameters;
@@ -795,7 +796,11 @@ class EventTimeWindowCheckpointingITCase {
}
}
- static class KeyedEventTimeGenerator implements
FailingSource.EventEmittingGenerator {
+ // implements both generator interfaces until all FailingSource consumers
are migrated to
+ // FailingCheckpointedSource
+ static class KeyedEventTimeGenerator
+ implements FailingSource.EventEmittingGenerator,
+ FailingCheckpointedSource.EventGenerator<Tuple2<Long,
IntType>> {
private final int keyUniverseSize;
private final int watermarkTrailing;
@@ -819,6 +824,19 @@ class EventTimeWindowCheckpointingITCase {
ctx.emitWatermark(new Watermark(eventSequenceNo -
watermarkTrailing));
}
+
+ @Override
+ public void emit(
+ int subtaskIndex,
+ long sequenceNo,
+ FailingCheckpointedSource.GeneratorOutput<Tuple2<Long,
IntType>> output) {
+ final IntType intTypeNext = new IntType((int) sequenceNo);
+ for (long i = 0; i < keyUniverseSize; i++) {
+ output.collect(new Tuple2<>(i, intTypeNext), sequenceNo);
+ }
+
+ output.emitWatermark(sequenceNo - watermarkTrailing);
+ }
}
private int numElementsPerKey() {
diff --git
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ProcessingTimeWindowCheckpointingITCase.java
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ProcessingTimeWindowCheckpointingITCase.java
index 8471e9ec187..f26739a6051 100644
---
a/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ProcessingTimeWindowCheckpointingITCase.java
+++
b/flink-tests/src/test/java/org/apache/flink/test/checkpointing/ProcessingTimeWindowCheckpointingITCase.java
@@ -18,26 +18,30 @@
package org.apache.flink.test.checkpointing;
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.OpenContext;
import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.typeinfo.TypeHint;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.TaskManagerOptions;
import org.apache.flink.runtime.testutils.MiniClusterResourceConfiguration;
+import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
-import org.apache.flink.streaming.api.functions.source.legacy.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.RichWindowFunction;
import
org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import
org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.util.RestartStrategyUtils;
-import org.apache.flink.test.checkpointing.utils.FailingSource;
import org.apache.flink.test.checkpointing.utils.IntType;
import org.apache.flink.test.checkpointing.utils.ValidatingSink;
import org.apache.flink.test.junit5.MiniClusterExtension;
+import org.apache.flink.test.util.source.FailingCheckpointedSource;
import org.apache.flink.util.Collector;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.TestLoggerExtension;
import org.junit.jupiter.api.Test;
@@ -75,6 +79,20 @@ class ProcessingTimeWindowCheckpointingITCase {
return config;
}
+ private static DataStream<Tuple2<Long, IntType>> createFailingSourceStream(
+ StreamExecutionEnvironment env, int numElements) {
+ final FailingCheckpointedSource<Tuple2<Long, IntType>> source =
+ FailingCheckpointedSource.of(
+ new Generator(),
+ numElements,
+ TypeInformation.of(new TypeHint<Tuple2<Long,
IntType>>() {}))
+ // the sink terminates the job once all windows are
validated
+ .withIdleAfterEmission();
+ // like the legacy FailingSource, the source is non-parallel
+ return env.fromSource(source, WatermarkStrategy.noWatermarks(),
"Failing Source")
+ .setParallelism(1);
+ }
+
// ------------------------------------------------------------------------
@Test
@@ -90,7 +108,7 @@ class ProcessingTimeWindowCheckpointingITCase {
SinkValidatorUpdaterAndChecker updaterAndChecker =
new SinkValidatorUpdaterAndChecker(numElements, 1);
- env.addSource(new FailingSource(new Generator(), numElements, true))
+ createFailingSourceStream(env, numElements)
.rebalance()
.keyBy(x -> x.f0)
.window(TumblingProcessingTimeWindows.of(Duration.ofMillis(100)))
@@ -143,7 +161,7 @@ class ProcessingTimeWindowCheckpointingITCase {
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 1, 0L);
SinkValidatorUpdaterAndChecker updaterAndChecker =
new SinkValidatorUpdaterAndChecker(numElements, 3);
- env.addSource(new FailingSource(new Generator(), numElements, true))
+ createFailingSourceStream(env, numElements)
.rebalance()
.keyBy(x -> x.f0)
.window(
@@ -198,7 +216,7 @@ class ProcessingTimeWindowCheckpointingITCase {
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 1, 0L);
SinkValidatorUpdaterAndChecker updaterAndChecker =
new SinkValidatorUpdaterAndChecker(numElements, 1);
- env.addSource(new FailingSource(new Generator(), numElements, true))
+ createFailingSourceStream(env, numElements)
.map(
new MapFunction<Tuple2<Long, IntType>, Tuple2<Long,
IntType>>() {
@Override
@@ -236,7 +254,7 @@ class ProcessingTimeWindowCheckpointingITCase {
RestartStrategyUtils.configureFixedDelayRestartStrategy(env, 1, 0L);
SinkValidatorUpdaterAndChecker updaterAndChecker =
new SinkValidatorUpdaterAndChecker(numElements, 3);
- env.addSource(new FailingSource(new Generator(), numElements, true))
+ createFailingSourceStream(env, numElements)
.map(
new MapFunction<Tuple2<Long, IntType>, Tuple2<Long,
IntType>>() {
@Override
@@ -268,12 +286,24 @@ class ProcessingTimeWindowCheckpointingITCase {
// Utilities
// ------------------------------------------------------------------------
- static class Generator implements FailingSource.EventEmittingGenerator {
+ static class Generator
+ implements FailingCheckpointedSource.EventGenerator<Tuple2<Long,
IntType>> {
@Override
- public void emitEvent(
- SourceFunction.SourceContext<Tuple2<Long, IntType>> ctx, int
eventSequenceNo) {
- ctx.collect(new Tuple2<>((long) eventSequenceNo, new
IntType(eventSequenceNo)));
+ public void emit(
+ int subtaskIndex,
+ long sequenceNo,
+ FailingCheckpointedSource.GeneratorOutput<Tuple2<Long,
IntType>> output) {
+ output.collect(new Tuple2<>(sequenceNo, new IntType((int)
sequenceNo)));
+ // Spread the elements over many processing-time windows, like the
legacy
+ // FailingSource did, so that every run restores open window state
and rolls back
+ // sink output on recovery. Duplicate detection itself does not
depend on this.
+ try {
+ Thread.sleep(1);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new FlinkRuntimeException(e);
+ }
}
}
