dawidwys commented on a change in pull request #16589:
URL: https://github.com/apache/flink/pull/16589#discussion_r678463750
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java
##########
@@ -151,7 +152,11 @@ protected DataInputStatus processEvent(BufferOrEvent
bufferOrEvent) {
final AbstractEvent event = bufferOrEvent.getEvent();
// TODO: with checkpointedInputGate.isFinished() we might not need to
support any events on
// this level.
- if (event.getClass() == EndOfPartitionEvent.class) {
+ if (event.getClass() == EndOfData.class) {
+ if (checkpointedInputGate.hasReceivedEndOfData()) {
+ return DataInputStatus.END_OF_DATA;
+ }
Review comment:
`EndOfPartitionEvent` is also handled in the
`AbstractStreamTaskNetworkInput`, but for `EndOfPartitionEvent`,
`CheckpointedInputGate` returns an empty buffer and then we have pretty similar
logic in `AbstractStreamTaskNetworkInput#emitNext:121`
I find simply returning the event rather than an empty buffer a cleaner
solution tbh.
--
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]
