AHeise commented on code in PR #25292:
URL: https://github.com/apache/flink/pull/25292#discussion_r1758859370
##########
flink-runtime/src/main/java/org/apache/flink/streaming/runtime/operators/sink/SinkWriterOperator.java:
##########
@@ -160,13 +171,38 @@ public void initializeState(StateInitializationContext
context) throws Exception
legacyCommitterState.clear();
}
}
- sinkWriter = writerStateHandler.createWriter(initContext, context);
+
+ endOfInputState =
context.getOperatorStateStore().getListState(END_OF_INPUT_STATE_DESC);
+ // We can have the following cases:
+ // 1. state is empty:
+ // - First time initialization
+ // - Restoring from a previous version of Flink
+ // - Upscaled from a final or regular checkpoint
+ // In all cases, we properly initialize the data and 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 EOI, we
regularly start the writer
+ // and potentially emit duplicate summaries if we indeed recovered
from a final
+ // checkpoint.
+ ArrayList<Boolean> previousState =
Lists.newArrayList(endOfInputState.get());
+ this.endOfInput = !previousState.isEmpty() &&
!previousState.contains(false);
+ sinkWriter =
+ this.endOfInput
+ ? new ClosedWriter<>()
Review Comment:
Fixed in a fixup commit. PTAL.
--
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]
