StephanEwen commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493400902
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/StreamSource.java
##########
@@ -107,7 +106,11 @@ public void run(final Object lockingObject,
// in theory, the subclasses of StreamSource
may implement the BoundedOneInput interface,
// so we still need the following call to end
the input
synchronized (lockingObject) {
- operatorChain.endMainOperatorInput(1);
+ if (this instanceof BoundedOneInput) {
+ ((BoundedOneInput)
this).endInput();
+ } else if (this instanceof
BoundedMultiInput) {
+ ((BoundedMultiInput)
this).endInput(1);
Review comment:
Ah, maybe I confused this (and I also find that part of the code not
very easy).
Is this the input of the source operator (there should be none)? I
understood this was the input of the successor to the source operator, meaning
the source tells its downstream successor that the input that it pushes to is
now done.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
