RocMarshal commented on code in PR #27883:
URL: https://github.com/apache/flink/pull/27883#discussion_r3097463722
##########
flink-connectors/flink-connector-datagen-test/src/test/java/org/apache/flink/connector/datagen/source/DataGeneratorSourceITCase.java:
##########
@@ -254,21 +255,42 @@ public TypeInformation getTypeInformation() {
}
}
+ /**
+ * A filter that only passes through elements received before the first
checkpoint completes.
+ *
+ * <p>The filter stops collecting elements in {@link
#notifyCheckpointComplete(long)} rather
+ * than in {@link #snapshotState(FunctionSnapshotContext)}, to avoid a
race condition where the
+ * checkpoint barrier arrives at this operator before all upstream
elements (emitted in the same
+ * checkpoint cycle) have been processed. Using {@code
notifyCheckpointComplete} ensures that
+ * the checkpoint has fully propagated through the pipeline before we stop
collecting.
+ */
private static class FirstCheckpointFilter
- implements FlatMapFunction<Long, Long>, CheckpointedFunction {
+ implements FlatMapFunction<Long, Long>, CheckpointedFunction,
CheckpointListener {
- private volatile boolean firstCheckpoint = true;
+ private volatile boolean firstCheckpointCompleted = false;
+ private long firstCheckpointId = Long.MIN_VALUE;
@Override
public void flatMap(Long value, Collector<Long> out) throws Exception {
- if (firstCheckpoint) {
+ if (!firstCheckpointCompleted) {
out.collect(value);
}
}
@Override
public void snapshotState(FunctionSnapshotContext context) throws
Exception {
- firstCheckpoint = false;
+ // Record the ID of the first checkpoint so we can stop collecting
when it completes.
+ if (firstCheckpointId == Long.MIN_VALUE) {
+ firstCheckpointId = context.getCheckpointId();
+ }
+ }
+
+ @Override
+ public void notifyCheckpointComplete(long checkpointId) throws
Exception {
+ // Stop collecting elements once the first checkpoint has
completed.
+ if (checkpointId >= firstCheckpointId && firstCheckpointId !=
Long.MIN_VALUE) {
Review Comment:
```suggestion
if (!firstCheckpointCompleted && checkpointId >=
firstCheckpointId && firstCheckpointId != Long.MIN_VALUE) {
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
