AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1763232011
##########
flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/stream/AbstractStreamingWriter.java:
##########
@@ -123,6 +142,27 @@ public void bucketInactive(Bucket<IN, String> bucket) {
bucketCheckInterval);
currentWatermark = Long.MIN_VALUE;
+
+ // Figure out if we have seen end of input before and if we should
anything downstream. 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
unnecessary .
+ // 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);
+ List<Boolean> previousState =
Lists.newArrayList(endOfInputState.get());
Review Comment:
Sure if you drop Java8 support first ;).
--
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]
