Skip to content

[FLINK-37605][runtime] Infer checkpoint id on endInput in sink #26433

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Apr 14, 2025
Merged

[FLINK-37605][runtime] Infer checkpoint id on endInput in sink #26433

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
778d976
[FLINK-37605][runtime] Clarify contract of endInput
AHeise Apr 9, 2025
17f0192
[FLINK-37605][runtime] Remove obsolete sink tests
AHeise Apr 13, 2025
e5f0df8
[FLINK-37605][runtime] Cleanup writer test
AHeise Apr 13, 2025
941e510
[FLINK-37605][runtime] Infer checkpoint id on endInput in sink
AHeise Apr 3, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 9 additions & 4 deletions .../main/java/org/apache/flink/connector/file/sink/compactor/operator/CompactorOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import org.apache.flink.core.memory.DataInputDeserializer;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputSerializer;
import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
import org.apache.flink.streaming.api.connector.sink2.CommittableMessage;
Expand Down Expand Up @@ -107,6 +108,8 @@ public class CompactorOperator
// submitted again while restoring
private ListState<Map<Long, List<CompactorRequest>>> remainingRequestsState;

private long lastKnownCheckpointId = CheckpointIDCounter.INITIAL_CHECKPOINT_ID - 1;

public CompactorOperator(
StreamOperatorParameters<CommittableMessage<FileSinkCommittable>> parameters,
FileCompactStrategy strategy,
Expand Down Expand Up @@ -139,15 +142,16 @@ public void processElement(StreamRecord<CompactorRequest> element) throws Except
@Override
public void endInput() throws Exception {
// add collecting requests into the final snapshot
checkpointRequests.put(CommittableMessage.EOI, collectingRequests);
long checkpointId = lastKnownCheckpointId + 1;
checkpointRequests.put(checkpointId, collectingRequests);
collectingRequests = new ArrayList<>();

// submit all requests and wait until they are done
submitUntil(CommittableMessage.EOI);
submitUntil(checkpointId);
assert checkpointRequests.isEmpty();

getAllTasksFuture().join();
emitCompacted(CommittableMessage.EOI);
emitCompacted(checkpointId);
assert compactingRequests.isEmpty();
}

Expand Down Expand Up @@ -225,6 +229,8 @@ private void submitUntil(long checkpointId) {
}

private void emitCompacted(long checkpointId) throws Exception {
lastKnownCheckpointId = checkpointId;

List<FileSinkCommittable> compacted = new ArrayList<>();
Iterator<Tuple2<CompactorRequest, CompletableFuture<Iterable<FileSinkCommittable>>>> iter =
compactingRequests.iterator();
Expand Down Expand Up @@ -252,7 +258,6 @@ private void emitCompacted(long checkpointId) throws Exception {
getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks(),
checkpointId,
compacted.size(),
compacted.size(),
0);
output.collect(new StreamRecord<>(summary));
for (FileSinkCommittable c : compacted) {
Expand Down
12 changes: 11 additions & 1 deletion ...time/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableMessage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,10 @@ public interface CommittableMessage<CommT> {
/**
* Special value for checkpointId for the end of input in case of batch commit or final
* checkpoint.
*
* @deprecated the special value is not used anymore at all (remove with Flink 2.2)
*/
@Deprecated(forRemoval = true)
long EOI = Long.MAX_VALUE;

/** The subtask that created this committable. */
Expand All @@ -35,6 +38,13 @@ public interface CommittableMessage<CommT> {
/**
* Returns the checkpoint id or EOI if this message belong to the final checkpoint or the batch
* commit.
*
* @deprecated the special value EOI is not used anymore
*/
long getCheckpointIdOrEOI();
@Deprecated(forRemoval = true)
default long getCheckpointIdOrEOI() {
return getCheckpointId();
}

long getCheckpointId();
}
4 changes: 2 additions & 2 deletions ...time/src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableSummary.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,7 +42,7 @@ public class CommittableSummary<CommT> implements CommittableMessage<CommT> {
/** The number of committables coming from the given subtask in the particular checkpoint. */
private final int numberOfCommittables;

@Deprecated
@Deprecated(forRemoval = true)
/** The number of committables that have not been successfully committed. */
private final int numberOfPendingCommittables;

Expand Down Expand Up @@ -88,7 +88,7 @@ public int getNumberOfSubtasks() {
return numberOfSubtasks;
}

public long getCheckpointIdOrEOI() {
public long getCheckpointId() {
return checkpointId;
}

Expand Down
2 changes: 1 addition & 1 deletion .../src/main/java/org/apache/flink/streaming/api/connector/sink2/CommittableWithLineage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ public int getSubtaskId() {
return subtaskId;
}

public long getCheckpointIdOrEOI() {
public long getCheckpointId() {
return checkpointId;
}

Expand Down
6 changes: 6 additions & 0 deletions flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/BoundedOneInput.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,12 @@ public interface BoundedOneInput {
/**
* It is notified that no more data will arrive from the input.
*
* <p>Stateful operators need to be aware that a restart with rescaling may occur after
* receiving this notification. A changed source split assignment may imply that the same
* subtask of this operator that received endInput, has its state after endInput snapshotted,
* and will receive new data after restart. Hence, the state should not contain any finalization
* that would make it impossible to process new data.
*
* <p><b>WARNING:</b> It is not safe to use this method to commit any transactions or other side
* effects! You can use this method to flush any buffered data that can later on be committed
* e.g. in a {@link StreamOperator#notifyCheckpointComplete(long)}.
Expand Down
23 changes: 10 additions & 13 deletions ...me/src/main/java/org/apache/flink/streaming/runtime/operators/sink/CommitterOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.flink.configuration.SinkOptions;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.metrics.groups.SinkCommitterMetricGroup;
import org.apache.flink.runtime.checkpoint.CheckpointIDCounter;
import org.apache.flink.runtime.metrics.groups.InternalSinkCommitterMetricGroup;
import org.apache.flink.runtime.state.StateInitializationContext;
import org.apache.flink.runtime.state.StateSnapshotContext;
Expand All @@ -51,7 +52,6 @@
import java.util.Collections;
import java.util.OptionalLong;

import static org.apache.flink.streaming.api.connector.sink2.CommittableMessage.EOI;
import static org.apache.flink.util.IOUtils.closeAll;
import static org.apache.flink.util.Preconditions.checkNotNull;

Expand All @@ -76,11 +76,9 @@ class CommitterOperator<CommT> extends AbstractStreamOperator<CommittableMessage
private SinkCommitterMetricGroup metricGroup;
private Committer<CommT> committer;
private CommittableCollector<CommT> committableCollector;
private long lastCompletedCheckpointId = -1;
private long lastCompletedCheckpointId = CheckpointIDCounter.INITIAL_CHECKPOINT_ID - 1;
private int maxRetries;

private boolean endInput = false;

/** The operator's state descriptor. */
private static final ListStateDescriptor<byte[]> STREAMING_COMMITTER_RAW_STATES_DESC =
new ListStateDescriptor<>(
Expand Down Expand Up @@ -134,11 +132,11 @@ public void initializeState(StateInitializationContext context) throws Exception
getRuntimeContext().getTaskInfo().getIndexOfThisSubtask(),
getRuntimeContext().getTaskInfo().getNumberOfParallelSubtasks(),
metricGroup));
if (context.isRestored()) {
if (checkpointId.isPresent()) {
committableCollectorState.get().forEach(cc -> committableCollector.merge(cc));
lastCompletedCheckpointId = checkpointId.getAsLong();
// try to re-commit recovered transactions as quickly as possible
commitAndEmitCheckpoints();
commitAndEmitCheckpoints(lastCompletedCheckpointId);
}
}

Expand All @@ -151,24 +149,23 @@ public void snapshotState(StateSnapshotContext context) throws Exception {

@Override
public void endInput() throws Exception {
endInput = true;
if (!isCheckpointingEnabled || isBatchMode) {
// There will be no final checkpoint, all committables should be committed here
commitAndEmitCheckpoints();
commitAndEmitCheckpoints(lastCompletedCheckpointId + 1);
}
}

@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
lastCompletedCheckpointId = Math.max(lastCompletedCheckpointId, checkpointId);
commitAndEmitCheckpoints();
commitAndEmitCheckpoints(Math.max(lastCompletedCheckpointId, checkpointId));
}

private void commitAndEmitCheckpoints() throws IOException, InterruptedException {
long completedCheckpointId = endInput ? EOI : lastCompletedCheckpointId;
private void commitAndEmitCheckpoints(long checkpointId)
throws IOException, InterruptedException {
lastCompletedCheckpointId = checkpointId;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: probably a basic question, but shouldn't we update the lastCompletedCheckpointId variable after we have completed the checkpoint, which I assume happens in the subsequent for loop? I was expecting the lastCompletedCheckpointId to be updated after the checkpointing loop in case there was an error during the checkpointing.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In general, transient state is lost on error. So whether we update before or after the loop doesn't matter because the exception will lead to a fail-over and everything is recalculated on recovery. Since everything is called from the main task thread (mailbox thread), there is no interleaving possible of this call and another call like endInput.

Now in this specific case, lastCompletedCheckpointId refers to the completed checkpoint id of Flink as a whole. Since this value is primarily set through notifyCheckpointCompleted, the checkpoint is already completed before the start of the method. So I'd like to keep it as the first statement because it's easier to read than if it's done at the end of the method.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the explanation @AHeise - that makes sense

for (CheckpointCommittableManager<CommT> checkpointManager :
committableCollector.getCheckpointCommittablesUpTo(completedCheckpointId)) {
committableCollector.getCheckpointCommittablesUpTo(checkpointId)) {
// ensure that all committables of the first checkpoint are fully committed before
// attempting the next committable
commitAndEmit(checkpointManager);
Expand Down
58 changes: 9 additions & 49 deletions ...e/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,6 @@
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.BooleanSerializer;
import org.apache.flink.api.common.typeutils.base.array.BytePrimitiveArraySerializer;
import org.apache.flink.api.connector.sink2.CommittingSinkWriter;
import org.apache.flink.api.connector.sink2.Sink;
Expand Down Expand Up @@ -54,8 +53,6 @@
import org.apache.flink.streaming.runtime.tasks.StreamTask;
import org.apache.flink.util.UserCodeClassLoader;

import org.apache.flink.shaded.guava33.com.google.common.collect.Lists;

import javax.annotation.Nullable;

import java.io.IOException;
Expand All @@ -64,6 +61,7 @@
import java.util.List;
import java.util.OptionalLong;

import static org.apache.flink.runtime.checkpoint.CheckpointIDCounter.INITIAL_CHECKPOINT_ID;
import static org.apache.flink.util.IOUtils.closeAll;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
Expand Down Expand Up @@ -93,13 +91,6 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab
@Nullable private final SimpleVersionedSerializer<CommT> committableSerializer;
private final List<CommT> legacyCommittables = new ArrayList<>();

/**
* Used to remember that EOI has already happened so that we don't emit the last committables of
* the final checkpoints twice.
*/
private static final ListStateDescriptor<Boolean> END_OF_INPUT_STATE_DESC =
new ListStateDescriptor<>("end_of_input_state", BooleanSerializer.INSTANCE);

/** The runtime information of the input element. */
private final Context<InputT> context;

Expand All @@ -118,10 +109,7 @@ class SinkWriterOperator<InputT, CommT> extends AbstractStreamOperator<Committab

private boolean endOfInput = false;

/**
* Remembers the endOfInput state for (final) checkpoints iff the operator emits committables.
*/
@Nullable private ListState<Boolean> endOfInputState;
private long lastKnownCheckpointId = INITIAL_CHECKPOINT_ID - 1;

SinkWriterOperator(
StreamOperatorParameters<CommittableMessage<CommT>> parameters,
Expand Down Expand Up @@ -164,8 +152,10 @@ protected void setup(
@Override
public void initializeState(StateInitializationContext context) throws Exception {
super.initializeState(context);
WriterInitContext initContext = createInitContext(context.getRestoredCheckpointId());
if (context.isRestored()) {
OptionalLong restoredCheckpointId = context.getRestoredCheckpointId();
WriterInitContext initContext = createInitContext(restoredCheckpointId);
if (restoredCheckpointId.isPresent()) {
lastKnownCheckpointId = restoredCheckpointId.getAsLong();
if (committableSerializer != null) {
final ListState<List<CommT>> legacyCommitterState =
new SimpleVersionedListState<>(
Expand All @@ -179,41 +169,12 @@ public void initializeState(StateInitializationContext context) throws Exception
}

sinkWriter = writerStateHandler.createWriter(initContext, context);

if (emitDownstream) {
// Figure out if we have seen end of input before and if we can suppress creating
// transactions and sending them downstream to the CommitterOperator. We have the
// following
// cases:
// 1. state is empty:
// - First time initialization
// - Restoring from a previous version of Flink that didn't handle EOI
// - Upscaled from a final or regular checkpoint
// In all cases, we regularly handle EOI, potentially resulting in duplicate summaries
// that the CommitterOperator needs to handle.
// 2. state is not empty:
// - This implies Flink restores from a version that handles EOI.
// - If there is one entry, no rescaling happened (for this subtask), so if it's true,
// we recover from a final checkpoint (for this subtask) and can ignore another EOI
// else we have a regular checkpoint.
// - If there are multiple entries, Flink downscaled, and we need to check if all are
// true and do the same as above. As soon as one entry is false, we regularly start
// the writer and potentially emit duplicate summaries if we indeed recovered from a
// final checkpoint.
endOfInputState = context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
ArrayList<Boolean> previousState = Lists.newArrayList(endOfInputState.get());
endOfInput = !previousState.isEmpty() && !previousState.contains(false);
}
}

@Override
public void snapshotState(StateSnapshotContext context) throws Exception {
super.snapshotState(context);
writerStateHandler.snapshotState(context.getCheckpointId());
if (endOfInputState != null) {
endOfInputState.clear();
endOfInputState.add(this.endOfInput);
}
}

@Override
Expand Down Expand Up @@ -243,17 +204,16 @@ public void processWatermark(Watermark mark) throws Exception {

@Override
public void endInput() throws Exception {
LOG.info("Received endInput");
if (!endOfInput) {
endOfInput = true;
if (endOfInputState != null) {
endOfInputState.add(true);
}
sinkWriter.flush(true);
emitCommittables(CommittableMessage.EOI);
emitCommittables(lastKnownCheckpointId + 1);
}
}

private void emitCommittables(long checkpointId) throws IOException, InterruptedException {
lastKnownCheckpointId = checkpointId;
if (!emitDownstream) {
// To support SinkV1 topologies with only a writer we have to call prepareCommit
// although no committables are forwarded
Expand Down
1 change: 1 addition & 0 deletions ...flink/streaming/runtime/operators/sink/committables/CheckpointCommittableManagerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,7 @@ void addSummary(CommittableSummary<CommT> summary) {
summary.getSubtaskId(),
checkpointId,
metricGroup);
// Remove branch once CommittableMessage.EOI has been removed (earliest 2.2)
if (checkpointId == CommittableMessage.EOI) {
SubtaskCommittableManager<CommT> merged =
subtasksCommittableManagers.merge(
Expand Down
12 changes: 0 additions & 12 deletions .../org/apache/flink/streaming/runtime/operators/sink/committables/CommittableCollector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,6 @@
import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.Objects;
import java.util.Optional;
import java.util.TreeMap;
import java.util.stream.Collectors;

Expand All @@ -49,8 +48,6 @@
*/
@Internal
public class CommittableCollector<CommT> {
private static final long EOI = Long.MAX_VALUE;

/** Mapping of checkpoint id to {@link CheckpointCommittableManagerImpl}. */
private final NavigableMap<Long, CheckpointCommittableManagerImpl<CommT>>
checkpointCommittables;
Expand Down Expand Up @@ -144,15 +141,6 @@ public Collection<? extends CheckpointCommittableManager<CommT>> getCheckpointCo
return new ArrayList<>(checkpointCommittables.headMap(checkpointId, true).values());
}

/**
* Returns {@link CheckpointCommittableManager} belonging to the last input.
*
* @return {@link CheckpointCommittableManager}
*/
public Optional<CheckpointCommittableManager<CommT>> getEndOfInputCommittable() {
return Optional.ofNullable(checkpointCommittables.get(EOI));
}

/**
* Returns whether all {@link CheckpointCommittableManager} currently hold by the collector are
* either committed or failed.
Expand Down
Loading