AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1763233550
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java:
##########
@@ -149,15 +190,20 @@ public void processElement(StreamRecord<IN> element)
throws Exception {
@Override
public void notifyCheckpointComplete(long checkpointId) throws Exception {
super.notifyCheckpointComplete(checkpointId);
- commitUpToCheckpoint(checkpointId);
+ if (!this.endOfInput) {
+ commitUpToCheckpoint(checkpointId);
+ }
}
@Override
public void endInput() throws Exception {
- buckets.onProcessingTime(Long.MAX_VALUE);
- helper.snapshotState(Long.MAX_VALUE);
- output.emitWatermark(new Watermark(Long.MAX_VALUE));
- commitUpToCheckpoint(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ buckets.onProcessingTime(Long.MAX_VALUE);
+ helper.snapshotState(Long.MAX_VALUE);
+ output.emitWatermark(new Watermark(Long.MAX_VALUE));
+ commitUpToCheckpoint(Long.MAX_VALUE);
Review Comment:
I thought about that but decided against it. This is not related to
CommittableSummary directly, so it would feel weird to import it just for the
EOI. And someone wise said that this class is deprecated anyways, so I didn't
want to change too much.
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
