[
https://issues.apache.org/jira/browse/FLINK-4391?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15635787#comment-15635787
]
ASF GitHub Bot commented on FLINK-4391:
---------------------------------------
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2629
The combination of `UNORDERED` with event time is not strictly meaningless.
The only thing which one has to do respect is that only those elements in the
queue from the beginning to the first watermark element are allowed to be
processed in an `UNORDERED` fashion. Only they have been emitted, the watermark
can be emitted. After the watermark has been emitted, the next batch of stream
records can be processed `UNORDERED`. An example:
Given `queue = s1, s2, s3, w1, s4, w2, s5, s6, w3` we can process `s1, s2`
`s3` `UNORDERED`. After they have been emitted, `w1` has to be emitted. And
only after the emission of `w1`, we can start emitting `s4`. After `w2` we can
process `s5` and `s6` `UNORDERED`. And so on...
Concerning the exactly once processing guarantees. The problem is the
following with the current implementation. When you call
`AsyncWaitOperator.snapshotState` it will call
`AsyncCollectorBuffer.getStreamElementsInBuffer` which will block the `Emitter`
thread. However, after `getStreamElementsInBuffer` has completed, the `Emitter`
thread is again allowed to emit new elements, right? Thus, this can happen
before the `StreamTask` has actually send the checkpoint barrier to downstream
operators because it is not guarded by the checkpoint lock. So you might have
an element `x1` contained in the checkpoint of the `AsyncWaitOperator` and in
one of the downstream operators. That's the reason why you only have at-least
once processing guarantees.
I fear that the hand-tailored solution for the `AsyncWaitOperator` in the
`StreamTask.performCheckpoint` method won't solve the problems with the exactly
once processing guarantees because of the afore-mentioned problems. I don't see
another way to solve the problem at the moment other than using the checkpoint
lock. Have you measured the performance penalty for multiple chained
`AsyncWaitOperators` when using the checkpoint lock?
> Provide support for asynchronous operations over streams
> --------------------------------------------------------
>
> Key: FLINK-4391
> URL: https://issues.apache.org/jira/browse/FLINK-4391
> Project: Flink
> Issue Type: New Feature
> Components: DataStream API
> Reporter: Jamie Grier
> Assignee: david.wang
>
> Many Flink users need to do asynchronous processing driven by data from a
> DataStream. The classic example would be joining against an external
> database in order to enrich a stream with extra information.
> It would be nice to add general support for this type of operation in the
> Flink API. Ideally this could simply take the form of a new operator that
> manages async operations, keeps so many of them in flight, and then emits
> results to downstream operators as the async operations complete.
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
