pnowojski commented on a change in pull request #16589:
URL: https://github.com/apache/flink/pull/16589#discussion_r680152919
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/tasks/StreamTask.java
##########
@@ -472,40 +486,60 @@ protected void
processInput(MailboxDefaultAction.Controller controller) throws E
new
ResumeWrapper(controller.suspendDefaultAction(timer), timer)));
}
- private void resetSynchronousSavepointId(long id, boolean succeeded) {
- if (!succeeded && activeSyncSavepointId != null &&
activeSyncSavepointId == id) {
- // allow to process further EndOfPartition events
- activeSyncSavepointId = null;
- operatorChain.setIgnoreEndOfInput(false);
+ protected void endData() throws Exception {
+ advanceToEndOfEventTime();
+ // finish all operators in a chain effect way
+ operatorChain.finishOperators(actionExecutor);
+
+ for (ResultPartitionWriter partitionWriter :
getEnvironment().getAllWriters()) {
+ partitionWriter.notifyEndOfData();
}
- syncSavepointId = null;
+
+ this.endOfDataReceived = true;
}
Review comment:
In the 2nd part of my comment I meant `endData()`, not only `endInput()`.
> If I understand the first part correctly you're referring solely to the
SourceOperatorStreamTask, am I right? All other tasks call endData from a
single path.
Mostly. `SourceStreamTask` can call `endData()` (via
`LegacySourceFunctionThread#completeProcessing`) also twice in case of
exceptions/interruptions, so at least not on the happy path.
> BTW, it's true we called endInput only in afterInvoke, but we were setting
a flag if we should actually execute it in the same number of places (more or
less).
Maybe, but setting a flag from a couple of places is a lesser problem vs
calling a major piece of lifecycle code from a couple of places.
Can we maybe at least unify `SourceStreamOperatorTask` and network input
tasks? I can see `SourceStreamTask` being different here after adding
`EndOfData` event, but FLIP-27 sources should have the same lifecycle and
runtime handling as network inputs.
--
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]
