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 250f6252a38267..7b37b2072ab534 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. ${project.version} test + + + + org.apache.flink + flink-core + ${project.version} + test-jar + test + \ 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 00000000000000..bd5ed6f360aac5 --- /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,438 @@ +/* + * 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.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.util.FlinkRuntimeException; + +import javax.annotation.Nullable; + +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.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. + * + *

Emission holds just past the {@link FailurePolicy} arming threshold until a snapshot arms the + * failure, continues to the failure position, and throws once the armed checkpoint completes. + * Recovery therefore always replays the records between the arming and failure positions, + * independent of emission speed. A failing policy requires checkpointing to 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. + * + *

{@link #withIdleAfterEmission()} keeps the source idle instead of finishing (for + * processing-time tests); {@link #withSimulatedStateLossOnRecovery()} deliberately breaks + * exactly-once, only for validating that a test still detects duplicates. + */ +@Experimental +public class FailingCheckpointedSource + implements Source, ResultTypeQueryable { + + 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 extends Serializable { + void emit(int subtaskIndex, long sequenceNo, GeneratorOutput output); + } + + /** Output for the {@link EventGenerator}, bridging to the reader output. */ + public interface GeneratorOutput { + void collect(T record); + + void collect(T record, long timestamp); + + void emitWatermark(long timestamp); + } + + /** Determines whether and when the source fails. */ + public static final class FailurePolicy implements Serializable { + + private static final long serialVersionUID = 1L; + private static final long NEVER = -1; + + private final long failAfterEmitCalls; + private final long armAfterEmitCalls; + + private FailurePolicy(long failAfterEmitCalls, long armAfterEmitCalls) { + this.failAfterEmitCalls = failAfterEmitCalls; + this.armAfterEmitCalls = armAfterEmitCalls; + } + + /** + * Fails subtask 0 once, after it has emitted {@code emitCalls} generator invocations and a + * checkpoint taken past half of that has completed. + */ + public static FailurePolicy failAfterEmitCalls(long emitCalls) { + checkArgument(emitCalls >= 1, "emitCalls must be >= 1"); + return new FailurePolicy(emitCalls, emitCalls / 2); + } + + /** Never fails; the source just emits and finishes. */ + public static FailurePolicy neverFail() { + return new FailurePolicy(NEVER, NEVER); + } + + boolean fails() { + return failAfterEmitCalls != NEVER; + } + } + + private final EventGenerator generator; + private final long emitCallsPerSubtask; + private final FailurePolicy failurePolicy; + private final TypeInformation producedType; + private final boolean idleAfterEmission; + private final boolean simulateStateLossOnRecovery; + + private FailingCheckpointedSource( + EventGenerator generator, + long emitCallsPerSubtask, + FailurePolicy failurePolicy, + TypeInformation producedType, + boolean idleAfterEmission, + boolean simulateStateLossOnRecovery) { + this.generator = checkNotNull(generator); + this.emitCallsPerSubtask = emitCallsPerSubtask; + this.failurePolicy = checkNotNull(failurePolicy); + this.producedType = checkNotNull(producedType); + this.idleAfterEmission = idleAfterEmission; + this.simulateStateLossOnRecovery = simulateStateLossOnRecovery; + checkArgument(emitCallsPerSubtask >= 1, "emitCallsPerSubtask must be >= 1"); + checkArgument( + !failurePolicy.fails() || failurePolicy.failAfterEmitCalls <= emitCallsPerSubtask, + "the failure position must not lie beyond the emitted sequence"); + } + + public static FailingCheckpointedSource of( + EventGenerator generator, + long emitCallsPerSubtask, + FailurePolicy failurePolicy, + TypeInformation producedType) { + return new FailingCheckpointedSource<>( + generator, emitCallsPerSubtask, failurePolicy, producedType, false, 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 withIdleAfterEmission() { + return new FailingCheckpointedSource<>( + generator, + emitCallsPerSubtask, + failurePolicy, + producedType, + true, + simulateStateLossOnRecovery); + } + + /** + * Returns a copy of this source whose readers discard the restored position after recovery. + * Only for validating that a test detects exactly-once violations; never use in a test's normal + * path. + */ + public FailingCheckpointedSource withSimulatedStateLossOnRecovery() { + return new FailingCheckpointedSource<>( + generator, + emitCallsPerSubtask, + failurePolicy, + producedType, + idleAfterEmission, + true); + } + + @Override + public Boundedness getBoundedness() { + return Boundedness.CONTINUOUS_UNBOUNDED; + } + + @Override + public SourceReader createReader(SourceReaderContext readerContext) { + return new FailingCheckpointedSourceReader<>( + generator, + emitCallsPerSubtask, + failurePolicy, + idleAfterEmission, + simulateStateLossOnRecovery); + } + + @Override + public SplitEnumerator createEnumerator( + SplitEnumeratorContext enumContext) { + return new SequenceSplitEnumerator(enumContext, true); + } + + @Override + public SplitEnumerator restoreEnumerator( + SplitEnumeratorContext 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 getSplitSerializer() { + return SequenceSplit.SERIALIZER; + } + + @Override + public SimpleVersionedSerializer getEnumeratorCheckpointSerializer() { + return VoidSerializer.INSTANCE; + } + + @Override + public TypeInformation getProducedType() { + return producedType; + } + + /** Assigns one position split per subtask on fresh starts. */ + private static final class SequenceSplitEnumerator + implements SplitEnumerator { + + private final SplitEnumeratorContext context; + private final boolean assignSplitsOnRegistration; + private final Map splitsToReassign = new HashMap<>(); + private final Set assignedSubtasks = new HashSet<>(); + + private SequenceSplitEnumerator( + SplitEnumeratorContext 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 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 + implements SourceReader { + + private final EventGenerator generator; + private final long emitCallsPerSubtask; + private final FailurePolicy failurePolicy; + private final boolean idleAfterEmission; + private final boolean simulateStateLossOnRecovery; + 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 availability = CompletableFuture.completedFuture(null); + + private FailingCheckpointedSourceReader( + EventGenerator generator, + long emitCallsPerSubtask, + FailurePolicy failurePolicy, + boolean idleAfterEmission, + boolean simulateStateLossOnRecovery) { + this.generator = generator; + this.emitCallsPerSubtask = emitCallsPerSubtask; + this.failurePolicy = failurePolicy; + this.idleAfterEmission = idleAfterEmission; + this.simulateStateLossOnRecovery = simulateStateLossOnRecovery; + } + + @Override + public void start() {} + + @Override + public InputStatus pollNext(ReaderOutput output) { + if (failureRequested && position >= failurePolicy.failAfterEmitCalls) { + throw new FlinkRuntimeException("Artificial Failure"); + } + if (!hasSplit) { + return pause(); + } + if (eligibleToFail) { + if (armedCheckpointId == null && position > failurePolicy.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 >= failurePolicy.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 snapshotState(long checkpointId) { + if (!hasSplit) { + return Collections.emptyList(); + } + // only arm past the threshold, so recovery demonstrably restores progress + if (eligibleToFail + && armedCheckpointId == null + && position > failurePolicy.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 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 = simulateStateLossOnRecovery ? 0 : split.getPosition(); + // restored position > 0 identifies the recovery attempt (no attempt number in + // SourceReaderContext) + eligibleToFail = + failurePolicy.fails() + && split.getSubtaskIndex() == 0 + && split.getPosition() == 0; + availability.complete(null); + } + + @Override + public CompletableFuture isAvailable() { + return availability; + } + + @Override + public void notifyNoMoreSplits() {} + + @Override + public void close() { + availability.complete(null); + } + + private final class BridgingGeneratorOutput implements GeneratorOutput { + + @Nullable private ReaderOutput 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/main/java/org/apache/flink/test/util/source/SequenceSplit.java b/flink-test-utils-parent/flink-test-utils-connector/src/main/java/org/apache/flink/test/util/source/SequenceSplit.java new file mode 100644 index 00000000000000..4108240d7beca8 --- /dev/null +++ b/flink-test-utils-parent/flink-test-utils-connector/src/main/java/org/apache/flink/test/util/source/SequenceSplit.java @@ -0,0 +1,104 @@ +/* + * 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.connector.source.SourceSplit; +import org.apache.flink.core.io.SimpleVersionedSerializer; +import org.apache.flink.core.memory.DataInputDeserializer; +import org.apache.flink.core.memory.DataOutputSerializer; + +import java.io.IOException; +import java.util.Objects; + +/** + * A split carrying the emission position of one source subtask, so that a test source can resume + * from the checkpointed position after recovery. The subtask index assumes a fixed source + * parallelism. + */ +@Experimental +public class SequenceSplit implements SourceSplit { + + /** Serializer for {@link SequenceSplit} instances. */ + public static final SimpleVersionedSerializer SERIALIZER = + new SequenceSplitSerializer(); + + private final int subtaskIndex; + private final long position; + + public SequenceSplit(int subtaskIndex, long position) { + this.subtaskIndex = subtaskIndex; + this.position = position; + } + + public int getSubtaskIndex() { + return subtaskIndex; + } + + public 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 class SequenceSplitSerializer + implements SimpleVersionedSerializer { + + @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()); + } + } +} 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 00000000000000..4ad3b1e3a3b6b1 --- /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,355 @@ +/* + * 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.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 = 10; + + private static FailingCheckpointedSource failingSource() { + return FailingCheckpointedSource.of( + (subtaskIndex, sequenceNo, output) -> output.collect(sequenceNo), + EMIT_CALLS, + FailingCheckpointedSource.FailurePolicy.failAfterEmitCalls(FAILURE_POSITION), + Types.LONG); + } + + private static SourceReader readerWithSplit( + FailingCheckpointedSource source, SequenceSplit split) { + final SourceReader reader = source.createReader(null); + reader.addSplits(Collections.singletonList(split)); + return reader; + } + + private static List range(long startInclusive, long endExclusive) { + return LongStream.range(startInclusive, endExclusive).boxed().collect(Collectors.toList()); + } + + @Test + void testFailsOnceAfterArmedCheckpointCompletes() throws Exception { + final SourceReader reader = + readerWithSplit(failingSource(), new SequenceSplit(0, 0)); + final TestReaderOutput 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 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 reader = + readerWithSplit(failingSource(), new SequenceSplit(0, 0)); + final TestReaderOutput 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 reader = + readerWithSplit(failingSource(), new SequenceSplit(0, FAILURE_POSITION)); + final TestReaderOutput 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 reader = + readerWithSplit(failingSource(), new SequenceSplit(1, 0)); + final TestReaderOutput 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 testNeverFailPolicyEmitsAllAndFinishes() throws Exception { + final FailingCheckpointedSource source = + FailingCheckpointedSource.of( + (subtaskIndex, sequenceNo, output) -> output.collect(sequenceNo), + EMIT_CALLS, + FailingCheckpointedSource.FailurePolicy.neverFail(), + Types.LONG); + final SourceReader reader = + readerWithSplit(source, new SequenceSplit(0, 0)); + final TestReaderOutput 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(0, EMIT_CALLS)); + } + + @Test + void testIdleAfterEmissionKeepsReaderAlive() throws Exception { + final SourceReader reader = + readerWithSplit( + failingSource().withIdleAfterEmission(), + new SequenceSplit(0, FAILURE_POSITION)); + final TestReaderOutput 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 testSimulatedStateLossReemitsFromZero() throws Exception { + // Red-check knob: a restored reader discards its position, producing duplicates that an + // exactly-once test must detect. + final SourceReader reader = + readerWithSplit( + failingSource().withSimulatedStateLossOnRecovery(), + new SequenceSplit(0, FAILURE_POSITION)); + final TestReaderOutput 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(0, EMIT_CALLS)); + } + + @Test + void testGeneratorEmitsMultipleRecordsWithTimestampsAtomically() throws Exception { + final FailingCheckpointedSource source = + FailingCheckpointedSource.of( + (subtaskIndex, sequenceNo, output) -> { + output.collect(sequenceNo, sequenceNo); + output.collect(-sequenceNo, sequenceNo); + output.emitWatermark(sequenceNo); + }, + 3, + FailingCheckpointedSource.FailurePolicy.neverFail(), + Types.LONG); + final SourceReader reader = + readerWithSplit(source, new SequenceSplit(0, 0)); + final TestReaderOutput 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 testAbortedArmedCheckpointRearmsOnNextSnapshot() throws Exception { + final SourceReader reader = + readerWithSplit(failingSource(), new SequenceSplit(0, 0)); + final TestReaderOutput 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 context = + new MockSplitEnumeratorContext<>(2); + final SplitEnumerator 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 context = + new MockSplitEnumeratorContext<>(1); + final SplitEnumerator 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 context = + new MockSplitEnumeratorContext<>(1); + final SplitEnumerator 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 context = + new MockSplitEnumeratorContext<>(1); + final SplitEnumerator enumerator = + failingSource().restoreEnumerator(context, null); + + enumerator.addReader(0); + + assertThat(context.getSplitsAssignmentSequence()).isEmpty(); + } + + private static List assignedSplits( + MockSplitEnumeratorContext 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 2a1f82028ddf19..e5a51e2cb232e0 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.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,24 @@ private static Configuration getConfiguration() { return config; } + private static DataStream> createFailingSourceStream( + StreamExecutionEnvironment env, + int numKeys, + int numElementsPerWindow, + int numElementsPerKey) { + final FailingCheckpointedSource> source = + FailingCheckpointedSource.of( + new EventTimeWindowCheckpointingITCase.KeyedEventTimeGenerator( + numKeys, numElementsPerWindow), + numElementsPerKey, + FailingCheckpointedSource.FailurePolicy.failAfterEmitCalls( + numElementsPerKey / 2), + TypeInformation.of(new TypeHint>() {})); + // like the legacy FailingSource, the source is non-parallel + return env.fromSource(source, WatermarkStrategy.noWatermarks(), "Failing Source") + .setParallelism(1); + } + // ------------------------------------------------------------------------ @Test @@ -88,11 +110,7 @@ void testTumblingTimeWindow() throws Exception { 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 +178,7 @@ void testSlidingTimeWindow() throws Exception { 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 +247,7 @@ void testPreAggregatedTumblingTimeWindow() throws Exception { 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 +319,7 @@ void testPreAggregatedSlidingTimeWindow() throws Exception { 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 c062a69de19b85..a4b990a55f5e83 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.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 @@ public boolean checkResult(Map windowCounts) { } } - 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> { private final int keyUniverseSize; private final int watermarkTrailing; @@ -819,6 +824,19 @@ public void emitEvent( ctx.emitWatermark(new Watermark(eventSequenceNo - watermarkTrailing)); } + + @Override + public void emit( + int subtaskIndex, + long sequenceNo, + FailingCheckpointedSource.GeneratorOutput> 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() {