You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/09/18 12:16:57 UTC
[GitHub] [flink] dawidwys opened a new pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
dawidwys opened a new pull request #13423:
URL: https://github.com/apache/flink/pull/13423
## What is the purpose of the change
In this PR PushingAsyncDataInput.DataOutput is notified
about the end of input. It also moves the propagation to the bounded
operators of that signal from OperatorChain to the DataOutputs.
## Verifying this change
This change is already covered by existing tests.
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / **no**)
- The serializers: (yes / **no** / don't know)
- The runtime per-record code paths (performance sensitive): (**yes** / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / **no**)
- If yes, how is the feature documented? (**not applicable** / docs / JavaDocs / not documented)
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] kl0u commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
kl0u commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492589663
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
Here can it be any other inputId apart from 1?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
The reason I am asking is because this check did not exist in any of the methods removed by this PR which were handling the input closing logic.
In my opinion, although from the code this check may be safe, to be on the safe side it may make sense to remove it so that the logic is the same. But feel free to disagree and leave it as is.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
The reason I am asking is because this check did not exist in any of the methods removed by this PR which were handling the input closing logic. This makes me wonder if we are correct on this.
In my opinion, although from the code this check may be safe, to be on the safe side it may make sense to remove it so that the logic is the same. But feel free to disagree and leave it as is.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-698209635
Ad. 2 Very good point. I totally missed that. Will add it.
Ad. 1 Will add tests.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-698209635
Ad. 2 Very good point. I totally missed that. Will add it.
Ad. 1 Will add tests.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-695955429
@flinkbot run azure
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493573809
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
Is it a bug?
Do we enforce anywhere that a multi input operator cannot implement the `BoundedOneInput`? What if user implements:
```
class CustomOperator extends AbstractStreamOperatorV2 implements BoundedOneInput, BoundedMultiInput
```
?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
}, {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686",
"triggerID" : "695955429",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 673a28c38792b6e9fd4532d9bf0d130fc65412e5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646) Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r494110420
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
I think this could lead to more confusion and bugs. It would be better to keep it simple and not support it. `endInput()` in your example above someone could understand as ending all inputs.
Also as always. Keeping it simpler now, means we can always extend it in the future if we change our mind. But if we decide to support some strange combination now "just because we can", backing out from it won't be possible without braking compatibility.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492666629
##########
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:
Not sure if I understand the comment.
Does a StreamSource even have inputs? Of course I might be missing something here, but isn't the input id, the id of an input that has finished? Honestly, I find this part of code quite shady, but I think the logic pre and post this change is the same. Again I might be wrong.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492666629
##########
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:
Sorry, I am not sure if I understand the comment.
Does a StreamSource even have inputs? Of course I might be missing something here, but isn't the input id, the id of an input that has finished? Honestly, I find this part of code quite shady (as there are actually no inputs that can finish, imo), but I think the logic pre and post this change is the same. Again I might be wrong.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
}, {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686",
"triggerID" : "695955429",
"triggerType" : "MANUAL"
}, {
"hash" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853",
"triggerID" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"triggerType" : "PUSH"
}, {
"hash" : "ef80141dd2339d94d2497c22b298474d5aee680f",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6956",
"triggerID" : "ef80141dd2339d94d2497c22b298474d5aee680f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0132280bc7d2a0b1147d75e9e63aa280880e5ff9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853)
* ef80141dd2339d94d2497c22b298474d5aee680f Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6956)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r494131060
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
You are right, it's currently an undefined behaviour. Imo all of the above should be disallowed. And at the very least, they were implicitly disallowed while we were implementing bounded inputs, vide no test coverage for those combinations.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-696628108
Hi @pnowojski could you take a look at the changes, as it touches some of the streaming runtime parts? Or maybe you could ask somebody else that works on the streaming runtime part?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493587911
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java
##########
@@ -354,5 +341,12 @@ public void emitStreamStatus(StreamStatus streamStatus) {
public void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception {
input.processLatencyMarker(latencyMarker);
}
+
+ @Override
+ public void endOutput() throws Exception {
Review comment:
I will extract it as helper methods.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492658937
##########
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:
This looks tricky. The source might run at other positions than input 1.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r494131060
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
You are right, it's currently an undefined behaviour. Imo all of the above should be disallowed.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r494131060
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
You are right, it's currently an undefined behaviour. Imo all of the above should be disallowed. And at the very least, they were implicitly disallowed by the runtime, vide no test coverage for those combinations.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694834575
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit 673a28c38792b6e9fd4532d9bf0d130fc65412e5 (Fri Sep 18 12:18:49 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-696628108
Hi @pnowojski could you take a look at the changes, as it touches some of the streaming runtime parts? Or maybe you could ask somebody else that works on the streaming runtime part?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys closed pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys closed pull request #13423:
URL: https://github.com/apache/flink/pull/13423
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
}, {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686",
"triggerID" : "695955429",
"triggerType" : "MANUAL"
}, {
"hash" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853",
"triggerID" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0132280bc7d2a0b1147d75e9e63aa280880e5ff9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 673a28c38792b6e9fd4532d9bf0d130fc65412e5 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492618028
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
It shouldn't. Nevertheless I think it is not safe to just assume so.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r494127458
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
If you say so.
BTW, I think this discussion is somewhat unrelated to the PR, but ...
I find the contract between `BoundedOneInput` and `BoundMultiInput` undefined/ill defined.
Are these combinations valid? In the current implementation
1. `TwoInputOperator implements BoundedOneInput` -> whenver any input finishes, we call BoundedOneInput#endInput, imo this is confusing
2. `OneInputOperator implements BoundedMultiInput` -> we call BoundedMultiInput#endInput(1) <- sort of ok, imo
3. `MultiInputOperator implements BoundedOneInput` -> see 1.
4. `MultiInputOperator implements BoundedMultiInput, BoundedOneInput` -> we only ever call BoundedOneInput#endInput for all finished inputs, which imo is misleading
I know that this PR, does not fixes it. In the PR I do ignore options 1 and 2 though.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] kl0u commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
kl0u commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492589663
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
Here can it be any other inputId apart from 1?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-706022049
In the end we decided to put the sorting logic into the DataInput instead of DataOutput. Therefore this Pr is not valid anymore.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r494110420
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
I think this could lead to more confusion and bugs. It would be better to keep it simple and not support it. `endInput()` in your example above someone could understand as ending all inputs.
Also as always. Keeping it simpler now, means we can always extend it in the future if we change our mind. But if we decide to support some strange combination now "just because we can", backing out from it won't be possible without braking compatibility.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
You are right, it's currently an undefined behaviour. Imo all of the above should be disallowed.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
You are right, it's currently an undefined behaviour. Imo all of the above should be disallowed. And at the very least, they were implicitly disallowed by the runtime, vide no test coverage for those combinations.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
You are right, it's currently an undefined behaviour. Imo all of the above should be disallowed. And at the very least, they were implicitly disallowed while we were implementing bounded inputs, vide no test coverage for those combinations.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-706022049
In the end we decided to put the sorting logic into the DataInput instead of DataOutput. Therefore this Pr is not valid anymore.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] pnowojski commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
pnowojski commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493272774
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -33,7 +33,7 @@
* {@link AbstractStreamOperatorV2}.
Review comment:
nit: maybe add a motivation behind this change into the commit message? Like copy from the jira ticket:
```
As suggested in the FLIP-140 we want to implement the sorting inside
of PushingAsyncDataInput.DataOutput.
To do that PushingAsyncDataInput.DataOutput is notified
about the end of input. It also moves the propagation to the bounded
operators of that signal from OperatorChain to the DataOutputs.
```
?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/PushingAsyncDataInput.java
##########
@@ -57,5 +57,7 @@
void emitStreamStatus(StreamStatus streamStatus) throws Exception;
void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception;
+
+ default void endOutput() throws Exception {}
Review comment:
Why is this exposed to the things implementing `PushingAsyncDataInput`? I guess to make implementation easier?
I mean, there should be no reason for `StreamTaskNetworkInput` or `StreamTaskSourceInput` to call `output.endOutput()`, as the runtime can do it for them and it can only lead to bugs if the method is called prematurely.
Maybe there is an easy way to hide it from the `emitNext()`? Like splitting into two separate interfaces?
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java
##########
@@ -354,5 +341,12 @@ public void emitStreamStatus(StreamStatus streamStatus) {
public void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception {
input.processLatencyMarker(latencyMarker);
}
+
+ @Override
+ public void endOutput() throws Exception {
Review comment:
I don't like you have duplicated the `instanceof` checks in many places :(
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java
##########
@@ -268,7 +252,12 @@ public InputProcessor(
}
public InputStatus processInput() throws Exception {
- return taskInput.emitNext(dataOutput);
+ InputStatus status = taskInput.emitNext(dataOutput);
+ if (status == InputStatus.END_OF_INPUT) {
+ dataOutput.endOutput();
Review comment:
It looks strange: you are asking `taskInput` to emit records to `dataOutput`. It tells you `END_OF_INPUT` and you are calling `endOutput()` on the `dataOutput` that you have just passed to the `taskInput` .
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
`inputId != 1` and `BoundedOneInput` is definitely a bug, so the code should fail. Add a `checkState`/`checkArgument`?
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492618028
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
It shouldn't. Nevertheless I think it is not safe to just assume so.
##########
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:
Not sure if I understand the comment.
Does a StreamSource even have inputs? Of course I might be missing something here, but isn't the input id, the id of an input that has finished? Honestly, I find this part of code quite shady, but I think the logic pre and post this change is the same. Again I might be wrong.
##########
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:
Sorry, I am not sure if I understand the comment.
Does a StreamSource even have inputs? Of course I might be missing something here, but isn't the input id, the id of an input that has finished? Honestly, I find this part of code quite shady, but I think the logic pre and post this change is the same. Again I might be wrong.
##########
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:
Sorry, I am not sure if I understand the comment.
Does a StreamSource even have inputs? Of course I might be missing something here, but isn't the input id, the id of an input that has finished? Honestly, I find this part of code quite shady (as there are actually no inputs that can finish, imo), but I think the logic pre and post this change is the same. Again I might be wrong.
##########
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:
Sorry, I am not sure if I understand the comment.
Does a StreamSource even have inputs? Of course I might be missing something here, but isn't the input id, the id of an input that has finished? Honestly, I find this part of orginal code quite shady (as there are actually no inputs that can finish, imo), but I think the logic pre and post this change is the same. Again I might be wrong.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
}, {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686",
"triggerID" : "695955429",
"triggerType" : "MANUAL"
}, {
"hash" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853",
"triggerID" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"triggerType" : "PUSH"
}, {
"hash" : "ef80141dd2339d94d2497c22b298474d5aee680f",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "ef80141dd2339d94d2497c22b298474d5aee680f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0132280bc7d2a0b1147d75e9e63aa280880e5ff9 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853)
* ef80141dd2339d94d2497c22b298474d5aee680f UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] kl0u commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
kl0u commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492682632
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
The reason I am asking is because this check did not exist in any of the methods removed by this PR which were handling the input closing logic. This makes me wonder if we are correct on this.
In my opinion, although from the code this check may be safe, to be on the safe side it may make sense to remove it so that the logic is the same. But feel free to disagree and leave it as is.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r494127458
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
If you say so.
BTW, I think this discussion is somewhat unrelated to the PR, but ...
I find the contract between `BoundedOneInput` and `BoundMultiInput` undefined/ill defined.
Are these combinations valid? In the current implementation
1. `TwoInputOperator implements BoundedOneInput` -> whenver any input finishes, we call BoundedOneInput#endInput, imo this is confusing
2. `OneInputOperator implements BoundedMultiInput` -> we call BoundedMultiInput#endInput(1) <- sort of ok, imo
3. `MultiInputOperator implements BoundedOneInput` -> see 1.
4. `MultiInputOperator implements BoundedMultiInput, BoundedOneInput` -> we only ever call BoundedOneInput#endInput for all finished inputs, which imo is misleading
I know that this PR, does not fixe it. In the PR I do ignore options 1 and 2 though.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
If you say so.
BTW, I think this discussion is somewhat unrelated to the PR, but ...
I find the contract between `BoundedOneInput` and `BoundMultiInput` undefined/ill defined.
Are these combinations valid? In the current implementation
1. `TwoInputOperator implements BoundedOneInput` -> whenver any input finishes, we call BoundedOneInput#endInput, imo this is confusing
2. `OneInputOperator implements BoundedMultiInput` -> we call BoundedMultiInput#endInput(1) <- sort of ok, imo
3. `MultiInputOperator implements BoundedOneInput` -> see 1.
4. `MultiInputOperator implements BoundedMultiInput, BoundedOneInput` -> we only ever call BoundedOneInput#endInput for all finished inputs, which imo is misleading
I know that this PR, does not fix it. In the PR I do ignore options 1 and 2 though.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493586642
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/PushingAsyncDataInput.java
##########
@@ -57,5 +57,7 @@
void emitStreamStatus(StreamStatus streamStatus) throws Exception;
void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception;
+
+ default void endOutput() throws Exception {}
Review comment:
Make sense, will hide it from there.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys closed pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys closed pull request #13423:
URL: https://github.com/apache/flink/pull/13423
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
}, {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686",
"triggerID" : "695955429",
"triggerType" : "MANUAL"
}, {
"hash" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 673a28c38792b6e9fd4532d9bf0d130fc65412e5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646) Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686)
* 0132280bc7d2a0b1147d75e9e63aa280880e5ff9 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
}, {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686",
"triggerID" : "695955429",
"triggerType" : "MANUAL"
} ]
}-->
## CI report:
* 673a28c38792b6e9fd4532d9bf0d130fc65412e5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646) Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 673a28c38792b6e9fd4532d9bf0d130fc65412e5 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
}, {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686",
"triggerID" : "695955429",
"triggerType" : "MANUAL"
}, {
"hash" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853",
"triggerID" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"triggerType" : "PUSH"
}, {
"hash" : "ef80141dd2339d94d2497c22b298474d5aee680f",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6956",
"triggerID" : "ef80141dd2339d94d2497c22b298474d5aee680f",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* ef80141dd2339d94d2497c22b298474d5aee680f Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6956)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-695955429
@flinkbot run azure
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r494127458
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
If you say so.
BTW, I think this discussion is somewhat unrelated to the PR, but ...
I find the contract between `BoundedOneInput` and `BoundMultiInput` undefined/ill defined.
Are these combinations valid? In the current implementation
1. `TwoInputOperator implements BoundedOneInput` -> whenver any input finishes, we call BoundedOneInput#endInput, imo this is confusing
2. `OneInputOperator implements BoundedMultiInput` -> we call BoundedMultiInput#endInput(1) <- sort of ok, imo
3. `MultiInputOperator implements BoundedOneInput` -> see 1.
4. `MultiInputOperator implements BoundedMultiInput, BoundedOneInput` -> we only ever call BoundedOneInput#endInput for all finished inputs, which imo is misleading
I know that this PR, does not fixes it. In the PR I do ignore options 1 and 2 though.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
If you say so.
BTW, I think this discussion is somewhat unrelated to the PR, but ...
I find the contract between `BoundedOneInput` and `BoundMultiInput` undefined/ill defined.
Are these combinations valid? In the current implementation
1. `TwoInputOperator implements BoundedOneInput` -> whenver any input finishes, we call BoundedOneInput#endInput, imo this is confusing
2. `OneInputOperator implements BoundedMultiInput` -> we call BoundedMultiInput#endInput(1) <- sort of ok, imo
3. `MultiInputOperator implements BoundedOneInput` -> see 1.
4. `MultiInputOperator implements BoundedMultiInput, BoundedOneInput` -> we only ever call BoundedOneInput#endInput for all finished inputs, which imo is misleading
I know that this PR, does not fixe it. In the PR I do ignore options 1 and 2 though.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
If you say so.
BTW, I think this discussion is somewhat unrelated to the PR, but ...
I find the contract between `BoundedOneInput` and `BoundMultiInput` undefined/ill defined.
Are these combinations valid? In the current implementation
1. `TwoInputOperator implements BoundedOneInput` -> whenver any input finishes, we call BoundedOneInput#endInput, imo this is confusing
2. `OneInputOperator implements BoundedMultiInput` -> we call BoundedMultiInput#endInput(1) <- sort of ok, imo
3. `MultiInputOperator implements BoundedOneInput` -> see 1.
4. `MultiInputOperator implements BoundedMultiInput, BoundedOneInput` -> we only ever call BoundedOneInput#endInput for all finished inputs, which imo is misleading
I know that this PR, does not fix it. In the PR I do ignore options 1 and 2 though.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492666629
##########
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:
Sorry, I am not sure if I understand the comment.
Does a StreamSource even have inputs? Of course I might be missing something here, but isn't the input id, the id of an input that has finished? Honestly, I find this part of code quite shady, but I think the logic pre and post this change is the same. Again I might be wrong.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492666629
##########
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:
Sorry, I am not sure if I understand the comment.
Does a StreamSource even have inputs? Of course I might be missing something here, but isn't the input id, the id of an input that has finished? Honestly, I find this part of orginal code quite shady (as there are actually no inputs that can finish, imo), but I think the logic pre and post this change is the same. Again I might be wrong.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493587911
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java
##########
@@ -354,5 +341,12 @@ public void emitStreamStatus(StreamStatus streamStatus) {
public void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception {
input.processLatencyMarker(latencyMarker);
}
+
+ @Override
+ public void endOutput() throws Exception {
Review comment:
I will extract it helper methods.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493592305
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
BTW, shouldn't the `BoundedMultiInput` extend from `BoundedOneInput`?
```
interface BoundedMultiInput extends BoundedOneInput {
default void endInput() throws Exception {
endInput(1)
}
/**
* It is notified that no more data will arrive on the input identified by the {@code inputId}.
* The {@code inputId} is numbered starting from 1, and `1` indicates the first input.
*/
void endInput(int inputId) throws Exception;
}
```
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493587579
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/StreamMultipleInputProcessor.java
##########
@@ -268,7 +252,12 @@ public InputProcessor(
}
public InputStatus processInput() throws Exception {
- return taskInput.emitNext(dataOutput);
+ InputStatus status = taskInput.emitNext(dataOutput);
+ if (status == InputStatus.END_OF_INPUT) {
+ dataOutput.endOutput();
Review comment:
After taking the previous comment into consideration, I think this will make sense. The `taskInput` will not be aware of the `endOutput` method any more, the `Processor` will though.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 673a28c38792b6e9fd4532d9bf0d130fc65412e5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] dawidwys commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
dawidwys commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r493586642
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/PushingAsyncDataInput.java
##########
@@ -57,5 +57,7 @@
void emitStreamStatus(StreamStatus streamStatus) throws Exception;
void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception;
+
+ default void endOutput() throws Exception {}
Review comment:
Make sense, I will hide it from there.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] kl0u commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
kl0u commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492682632
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractInput.java
##########
@@ -68,4 +68,13 @@ public void processLatencyMarker(LatencyMarker latencyMarker) throws Exception {
public void setKeyContextElement(StreamRecord record) throws Exception {
owner.internalSetKeyContextElement(record, stateKeySelector);
}
+
+ @Override
+ public void endInput() throws Exception {
+ if (owner instanceof BoundedOneInput && inputId == 1) {
Review comment:
The reason I am asking is because this check did not exist in any of the methods removed by this PR which were handling the input closing logic.
In my opinion, although from the code this check may be safe, to be on the safe side it may make sense to remove it so that the logic is the same. But feel free to disagree and leave it as is.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on a change in pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on a change in pull request #13423:
URL: https://github.com/apache/flink/pull/13423#discussion_r492658937
##########
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:
This looks tricky. The source might run at other positions than input 1.
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646",
"triggerID" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"triggerType" : "PUSH"
}, {
"hash" : "673a28c38792b6e9fd4532d9bf0d130fc65412e5",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686",
"triggerID" : "695955429",
"triggerType" : "MANUAL"
}, {
"hash" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853",
"triggerID" : "0132280bc7d2a0b1147d75e9e63aa280880e5ff9",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 673a28c38792b6e9fd4532d9bf0d130fc65412e5 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6646) Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6686)
* 0132280bc7d2a0b1147d75e9e63aa280880e5ff9 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6853)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
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:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13423: [FLINK-19269] Make the PushingAsyncDataInput.DataOutput aware of endOfInput
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13423:
URL: https://github.com/apache/flink/pull/13423#issuecomment-694841940
----------------------------------------------------------------
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:
users@infra.apache.org