AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1756485743
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -195,12 +216,18 @@ public void processWatermark(Watermark mark) throws
Exception {
@Override
public void endInput() throws Exception {
- endOfInput = true;
- sinkWriter.flush(true);
- emitCommittables(Long.MAX_VALUE);
+ if (!this.endOfInput) {
+ this.endOfInput = true;
+ // endOfInputState is union state, so it's enough if one task adds
something to it
Review Comment:
You are correct that this was not correct, especially also on partially
completed tasks.
I changed it to regular state and let the committer handle duplicates now.
The main idea is to avoid extra transactions as much as possible in the writer
and if it's not possible, let the committer clean them up anyhow.
--
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]
