You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by Stephan Ewen <se...@apache.org> on 2020/05/09 16:50:44 UTC
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Thanks, Aljoscha, for picking this up.
I agree with the approach of doing the here proposed set of changes for
now. It already makes things simpler and adds idleness support everywhere.
Rich functions and state always add complexity, let's do this in a next
step, if we have a really compelling case.
On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <al...@apache.org>
wrote:
> Regarding the WatermarkGenerator (WG) interface itself. The proposal is
> basically to turn emitting into a "flatMap", we give the
> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG can
> decide whether to output a watermark or not and can also mark the output
> as idle. Changing the interface to return a Watermark (as the previous
> watermark assigner interface did) would not allow that flexibility.
>
> Regarding checkpointing the watermark and keeping track of the minimum
> watermark, this would be the responsibility of the framework (or the
> KafkaConsumer in the current implementation). The user-supplied WG does
> not need to make sure the watermark doesn't regress.
>
> Regarding making the WG a "rich function", I can see the potential
> benefit but I also see a lot of pitfalls. For example, how should the
> watermark state be handled in the case of scale-in? It could be made to
> work in the Kafka case by attaching the state to the partition state
> that we keep, but then we have potential backwards compatibility
> problems also for the WM state. Does the WG usually need to keep the
> state or might it be enough if the state is transient, i.e. if you have
> a restart the WG would loose its histogram but it would rebuild it
> quickly and you would get back to the same steady state as before.
>
> Best,
> Aljoscha
>
> On 27.04.20 12:12, David Anderson wrote:
> > Overall I like this proposal; thanks for bringing it forward, Aljoscha.
> >
> > I also like the idea of making the Watermark generator a rich function --
> > this should make it more straightforward to implement smarter watermark
> > generators. Eg, one that uses state to keep statistics about the actual
> > out-of-orderness, and uses those statistics to implement a variable
> delay.
> >
> > David
> >
> > On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <kk...@gmail.com>
> wrote:
> >
> >> Hi Aljoscha,
> >>
> >> Thanks for opening the discussion!
> >>
> >> I have two comments on the FLIP:
> >> 1) we could add lifecycle methods to the Generator, i.e. open()/
> >> close(), probably with a Context as argument: I have not fully thought
> >> this through but I think that this is more aligned with the rest of
> >> our rich functions. In addition, it will allow, for example, to
> >> initialize the Watermark value, if we decide to checkpoint the
> >> watermark (see [1]) (I also do not know if Table/SQL needs to do
> >> anything in the open()).
> >> 2) aligned with the above, and with the case where we want to
> >> checkpoint the watermark in mind, I am wondering about how we could
> >> implement this in the future. In the FLIP, it is proposed to expose
> >> the WatermarkOutput in the methods of the WatermarkGenerator. Given
> >> that there is the implicit contract that watermarks are
> >> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
> >> assume) a check that will compare the last emitted WM against the
> >> provided one, and emit it only if it is >=. If not, then we risk
> >> having the user shooting himself on the foot if he/she accidentally
> >> forgets the check. Given that the WatermarkGenerator and its caller do
> >> not know if the watermark was finally emitted or not (the
> >> WatermarkOutput#emitWatermark returns void), who will be responsible
> >> for checkpointing the WM?
> >>
> >> Given this, why not having the methods as:
> >>
> >> public interface WatermarkGenerator<T> {
> >>
> >> Watermark onEvent(T event, long eventTimestamp, WatermarkOutput
> >> output);
> >>
> >> Watermark onPeriodicEmit(WatermarkOutput output);
> >> }
> >>
> >> and the caller will be the one enforcing any invariants, such as
> >> non-decreasing watermarks. In this way, the caller can checkpoint
> >> anything that is needed as it will have complete knowledge as to if
> >> the WM was emitted or not.
> >>
> >> What do you think?
> >>
> >> Cheers,
> >> Kostas
> >>
> >> [1] https://issues.apache.org/jira/browse/FLINK-5601
> >>
> >> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <tw...@apache.org>
> wrote:
> >>>
> >>> Thanks for the proposal Aljoscha. This is a very useful unification. We
> >>> have considered this FLIP already in the interfaces for FLIP-95 [1] and
> >>> look forward to update to the new unified watermark generators once
> >>> FLIP-126 has been accepted.
> >>>
> >>> Regards,
> >>> Timo
> >>>
> >>> [1] https://github.com/apache/flink/pull/11692
> >>>
> >>> On 20.04.20 18:10, Aljoscha Krettek wrote:
> >>>> Hi Everyone!
> >>>>
> >>>> We would like to start a discussion on "FLIP-126: Unify (and separate)
> >>>> Watermark Assigners" [1]. This work was started by Stephan in an
> >>>> experimental branch. I expanded on that work to provide a PoC for the
> >>>> changes proposed in this FLIP: [2].
> >>>>
> >>>> Currently, we have two different flavours of Watermark
> >>>> Assigners: AssignerWithPunctuatedWatermarks
> >>>> and AssignerWithPeriodicWatermarks. Both of them extend
> >>>> from TimestampAssigner. This means that sources that want to support
> >>>> watermark assignment/extraction in the source need to support two
> >>>> separate interfaces, we have two operator implementations for the
> >>>> different flavours. Also, this makes features such as generic support
> >>>> for idleness detection more complicated to implemented because we
> again
> >>>> have to support two types of watermark assigners.
> >>>>
> >>>> In this FLIP we propose two things:
> >>>>
> >>>> Unify the Watermark Assigners into one Interface WatermarkGenerator
> >>>> Separate this new interface from the TimestampAssigner
> >>>> The motivation for the first is to simplify future implementations and
> >>>> code duplication. The motivation for the second point is again code
> >>>> deduplication, most assigners currently have to extend from some base
> >>>> timestamp extractor or duplicate the extraction logic, or users have
> to
> >>>> override an abstract method of the watermark assigner to provide the
> >>>> timestamp extraction logic.
> >>>>
> >>>> Additionally, we propose to add a generic wrapping WatermarkGenerator
> >>>> that provides idleness detection, i.e. it can mark a stream/partition
> >> as
> >>>> idle if no data arrives after a configured timeout.
> >>>>
> >>>> The "unify and separate" part refers to the fact that we want to unify
> >>>> punctuated and periodic assigners but at the same time split the
> >>>> timestamp assigner from the watermark generator.
> >>>>
> >>>> Please find more details in the FLIP [1]. Looking forward to
> >>>> your feedback.
> >>>>
> >>>> Best,
> >>>> Aljoscha
> >>>>
> >>>> [1]
> >>>>
> >>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
> >>>>
> >>>>
> >>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
> >>>
> >>
> >
>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Aljoscha Krettek <al...@apache.org>.
The specifics of naming diverged a bit from the FLIP during
implementation but that should be fine. What matters in the end is the
intention of the FLIP and that the code that is committed in the end is
good and consistent in itself.
Best,
Aljoscha
On 24.05.20 05:12, Guanghui Zhang wrote:
> Hi, @Aljoscha,
> the function param currentTimestamp comment does not match the
> recordTimestamp "long extractTimestamp(T element, long recordTimestamp)"
> on wiki.
>
> Best,
> Zhangguanghui
>
> Dawid Wysakowicz <dw...@apache.org> 于2020年5月13日周三 上午12:28写道：
>
>> Thank you for the update and sorry again for chiming in so late...
>>
>> Best,
>>
>> Dawid
>>
>>
>> On 12/05/2020 18:21, Aljoscha Krettek wrote:
>>> Yes, I am also ok with a SerializableTimestampAssigner. This only
>>> looks a bit clumsy in the API but as a user (that uses lambdas) you
>>> should not see this. I pushed changes for this to my branch:
>>> https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
>>>
>>> And yes, recordTimestamp sounds good for the TimestampAssigner. I
>>> admit I didn't read this well enough and only saw nativeTimestamp.
>>>
>>> Best,
>>> Aljoscha
>>>
>>> On 12.05.20 17:16, Dawid Wysakowicz wrote:
>>>> I have similar thoughts to @Stephan
>>>>
>>>> Ad. 1 I tried something like this on your branch:
>>>>
>>>> /**
>>>> * Adds the given {@link TimestampAssigner} to this {@link
>>>> WatermarkStrategies}. For top-level classes that implement both
>>>> Serializable and TimestampAssigner
>>>> */
>>>> public <TA extends TimestampAssigner<T> & Serializable>
>>>> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner) {
>>>> checkNotNull(timestampAssigner, "timestampAssigner");
>>>> this.timestampAssigner = timestampAssigner;
>>>> return this;
>>>> }
>>>>
>>>> @FunctionalInterface
>>>> public interface SerializableTimestampAssigner<T> extends
>>>> TimestampAssigner<T>, Serializable {
>>>> }
>>>>
>>>> /**
>>>> * Adds the given {@link TimestampAssigner} to this {@link
>>>> WatermarkStrategies}.
>>>> * Helper method for serializable lambdas.
>>>> */
>>>> public WatermarkStrategies<T>
>>>> withTimestampAssigner(SerializableTimestampAssigner<T>
>>>> timestampAssigner) {
>>>> checkNotNull(timestampAssigner, "timestampAssigner");
>>>> this.timestampAssigner = timestampAssigner;
>>>> return this;
>>>> }
>>>>
>>>> But I understand if that's too hacky. It's just a pity that we must
>>>> enforce limitations on an interface that are not strictly necessary.
>>>>
>>>> Ad 2/3
>>>>
>>>> I am aware the watermark assigner/timestamp extractor can be applied
>>>> further down the graph. Originally I also wanted to suggest
>>>> sourceTimestamp and SourceTimestampAssigner, but then I realized it can
>>>> be used also after the sources as you correctly pointed out. Even if the
>>>> TimestampAssigner is used after the source there might be some
>>>> native/record timestamp in the StreamRecord, that could've been
>>>> extracted by previous assigner.
>>>>
>>>> Best,
>>>>
>>>> Dawid
>>>>
>>>> On 12/05/2020 16:47, Stephan Ewen wrote:
>>>>> @Aljoscha
>>>>>
>>>>> About (1) could we have an interface SerializableTimestampAssigner that
>>>>> simply mixes in the java.io.Serializable interface? Or will this be too
>>>>> clumsy?
>>>>>
>>>>> About (3) RecordTimeStamp seems to fit both cases (in-source-record
>>>>> timestamp, in stream-record timestamp).
>>>>>
>>>>> On Tue, May 12, 2020 at 4:12 PM Aljoscha Krettek <al...@apache.org>
>>>>> wrote:
>>>>>
>>>>>> Definitely +1 to point 2) raised by Dawid. I'm not sure on points
>>>>>> 1) and
>>>>>> 3).
>>>>>>
>>>>>> 1) I can see the benefit of that but in reality most timestamp
>>>>>> assigners
>>>>>> will probably need to be Serializable. If you look at my (updated) POC
>>>>>> branch [1] you can see how a TimestampAssigner would be specified
>>>>>> on the
>>>>>> WatermarkStrategies helper class: [2]. The signature of this would
>>>>>> have
>>>>>> to be changed to something like:
>>>>>>
>>>>>> public <TA extends TimestampAssigner<T> & Serializable>
>>>>>> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner)
>>>>>>
>>>>>> Then, however, it would not be possible for users to specify a
>>>>>> lambda or
>>>>>> anonymous inner function for the TimestampAssigner like this:
>>>>>>
>>>>>> WatermarkStrategy<Long> testWmStrategy = WatermarkStrategies
>>>>>> .forGenerator(new PeriodicTestWatermarkGenerator())
>>>>>> .withTimestampAssigner((event, timestamp) -> event)
>>>>>> .build();
>>>>>>
>>>>>> 3) This makes sense if we only allow WatermarkStrategies on sources,
>>>>>> where the previous timestamp really is the "native" timestamp.
>>>>>> Currently, we also allow setting watermark strategies at arbitrary
>>>>>> points in the graph. I'm thinking we probably should only allow
>>>>>> that in
>>>>>> sources but it's not the reality currently. I'm not against
>>>>>> renaming it,
>>>>>> just voicing those thoughts.
>>>>>>
>>>>>> Best,
>>>>>> Aljoscha
>>>>>>
>>>>>>
>>>>>> [1]
>>>>>>
>> https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
>>>>>> [2]
>>>>>>
>>>>>>
>> https://github.com/aljoscha/flink/blob/flink-xxx-wm-generators-rebased/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java#L81
>>>>>>
>>>>>>
>>>>>> On 12.05.20 15:48, Stephan Ewen wrote:
>>>>>>> +1 to all of Dawid's suggestions, makes a lot of sense to me
>>>>>>>
>>>>>>> On Tue, May 12, 2020 at 2:32 PM Dawid Wysakowicz
>>>>>>> <dwysakowicz@apache.org
>>>>>>>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Hi Aljoscha,
>>>>>>>>
>>>>>>>> Sorry for adding comments during the vote, but I have some really
>>>>>>>> minor
>>>>>>>> suggestions that should not influence the voting thread imo.
>>>>>>>>
>>>>>>>> 1) Does it make sense to have the TimestampAssigner extend from
>>>>>>>> Flink's
>>>>>>>> Function? This implies it has to be serializable which with the
>>>>>>>> factory
>>>>>>>> pattern is not strictly necessary, right? BTW I really like that you
>>>>>>>> suggested the FunctionInterface annotation there.
>>>>>>>>
>>>>>>>> 2) Could we rename the IdentityTimestampAssigner to e.g.
>>>>>>>>
>>>>>>
>> RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
>>>>>>
>>>>>>>> Personally I found the IdentityTimestampAssigner a bit misleading
>>>>>>>> as it
>>>>>>>> usually mean a no-op. Which did not click for me, as I assumed it
>>>>>>>> somehow returns the incoming record itself.
>>>>>>>>
>>>>>>>> 3) Could we rename the second parameter of
>>>>>>>> TimestampAssigner#extract to
>>>>>>>> e.g. recordTimestamp/nativeTimestamp? This is similar to the point
>>>>>>>> above. This parameter was also a bit confusing for me as I
>>>>>>>> thought at
>>>>>>>> times its somehow related to
>>>>>>>> TimerService#currentProcessingTimestamp()/currentWatermark() as the
>>>>>>>> whole system currentTimestamp.
>>>>>>>>
>>>>>>>> Other than those three points I like the proposal and I was about to
>>>>>>>> vote +1 if it was not for those three points.
>>>>>>>>
>>>>>>>> Best,
>>>>>>>>
>>>>>>>> Dawid
>>>>>>>>
>>>>>>>> On 11/05/2020 16:57, Jark Wu wrote:
>>>>>>>>> Thanks for the explanation. I like the fatory pattern to make the
>>>>>> member
>>>>>>>>> variables immutable and final.
>>>>>>>>>
>>>>>>>>> So +1 to the proposal.
>>>>>>>>>
>>>>>>>>> Best,
>>>>>>>>> Jark
>>>>>>>>>
>>>>>>>>> On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org>
>>>>>>>>> wrote:
>>>>>>>>>
>>>>>>>>>> I am fine with that.
>>>>>>>>>>
>>>>>>>>>> Much of the principles seem agreed upon. I understand the need to
>>>>>>>> support
>>>>>>>>>> code-generated extractors and we should support most of it
>>>>>>>>>> already (as
>>>>>>>>>> Aljoscha mentioned via the factories) can extend this if needed.
>>>>>>>>>>
>>>>>>>>>> I think that the factory approach supports code-generated
>>>>>>>>>> extractors
>>>>>> in
>>>>>>>> a
>>>>>>>>>> cleaner way even than an extractor with an open/init method.
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek
>>>>>>>>>> <aljoscha@apache.org
>>>>>>>>>> wrote:
>>>>>>>>>>
>>>>>>>>>>> We're slightly running out of time. I would propose we vote on
>>>>>>>>>>> the
>>>>>>>> basic
>>>>>>>>>>> principle and remain open to later additions. This feature is
>>>>>>>>>>> quite
>>>>>>>>>>> important to make the new Kafka Source that is developed as
>>>>>>>>>>> part of
>>>>>>>>>>> FLIP-27 useful. Otherwise we would have to use the legacy
>>>>>>>>>>> interfaces
>>>>>> in
>>>>>>>>>>> the newly added connector.
>>>>>>>>>>>
>>>>>>>>>>> I know that's a bit unorthodox but would everyone be OK with
>>>>>>>>>>> what's
>>>>>>>>>>> currently there and then we iterate?
>>>>>>>>>>>
>>>>>>>>>>> Best,
>>>>>>>>>>> Aljoscha
>>>>>>>>>>>
>>>>>>>>>>> On 11.05.20 13:57, Aljoscha Krettek wrote:
>>>>>>>>>>>> Ah, I meant to write this in my previous email, sorry about
>>>>>>>>>>>> that.
>>>>>>>>>>>>
>>>>>>>>>>>> The WatermarkStrategy, which is basically a factory for a
>>>>>>>>>>>> WatermarkGenerator is the replacement for the open() method.
>>>>>>>>>>>> This is
>>>>>>>>>> the
>>>>>>>>>>>> same strategy that was followed for StreamOperatorFactory,
>>>>>>>>>>>> which was
>>>>>>>>>>>> introduced to allow code generation in the Table API [1]. If
>>>>>>>>>>>> we need
>>>>>>>>>>>> metrics or other things we would add that as a parameter to the
>>>>>>>> factory
>>>>>>>>>>>> method. What do you think?
>>>>>>>>>>>>
>>>>>>>>>>>> Best,
>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>
>>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
>>>>>>>>>>>>
>>>>>>>>>>>> On 10.05.20 05:07, Jark Wu wrote:
>>>>>>>>>>>>> Hi,
>>>>>>>>>>>>>
>>>>>>>>>>>>> Regarding to the `open()/close()`, I think it's necessary for
>>>>>>>>>>>>> Table&SQL to
>>>>>>>>>>>>> compile the generated code.
>>>>>>>>>>>>> In Table&SQL, the watermark strategy and event-timestamp is
>>>>>>>>>>>>> defined
>>>>>>>>>>> using
>>>>>>>>>>>>> SQL expressions, we will
>>>>>>>>>>>>> translate and generate Java code for the expressions. If we
>>>>>>>>>>>>> have
>>>>>>>>>>>>> `open()/close()`, we don't need lazy initialization.
>>>>>>>>>>>>> Besides that, I can see a need to report some metrics, e.g. the
>>>>>>>>>> current
>>>>>>>>>>>>> watermark, the dirty timestamps (null value), etc.
>>>>>>>>>>>>> So I think a simple `open()/close()` with a context which
>>>>>>>>>>>>> can get
>>>>>>>>>>>>> MetricGroup is nice and not complex for the first version.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Best,
>>>>>>>>>>>>> Jark
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org>
>>>>>> wrote:
>>>>>>>>>>>>>> Thanks, Aljoscha, for picking this up.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> I agree with the approach of doing the here proposed set of
>>>>>> changes
>>>>>>>>>> for
>>>>>>>>>>>>>> now. It already makes things simpler and adds idleness support
>>>>>>>>>>>>>> everywhere.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Rich functions and state always add complexity, let's do
>>>>>>>>>>>>>> this in a
>>>>>>>>>> next
>>>>>>>>>>>>>> step, if we have a really compelling case.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
>>>>>>>>>> aljoscha@apache.org>
>>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The
>>>>>>>> proposal
>>>>>>>>>>> is
>>>>>>>>>>>>>>> basically to turn emitting into a "flatMap", we give the
>>>>>>>>>>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and
>>>>>>>>>>>>>>> the WG
>>>>>>>>>> can
>>>>>>>>>>>>>>> decide whether to output a watermark or not and can also
>>>>>>>>>>>>>>> mark the
>>>>>>>>>>>>>>> output
>>>>>>>>>>>>>>> as idle. Changing the interface to return a Watermark (as the
>>>>>>>>>> previous
>>>>>>>>>>>>>>> watermark assigner interface did) would not allow that
>>>>>> flexibility.
>>>>>>>>>>>>>>> Regarding checkpointing the watermark and keeping track of
>>>>>>>>>>>>>>> the
>>>>>>>>>> minimum
>>>>>>>>>>>>>>> watermark, this would be the responsibility of the
>>>>>>>>>>>>>>> framework (or
>>>>>>>> the
>>>>>>>>>>>>>>> KafkaConsumer in the current implementation). The
>>>>>>>>>>>>>>> user-supplied
>>>>>> WG
>>>>>>>>>>> does
>>>>>>>>>>>>>>> not need to make sure the watermark doesn't regress.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Regarding making the WG a "rich function", I can see the
>>>>>> potential
>>>>>>>>>>>>>>> benefit but I also see a lot of pitfalls. For example, how
>>>>>>>>>>>>>>> should
>>>>>>>>>> the
>>>>>>>>>>>>>>> watermark state be handled in the case of scale-in? It
>>>>>>>>>>>>>>> could be
>>>>>>>> made
>>>>>>>>>>> to
>>>>>>>>>>>>>>> work in the Kafka case by attaching the state to the
>>>>>>>>>>>>>>> partition
>>>>>>>> state
>>>>>>>>>>>>>>> that we keep, but then we have potential backwards
>>>>>>>>>>>>>>> compatibility
>>>>>>>>>>>>>>> problems also for the WM state. Does the WG usually need
>>>>>>>>>>>>>>> to keep
>>>>>>>> the
>>>>>>>>>>>>>>> state or might it be enough if the state is transient,
>>>>>>>>>>>>>>> i.e. if
>>>>>> you
>>>>>>>>>>> have
>>>>>>>>>>>>>>> a restart the WG would loose its histogram but it would
>>>>>>>>>>>>>>> rebuild
>>>>>> it
>>>>>>>>>>>>>>> quickly and you would get back to the same steady state as
>>>>>> before.
>>>>>>>>>>>>>>> Best,
>>>>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> On 27.04.20 12:12, David Anderson wrote:
>>>>>>>>>>>>>>>> Overall I like this proposal; thanks for bringing it
>>>>>>>>>>>>>>>> forward,
>>>>>>>>>>>>>>>> Aljoscha.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> I also like the idea of making the Watermark generator a
>>>>>>>>>>>>>>>> rich
>>>>>>>>>>> function
>>>>>>>>>>>>>> --
>>>>>>>>>>>>>>>> this should make it more straightforward to implement
>>>>>>>>>>>>>>>> smarter
>>>>>>>>>>>>>>>> watermark
>>>>>>>>>>>>>>>> generators. Eg, one that uses state to keep statistics
>>>>>>>>>>>>>>>> about the
>>>>>>>>>>>>>>>> actual
>>>>>>>>>>>>>>>> out-of-orderness, and uses those statistics to implement a
>>>>>>>> variable
>>>>>>>>>>>>>>> delay.
>>>>>>>>>>>>>>>> David
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
>>>>>>>>>> kkloudas@gmail.com>
>>>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>>>> Hi Aljoscha,
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Thanks for opening the discussion!
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> I have two comments on the FLIP:
>>>>>>>>>>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e.
>>>>>> open()/
>>>>>>>>>>>>>>>>> close(), probably with a Context as argument: I have not
>>>>>>>>>>>>>>>>> fully
>>>>>>>>>>>>>>>>> thought
>>>>>>>>>>>>>>>>> this through but I think that this is more aligned with the
>>>>>> rest
>>>>>>>>>> of
>>>>>>>>>>>>>>>>> our rich functions. In addition, it will allow, for
>>>>>>>>>>>>>>>>> example, to
>>>>>>>>>>>>>>>>> initialize the Watermark value, if we decide to
>>>>>>>>>>>>>>>>> checkpoint the
>>>>>>>>>>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL
>>>>>>>>>>>>>>>>> needs to
>>>>>> do
>>>>>>>>>>>>>>>>> anything in the open()).
>>>>>>>>>>>>>>>>> 2) aligned with the above, and with the case where we
>>>>>>>>>>>>>>>>> want to
>>>>>>>>>>>>>>>>> checkpoint the watermark in mind, I am wondering about
>>>>>>>>>>>>>>>>> how we
>>>>>>>>>> could
>>>>>>>>>>>>>>>>> implement this in the future. In the FLIP, it is
>>>>>>>>>>>>>>>>> proposed to
>>>>>>>>>> expose
>>>>>>>>>>>>>>>>> the WatermarkOutput in the methods of the
>>>>>>>>>>>>>>>>> WatermarkGenerator.
>>>>>>>>>> Given
>>>>>>>>>>>>>>>>> that there is the implicit contract that watermarks are
>>>>>>>>>>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will
>>>>>>>>>>>>>>>>> have
>>>>>> (I
>>>>>>>>>>>>>>>>> assume) a check that will compare the last emitted WM
>>>>>>>>>>>>>>>>> against
>>>>>> the
>>>>>>>>>>>>>>>>> provided one, and emit it only if it is >=. If not, then we
>>>>>> risk
>>>>>>>>>>>>>>>>> having the user shooting himself on the foot if he/she
>>>>>>>>>> accidentally
>>>>>>>>>>>>>>>>> forgets the check. Given that the WatermarkGenerator and
>>>>>>>>>>>>>>>>> its
>>>>>>>>>>>>>>>>> caller do
>>>>>>>>>>>>>>>>> not know if the watermark was finally emitted or not (the
>>>>>>>>>>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
>>>>>>>>>> responsible
>>>>>>>>>>>>>>>>> for checkpointing the WM?
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Given this, why not having the methods as:
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> public interface WatermarkGenerator<T> {
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
>>>>>>>>>>> WatermarkOutput
>>>>>>>>>>>>>>>>> output);
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
>>>>>>>>>>>>>>>>> }
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> and the caller will be the one enforcing any invariants,
>>>>>>>>>>>>>>>>> such
>>>>>> as
>>>>>>>>>>>>>>>>> non-decreasing watermarks. In this way, the caller can
>>>>>> checkpoint
>>>>>>>>>>>>>>>>> anything that is needed as it will have complete
>>>>>>>>>>>>>>>>> knowledge as
>>>>>> to
>>>>>>>>>> if
>>>>>>>>>>>>>>>>> the WM was emitted or not.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> What do you think?
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>>>>>> Kostas
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <
>>>>>> twalthr@apache.org
>>>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
>>>>>>>>>>> unification.
>>>>>>>>>>>>>> We
>>>>>>>>>>>>>>>>>> have considered this FLIP already in the interfaces for
>>>>>> FLIP-95
>>>>>>>>>> [1]
>>>>>>>>>>>>>> and
>>>>>>>>>>>>>>>>>> look forward to update to the new unified watermark
>>>>>>>>>>>>>>>>>> generators
>>>>>>>>>> once
>>>>>>>>>>>>>>>>>> FLIP-126 has been accepted.
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> Regards,
>>>>>>>>>>>>>>>>>> Timo
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
>>>>>>>>>>>>>>>>>>> Hi Everyone!
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> We would like to start a discussion on "FLIP-126:
>>>>>>>>>>>>>>>>>>> Unify (and
>>>>>>>>>>>>>> separate)
>>>>>>>>>>>>>>>>>>> Watermark Assigners" [1]. This work was started by
>>>>>>>>>>>>>>>>>>> Stephan in
>>>>>>>> an
>>>>>>>>>>>>>>>>>>> experimental branch. I expanded on that work to
>>>>>>>>>>>>>>>>>>> provide a PoC
>>>>>>>>>> for
>>>>>>>>>>>>>> the
>>>>>>>>>>>>>>>>>>> changes proposed in this FLIP: [2].
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Currently, we have two different flavours of Watermark
>>>>>>>>>>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
>>>>>>>>>>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
>>>>>>>>>>>>>>>>>>> from TimestampAssigner. This means that sources that
>>>>>>>>>>>>>>>>>>> want to
>>>>>>>>>>>>>>>>>>> support
>>>>>>>>>>>>>>>>>>> watermark assignment/extraction in the source need to
>>>>>>>>>>>>>>>>>>> support
>>>>>>>>>> two
>>>>>>>>>>>>>>>>>>> separate interfaces, we have two operator
>>>>>>>>>>>>>>>>>>> implementations for
>>>>>>>>>> the
>>>>>>>>>>>>>>>>>>> different flavours. Also, this makes features such as
>>>>>>>>>>>>>>>>>>> generic
>>>>>>>>>>>>>> support
>>>>>>>>>>>>>>>>>>> for idleness detection more complicated to implemented
>>>>>> because
>>>>>>>>>> we
>>>>>>>>>>>>>>> again
>>>>>>>>>>>>>>>>>>> have to support two types of watermark assigners.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> In this FLIP we propose two things:
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Unify the Watermark Assigners into one Interface
>>>>>>>>>>> WatermarkGenerator
>>>>>>>>>>>>>>>>>>> Separate this new interface from the TimestampAssigner
>>>>>>>>>>>>>>>>>>> The motivation for the first is to simplify future
>>>>>>>>>> implementations
>>>>>>>>>>>>>> and
>>>>>>>>>>>>>>>>>>> code duplication. The motivation for the second point is
>>>>>> again
>>>>>>>>>>> code
>>>>>>>>>>>>>>>>>>> deduplication, most assigners currently have to extend
>>>>>>>>>>>>>>>>>>> from
>>>>>>>> some
>>>>>>>>>>>>>> base
>>>>>>>>>>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or
>>>>>> users
>>>>>>>>>>>>>>>>>>> have
>>>>>>>>>>>>>>> to
>>>>>>>>>>>>>>>>>>> override an abstract method of the watermark assigner to
>>>>>>>> provide
>>>>>>>>>>>>>>>>>>> the
>>>>>>>>>>>>>>>>>>> timestamp extraction logic.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Additionally, we propose to add a generic wrapping
>>>>>>>>>>>>>> WatermarkGenerator
>>>>>>>>>>>>>>>>>>> that provides idleness detection, i.e. it can mark a
>>>>>>>>>>>>>> stream/partition
>>>>>>>>>>>>>>>>> as
>>>>>>>>>>>>>>>>>>> idle if no data arrives after a configured timeout.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> The "unify and separate" part refers to the fact that
>>>>>>>>>>>>>>>>>>> we want
>>>>>>>> to
>>>>>>>>>>>>>> unify
>>>>>>>>>>>>>>>>>>> punctuated and periodic assigners but at the same time
>>>>>>>>>>>>>>>>>>> split
>>>>>>>> the
>>>>>>>>>>>>>>>>>>> timestamp assigner from the watermark generator.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Please find more details in the FLIP [1]. Looking
>>>>>>>>>>>>>>>>>>> forward to
>>>>>>>>>>>>>>>>>>> your feedback.
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> Best,
>>>>>>>>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>> [1]
>>>>>>>>>>>>>>>>>>>
>>>>>>
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
>>>>>>
>>>>>>>>>>>>>>>>>>> [2]
>>>>>> https://github.com/aljoscha/flink/tree/stephan-event-time
>>>>>>>>
>>>>>>
>>>>
>>>
>>
>>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Guanghui Zhang <be...@gmail.com>.
Hi, @Aljoscha,
the function param currentTimestamp comment does not match the
recordTimestamp "long extractTimestamp(T element, long recordTimestamp)"
on wiki.
Best,
Zhangguanghui
Dawid Wysakowicz <dw...@apache.org> 于2020年5月13日周三 上午12:28写道：
> Thank you for the update and sorry again for chiming in so late...
>
> Best,
>
> Dawid
>
>
> On 12/05/2020 18:21, Aljoscha Krettek wrote:
> > Yes, I am also ok with a SerializableTimestampAssigner. This only
> > looks a bit clumsy in the API but as a user (that uses lambdas) you
> > should not see this. I pushed changes for this to my branch:
> > https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
> >
> > And yes, recordTimestamp sounds good for the TimestampAssigner. I
> > admit I didn't read this well enough and only saw nativeTimestamp.
> >
> > Best,
> > Aljoscha
> >
> > On 12.05.20 17:16, Dawid Wysakowicz wrote:
> >> I have similar thoughts to @Stephan
> >>
> >> Ad. 1 I tried something like this on your branch:
> >>
> >> /**
> >> * Adds the given {@link TimestampAssigner} to this {@link
> >> WatermarkStrategies}. For top-level classes that implement both
> >> Serializable and TimestampAssigner
> >> */
> >> public <TA extends TimestampAssigner<T> & Serializable>
> >> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner) {
> >> checkNotNull(timestampAssigner, "timestampAssigner");
> >> this.timestampAssigner = timestampAssigner;
> >> return this;
> >> }
> >>
> >> @FunctionalInterface
> >> public interface SerializableTimestampAssigner<T> extends
> >> TimestampAssigner<T>, Serializable {
> >> }
> >>
> >> /**
> >> * Adds the given {@link TimestampAssigner} to this {@link
> >> WatermarkStrategies}.
> >> * Helper method for serializable lambdas.
> >> */
> >> public WatermarkStrategies<T>
> >> withTimestampAssigner(SerializableTimestampAssigner<T>
> >> timestampAssigner) {
> >> checkNotNull(timestampAssigner, "timestampAssigner");
> >> this.timestampAssigner = timestampAssigner;
> >> return this;
> >> }
> >>
> >> But I understand if that's too hacky. It's just a pity that we must
> >> enforce limitations on an interface that are not strictly necessary.
> >>
> >> Ad 2/3
> >>
> >> I am aware the watermark assigner/timestamp extractor can be applied
> >> further down the graph. Originally I also wanted to suggest
> >> sourceTimestamp and SourceTimestampAssigner, but then I realized it can
> >> be used also after the sources as you correctly pointed out. Even if the
> >> TimestampAssigner is used after the source there might be some
> >> native/record timestamp in the StreamRecord, that could've been
> >> extracted by previous assigner.
> >>
> >> Best,
> >>
> >> Dawid
> >>
> >> On 12/05/2020 16:47, Stephan Ewen wrote:
> >>> @Aljoscha
> >>>
> >>> About (1) could we have an interface SerializableTimestampAssigner that
> >>> simply mixes in the java.io.Serializable interface? Or will this be too
> >>> clumsy?
> >>>
> >>> About (3) RecordTimeStamp seems to fit both cases (in-source-record
> >>> timestamp, in stream-record timestamp).
> >>>
> >>> On Tue, May 12, 2020 at 4:12 PM Aljoscha Krettek <al...@apache.org>
> >>> wrote:
> >>>
> >>>> Definitely +1 to point 2) raised by Dawid. I'm not sure on points
> >>>> 1) and
> >>>> 3).
> >>>>
> >>>> 1) I can see the benefit of that but in reality most timestamp
> >>>> assigners
> >>>> will probably need to be Serializable. If you look at my (updated) POC
> >>>> branch [1] you can see how a TimestampAssigner would be specified
> >>>> on the
> >>>> WatermarkStrategies helper class: [2]. The signature of this would
> >>>> have
> >>>> to be changed to something like:
> >>>>
> >>>> public <TA extends TimestampAssigner<T> & Serializable>
> >>>> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner)
> >>>>
> >>>> Then, however, it would not be possible for users to specify a
> >>>> lambda or
> >>>> anonymous inner function for the TimestampAssigner like this:
> >>>>
> >>>> WatermarkStrategy<Long> testWmStrategy = WatermarkStrategies
> >>>> .forGenerator(new PeriodicTestWatermarkGenerator())
> >>>> .withTimestampAssigner((event, timestamp) -> event)
> >>>> .build();
> >>>>
> >>>> 3) This makes sense if we only allow WatermarkStrategies on sources,
> >>>> where the previous timestamp really is the "native" timestamp.
> >>>> Currently, we also allow setting watermark strategies at arbitrary
> >>>> points in the graph. I'm thinking we probably should only allow
> >>>> that in
> >>>> sources but it's not the reality currently. I'm not against
> >>>> renaming it,
> >>>> just voicing those thoughts.
> >>>>
> >>>> Best,
> >>>> Aljoscha
> >>>>
> >>>>
> >>>> [1]
> >>>>
> https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
> >>>> [2]
> >>>>
> >>>>
> https://github.com/aljoscha/flink/blob/flink-xxx-wm-generators-rebased/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java#L81
> >>>>
> >>>>
> >>>> On 12.05.20 15:48, Stephan Ewen wrote:
> >>>>> +1 to all of Dawid's suggestions, makes a lot of sense to me
> >>>>>
> >>>>> On Tue, May 12, 2020 at 2:32 PM Dawid Wysakowicz
> >>>>> <dwysakowicz@apache.org
> >>>>>
> >>>>> wrote:
> >>>>>
> >>>>>> Hi Aljoscha,
> >>>>>>
> >>>>>> Sorry for adding comments during the vote, but I have some really
> >>>>>> minor
> >>>>>> suggestions that should not influence the voting thread imo.
> >>>>>>
> >>>>>> 1) Does it make sense to have the TimestampAssigner extend from
> >>>>>> Flink's
> >>>>>> Function? This implies it has to be serializable which with the
> >>>>>> factory
> >>>>>> pattern is not strictly necessary, right? BTW I really like that you
> >>>>>> suggested the FunctionInterface annotation there.
> >>>>>>
> >>>>>> 2) Could we rename the IdentityTimestampAssigner to e.g.
> >>>>>>
> >>>>
> RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
> >>>>
> >>>>>> Personally I found the IdentityTimestampAssigner a bit misleading
> >>>>>> as it
> >>>>>> usually mean a no-op. Which did not click for me, as I assumed it
> >>>>>> somehow returns the incoming record itself.
> >>>>>>
> >>>>>> 3) Could we rename the second parameter of
> >>>>>> TimestampAssigner#extract to
> >>>>>> e.g. recordTimestamp/nativeTimestamp? This is similar to the point
> >>>>>> above. This parameter was also a bit confusing for me as I
> >>>>>> thought at
> >>>>>> times its somehow related to
> >>>>>> TimerService#currentProcessingTimestamp()/currentWatermark() as the
> >>>>>> whole system currentTimestamp.
> >>>>>>
> >>>>>> Other than those three points I like the proposal and I was about to
> >>>>>> vote +1 if it was not for those three points.
> >>>>>>
> >>>>>> Best,
> >>>>>>
> >>>>>> Dawid
> >>>>>>
> >>>>>> On 11/05/2020 16:57, Jark Wu wrote:
> >>>>>>> Thanks for the explanation. I like the fatory pattern to make the
> >>>> member
> >>>>>>> variables immutable and final.
> >>>>>>>
> >>>>>>> So +1 to the proposal.
> >>>>>>>
> >>>>>>> Best,
> >>>>>>> Jark
> >>>>>>>
> >>>>>>> On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org>
> >>>>>>> wrote:
> >>>>>>>
> >>>>>>>> I am fine with that.
> >>>>>>>>
> >>>>>>>> Much of the principles seem agreed upon. I understand the need to
> >>>>>> support
> >>>>>>>> code-generated extractors and we should support most of it
> >>>>>>>> already (as
> >>>>>>>> Aljoscha mentioned via the factories) can extend this if needed.
> >>>>>>>>
> >>>>>>>> I think that the factory approach supports code-generated
> >>>>>>>> extractors
> >>>> in
> >>>>>> a
> >>>>>>>> cleaner way even than an extractor with an open/init method.
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek
> >>>>>>>> <aljoscha@apache.org
> >>>>>>>> wrote:
> >>>>>>>>
> >>>>>>>>> We're slightly running out of time. I would propose we vote on
> >>>>>>>>> the
> >>>>>> basic
> >>>>>>>>> principle and remain open to later additions. This feature is
> >>>>>>>>> quite
> >>>>>>>>> important to make the new Kafka Source that is developed as
> >>>>>>>>> part of
> >>>>>>>>> FLIP-27 useful. Otherwise we would have to use the legacy
> >>>>>>>>> interfaces
> >>>> in
> >>>>>>>>> the newly added connector.
> >>>>>>>>>
> >>>>>>>>> I know that's a bit unorthodox but would everyone be OK with
> >>>>>>>>> what's
> >>>>>>>>> currently there and then we iterate?
> >>>>>>>>>
> >>>>>>>>> Best,
> >>>>>>>>> Aljoscha
> >>>>>>>>>
> >>>>>>>>> On 11.05.20 13:57, Aljoscha Krettek wrote:
> >>>>>>>>>> Ah, I meant to write this in my previous email, sorry about
> >>>>>>>>>> that.
> >>>>>>>>>>
> >>>>>>>>>> The WatermarkStrategy, which is basically a factory for a
> >>>>>>>>>> WatermarkGenerator is the replacement for the open() method.
> >>>>>>>>>> This is
> >>>>>>>> the
> >>>>>>>>>> same strategy that was followed for StreamOperatorFactory,
> >>>>>>>>>> which was
> >>>>>>>>>> introduced to allow code generation in the Table API [1]. If
> >>>>>>>>>> we need
> >>>>>>>>>> metrics or other things we would add that as a parameter to the
> >>>>>> factory
> >>>>>>>>>> method. What do you think?
> >>>>>>>>>>
> >>>>>>>>>> Best,
> >>>>>>>>>> Aljoscha
> >>>>>>>>>>
> >>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
> >>>>>>>>>>
> >>>>>>>>>> On 10.05.20 05:07, Jark Wu wrote:
> >>>>>>>>>>> Hi,
> >>>>>>>>>>>
> >>>>>>>>>>> Regarding to the `open()/close()`, I think it's necessary for
> >>>>>>>>>>> Table&SQL to
> >>>>>>>>>>> compile the generated code.
> >>>>>>>>>>> In Table&SQL, the watermark strategy and event-timestamp is
> >>>>>>>>>>> defined
> >>>>>>>>> using
> >>>>>>>>>>> SQL expressions, we will
> >>>>>>>>>>> translate and generate Java code for the expressions. If we
> >>>>>>>>>>> have
> >>>>>>>>>>> `open()/close()`, we don't need lazy initialization.
> >>>>>>>>>>> Besides that, I can see a need to report some metrics, e.g. the
> >>>>>>>> current
> >>>>>>>>>>> watermark, the dirty timestamps (null value), etc.
> >>>>>>>>>>> So I think a simple `open()/close()` with a context which
> >>>>>>>>>>> can get
> >>>>>>>>>>> MetricGroup is nice and not complex for the first version.
> >>>>>>>>>>>
> >>>>>>>>>>> Best,
> >>>>>>>>>>> Jark
> >>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org>
> >>>> wrote:
> >>>>>>>>>>>> Thanks, Aljoscha, for picking this up.
> >>>>>>>>>>>>
> >>>>>>>>>>>> I agree with the approach of doing the here proposed set of
> >>>> changes
> >>>>>>>> for
> >>>>>>>>>>>> now. It already makes things simpler and adds idleness support
> >>>>>>>>>>>> everywhere.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Rich functions and state always add complexity, let's do
> >>>>>>>>>>>> this in a
> >>>>>>>> next
> >>>>>>>>>>>> step, if we have a really compelling case.
> >>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
> >>>>>>>> aljoscha@apache.org>
> >>>>>>>>>>>> wrote:
> >>>>>>>>>>>>
> >>>>>>>>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The
> >>>>>> proposal
> >>>>>>>>> is
> >>>>>>>>>>>>> basically to turn emitting into a "flatMap", we give the
> >>>>>>>>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and
> >>>>>>>>>>>>> the WG
> >>>>>>>> can
> >>>>>>>>>>>>> decide whether to output a watermark or not and can also
> >>>>>>>>>>>>> mark the
> >>>>>>>>>>>>> output
> >>>>>>>>>>>>> as idle. Changing the interface to return a Watermark (as the
> >>>>>>>> previous
> >>>>>>>>>>>>> watermark assigner interface did) would not allow that
> >>>> flexibility.
> >>>>>>>>>>>>> Regarding checkpointing the watermark and keeping track of
> >>>>>>>>>>>>> the
> >>>>>>>> minimum
> >>>>>>>>>>>>> watermark, this would be the responsibility of the
> >>>>>>>>>>>>> framework (or
> >>>>>> the
> >>>>>>>>>>>>> KafkaConsumer in the current implementation). The
> >>>>>>>>>>>>> user-supplied
> >>>> WG
> >>>>>>>>> does
> >>>>>>>>>>>>> not need to make sure the watermark doesn't regress.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Regarding making the WG a "rich function", I can see the
> >>>> potential
> >>>>>>>>>>>>> benefit but I also see a lot of pitfalls. For example, how
> >>>>>>>>>>>>> should
> >>>>>>>> the
> >>>>>>>>>>>>> watermark state be handled in the case of scale-in? It
> >>>>>>>>>>>>> could be
> >>>>>> made
> >>>>>>>>> to
> >>>>>>>>>>>>> work in the Kafka case by attaching the state to the
> >>>>>>>>>>>>> partition
> >>>>>> state
> >>>>>>>>>>>>> that we keep, but then we have potential backwards
> >>>>>>>>>>>>> compatibility
> >>>>>>>>>>>>> problems also for the WM state. Does the WG usually need
> >>>>>>>>>>>>> to keep
> >>>>>> the
> >>>>>>>>>>>>> state or might it be enough if the state is transient,
> >>>>>>>>>>>>> i.e. if
> >>>> you
> >>>>>>>>> have
> >>>>>>>>>>>>> a restart the WG would loose its histogram but it would
> >>>>>>>>>>>>> rebuild
> >>>> it
> >>>>>>>>>>>>> quickly and you would get back to the same steady state as
> >>>> before.
> >>>>>>>>>>>>> Best,
> >>>>>>>>>>>>> Aljoscha
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> On 27.04.20 12:12, David Anderson wrote:
> >>>>>>>>>>>>>> Overall I like this proposal; thanks for bringing it
> >>>>>>>>>>>>>> forward,
> >>>>>>>>>>>>>> Aljoscha.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> I also like the idea of making the Watermark generator a
> >>>>>>>>>>>>>> rich
> >>>>>>>>> function
> >>>>>>>>>>>> --
> >>>>>>>>>>>>>> this should make it more straightforward to implement
> >>>>>>>>>>>>>> smarter
> >>>>>>>>>>>>>> watermark
> >>>>>>>>>>>>>> generators. Eg, one that uses state to keep statistics
> >>>>>>>>>>>>>> about the
> >>>>>>>>>>>>>> actual
> >>>>>>>>>>>>>> out-of-orderness, and uses those statistics to implement a
> >>>>>> variable
> >>>>>>>>>>>>> delay.
> >>>>>>>>>>>>>> David
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
> >>>>>>>> kkloudas@gmail.com>
> >>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>> Hi Aljoscha,
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Thanks for opening the discussion!
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> I have two comments on the FLIP:
> >>>>>>>>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e.
> >>>> open()/
> >>>>>>>>>>>>>>> close(), probably with a Context as argument: I have not
> >>>>>>>>>>>>>>> fully
> >>>>>>>>>>>>>>> thought
> >>>>>>>>>>>>>>> this through but I think that this is more aligned with the
> >>>> rest
> >>>>>>>> of
> >>>>>>>>>>>>>>> our rich functions. In addition, it will allow, for
> >>>>>>>>>>>>>>> example, to
> >>>>>>>>>>>>>>> initialize the Watermark value, if we decide to
> >>>>>>>>>>>>>>> checkpoint the
> >>>>>>>>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL
> >>>>>>>>>>>>>>> needs to
> >>>> do
> >>>>>>>>>>>>>>> anything in the open()).
> >>>>>>>>>>>>>>> 2) aligned with the above, and with the case where we
> >>>>>>>>>>>>>>> want to
> >>>>>>>>>>>>>>> checkpoint the watermark in mind, I am wondering about
> >>>>>>>>>>>>>>> how we
> >>>>>>>> could
> >>>>>>>>>>>>>>> implement this in the future. In the FLIP, it is
> >>>>>>>>>>>>>>> proposed to
> >>>>>>>> expose
> >>>>>>>>>>>>>>> the WatermarkOutput in the methods of the
> >>>>>>>>>>>>>>> WatermarkGenerator.
> >>>>>>>> Given
> >>>>>>>>>>>>>>> that there is the implicit contract that watermarks are
> >>>>>>>>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will
> >>>>>>>>>>>>>>> have
> >>>> (I
> >>>>>>>>>>>>>>> assume) a check that will compare the last emitted WM
> >>>>>>>>>>>>>>> against
> >>>> the
> >>>>>>>>>>>>>>> provided one, and emit it only if it is >=. If not, then we
> >>>> risk
> >>>>>>>>>>>>>>> having the user shooting himself on the foot if he/she
> >>>>>>>> accidentally
> >>>>>>>>>>>>>>> forgets the check. Given that the WatermarkGenerator and
> >>>>>>>>>>>>>>> its
> >>>>>>>>>>>>>>> caller do
> >>>>>>>>>>>>>>> not know if the watermark was finally emitted or not (the
> >>>>>>>>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
> >>>>>>>> responsible
> >>>>>>>>>>>>>>> for checkpointing the WM?
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Given this, why not having the methods as:
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> public interface WatermarkGenerator<T> {
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
> >>>>>>>>> WatermarkOutput
> >>>>>>>>>>>>>>> output);
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
> >>>>>>>>>>>>>>> }
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> and the caller will be the one enforcing any invariants,
> >>>>>>>>>>>>>>> such
> >>>> as
> >>>>>>>>>>>>>>> non-decreasing watermarks. In this way, the caller can
> >>>> checkpoint
> >>>>>>>>>>>>>>> anything that is needed as it will have complete
> >>>>>>>>>>>>>>> knowledge as
> >>>> to
> >>>>>>>> if
> >>>>>>>>>>>>>>> the WM was emitted or not.
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> What do you think?
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Cheers,
> >>>>>>>>>>>>>>> Kostas
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <
> >>>> twalthr@apache.org
> >>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
> >>>>>>>>> unification.
> >>>>>>>>>>>> We
> >>>>>>>>>>>>>>>> have considered this FLIP already in the interfaces for
> >>>> FLIP-95
> >>>>>>>> [1]
> >>>>>>>>>>>> and
> >>>>>>>>>>>>>>>> look forward to update to the new unified watermark
> >>>>>>>>>>>>>>>> generators
> >>>>>>>> once
> >>>>>>>>>>>>>>>> FLIP-126 has been accepted.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Regards,
> >>>>>>>>>>>>>>>> Timo
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
> >>>>>>>>>>>>>>>>> Hi Everyone!
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> We would like to start a discussion on "FLIP-126:
> >>>>>>>>>>>>>>>>> Unify (and
> >>>>>>>>>>>> separate)
> >>>>>>>>>>>>>>>>> Watermark Assigners" [1]. This work was started by
> >>>>>>>>>>>>>>>>> Stephan in
> >>>>>> an
> >>>>>>>>>>>>>>>>> experimental branch. I expanded on that work to
> >>>>>>>>>>>>>>>>> provide a PoC
> >>>>>>>> for
> >>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>> changes proposed in this FLIP: [2].
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Currently, we have two different flavours of Watermark
> >>>>>>>>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
> >>>>>>>>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
> >>>>>>>>>>>>>>>>> from TimestampAssigner. This means that sources that
> >>>>>>>>>>>>>>>>> want to
> >>>>>>>>>>>>>>>>> support
> >>>>>>>>>>>>>>>>> watermark assignment/extraction in the source need to
> >>>>>>>>>>>>>>>>> support
> >>>>>>>> two
> >>>>>>>>>>>>>>>>> separate interfaces, we have two operator
> >>>>>>>>>>>>>>>>> implementations for
> >>>>>>>> the
> >>>>>>>>>>>>>>>>> different flavours. Also, this makes features such as
> >>>>>>>>>>>>>>>>> generic
> >>>>>>>>>>>> support
> >>>>>>>>>>>>>>>>> for idleness detection more complicated to implemented
> >>>> because
> >>>>>>>> we
> >>>>>>>>>>>>> again
> >>>>>>>>>>>>>>>>> have to support two types of watermark assigners.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> In this FLIP we propose two things:
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Unify the Watermark Assigners into one Interface
> >>>>>>>>> WatermarkGenerator
> >>>>>>>>>>>>>>>>> Separate this new interface from the TimestampAssigner
> >>>>>>>>>>>>>>>>> The motivation for the first is to simplify future
> >>>>>>>> implementations
> >>>>>>>>>>>> and
> >>>>>>>>>>>>>>>>> code duplication. The motivation for the second point is
> >>>> again
> >>>>>>>>> code
> >>>>>>>>>>>>>>>>> deduplication, most assigners currently have to extend
> >>>>>>>>>>>>>>>>> from
> >>>>>> some
> >>>>>>>>>>>> base
> >>>>>>>>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or
> >>>> users
> >>>>>>>>>>>>>>>>> have
> >>>>>>>>>>>>> to
> >>>>>>>>>>>>>>>>> override an abstract method of the watermark assigner to
> >>>>>> provide
> >>>>>>>>>>>>>>>>> the
> >>>>>>>>>>>>>>>>> timestamp extraction logic.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Additionally, we propose to add a generic wrapping
> >>>>>>>>>>>> WatermarkGenerator
> >>>>>>>>>>>>>>>>> that provides idleness detection, i.e. it can mark a
> >>>>>>>>>>>> stream/partition
> >>>>>>>>>>>>>>> as
> >>>>>>>>>>>>>>>>> idle if no data arrives after a configured timeout.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> The "unify and separate" part refers to the fact that
> >>>>>>>>>>>>>>>>> we want
> >>>>>> to
> >>>>>>>>>>>> unify
> >>>>>>>>>>>>>>>>> punctuated and periodic assigners but at the same time
> >>>>>>>>>>>>>>>>> split
> >>>>>> the
> >>>>>>>>>>>>>>>>> timestamp assigner from the watermark generator.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Please find more details in the FLIP [1]. Looking
> >>>>>>>>>>>>>>>>> forward to
> >>>>>>>>>>>>>>>>> your feedback.
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> Best,
> >>>>>>>>>>>>>>>>> Aljoscha
> >>>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>>> [1]
> >>>>>>>>>>>>>>>>>
> >>>>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
> >>>>
> >>>>>>>>>>>>>>>>> [2]
> >>>> https://github.com/aljoscha/flink/tree/stephan-event-time
> >>>>>>
> >>>>
> >>
> >
>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Dawid Wysakowicz <dw...@apache.org>.
Thank you for the update and sorry again for chiming in so late...
Best,
Dawid
On 12/05/2020 18:21, Aljoscha Krettek wrote:
> Yes, I am also ok with a SerializableTimestampAssigner. This only
> looks a bit clumsy in the API but as a user (that uses lambdas) you
> should not see this. I pushed changes for this to my branch:
> https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
>
> And yes, recordTimestamp sounds good for the TimestampAssigner. I
> admit I didn't read this well enough and only saw nativeTimestamp.
>
> Best,
> Aljoscha
>
> On 12.05.20 17:16, Dawid Wysakowicz wrote:
>> I have similar thoughts to @Stephan
>>
>> Ad. 1 I tried something like this on your branch:
>>
>> /**
>> * Adds the given {@link TimestampAssigner} to this {@link
>> WatermarkStrategies}. For top-level classes that implement both
>> Serializable and TimestampAssigner
>> */
>> public <TA extends TimestampAssigner<T> & Serializable>
>> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner) {
>> checkNotNull(timestampAssigner, "timestampAssigner");
>> this.timestampAssigner = timestampAssigner;
>> return this;
>> }
>>
>> @FunctionalInterface
>> public interface SerializableTimestampAssigner<T> extends
>> TimestampAssigner<T>, Serializable {
>> }
>>
>> /**
>> * Adds the given {@link TimestampAssigner} to this {@link
>> WatermarkStrategies}.
>> * Helper method for serializable lambdas.
>> */
>> public WatermarkStrategies<T>
>> withTimestampAssigner(SerializableTimestampAssigner<T>
>> timestampAssigner) {
>> checkNotNull(timestampAssigner, "timestampAssigner");
>> this.timestampAssigner = timestampAssigner;
>> return this;
>> }
>>
>> But I understand if that's too hacky. It's just a pity that we must
>> enforce limitations on an interface that are not strictly necessary.
>>
>> Ad 2/3
>>
>> I am aware the watermark assigner/timestamp extractor can be applied
>> further down the graph. Originally I also wanted to suggest
>> sourceTimestamp and SourceTimestampAssigner, but then I realized it can
>> be used also after the sources as you correctly pointed out. Even if the
>> TimestampAssigner is used after the source there might be some
>> native/record timestamp in the StreamRecord, that could've been
>> extracted by previous assigner.
>>
>> Best,
>>
>> Dawid
>>
>> On 12/05/2020 16:47, Stephan Ewen wrote:
>>> @Aljoscha
>>>
>>> About (1) could we have an interface SerializableTimestampAssigner that
>>> simply mixes in the java.io.Serializable interface? Or will this be too
>>> clumsy?
>>>
>>> About (3) RecordTimeStamp seems to fit both cases (in-source-record
>>> timestamp, in stream-record timestamp).
>>>
>>> On Tue, May 12, 2020 at 4:12 PM Aljoscha Krettek <al...@apache.org>
>>> wrote:
>>>
>>>> Definitely +1 to point 2) raised by Dawid. I'm not sure on points
>>>> 1) and
>>>> 3).
>>>>
>>>> 1) I can see the benefit of that but in reality most timestamp
>>>> assigners
>>>> will probably need to be Serializable. If you look at my (updated) POC
>>>> branch [1] you can see how a TimestampAssigner would be specified
>>>> on the
>>>> WatermarkStrategies helper class: [2]. The signature of this would
>>>> have
>>>> to be changed to something like:
>>>>
>>>> public <TA extends TimestampAssigner<T> & Serializable>
>>>> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner)
>>>>
>>>> Then, however, it would not be possible for users to specify a
>>>> lambda or
>>>> anonymous inner function for the TimestampAssigner like this:
>>>>
>>>> WatermarkStrategy<Long> testWmStrategy = WatermarkStrategies
>>>> .forGenerator(new PeriodicTestWatermarkGenerator())
>>>> .withTimestampAssigner((event, timestamp) -> event)
>>>> .build();
>>>>
>>>> 3) This makes sense if we only allow WatermarkStrategies on sources,
>>>> where the previous timestamp really is the "native" timestamp.
>>>> Currently, we also allow setting watermark strategies at arbitrary
>>>> points in the graph. I'm thinking we probably should only allow
>>>> that in
>>>> sources but it's not the reality currently. I'm not against
>>>> renaming it,
>>>> just voicing those thoughts.
>>>>
>>>> Best,
>>>> Aljoscha
>>>>
>>>>
>>>> [1]
>>>> https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
>>>> [2]
>>>>
>>>> https://github.com/aljoscha/flink/blob/flink-xxx-wm-generators-rebased/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java#L81
>>>>
>>>>
>>>> On 12.05.20 15:48, Stephan Ewen wrote:
>>>>> +1 to all of Dawid's suggestions, makes a lot of sense to me
>>>>>
>>>>> On Tue, May 12, 2020 at 2:32 PM Dawid Wysakowicz
>>>>> <dwysakowicz@apache.org
>>>>>
>>>>> wrote:
>>>>>
>>>>>> Hi Aljoscha,
>>>>>>
>>>>>> Sorry for adding comments during the vote, but I have some really
>>>>>> minor
>>>>>> suggestions that should not influence the voting thread imo.
>>>>>>
>>>>>> 1) Does it make sense to have the TimestampAssigner extend from
>>>>>> Flink's
>>>>>> Function? This implies it has to be serializable which with the
>>>>>> factory
>>>>>> pattern is not strictly necessary, right? BTW I really like that you
>>>>>> suggested the FunctionInterface annotation there.
>>>>>>
>>>>>> 2) Could we rename the IdentityTimestampAssigner to e.g.
>>>>>>
>>>> RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
>>>>
>>>>>> Personally I found the IdentityTimestampAssigner a bit misleading
>>>>>> as it
>>>>>> usually mean a no-op. Which did not click for me, as I assumed it
>>>>>> somehow returns the incoming record itself.
>>>>>>
>>>>>> 3) Could we rename the second parameter of
>>>>>> TimestampAssigner#extract to
>>>>>> e.g. recordTimestamp/nativeTimestamp? This is similar to the point
>>>>>> above. This parameter was also a bit confusing for me as I
>>>>>> thought at
>>>>>> times its somehow related to
>>>>>> TimerService#currentProcessingTimestamp()/currentWatermark() as the
>>>>>> whole system currentTimestamp.
>>>>>>
>>>>>> Other than those three points I like the proposal and I was about to
>>>>>> vote +1 if it was not for those three points.
>>>>>>
>>>>>> Best,
>>>>>>
>>>>>> Dawid
>>>>>>
>>>>>> On 11/05/2020 16:57, Jark Wu wrote:
>>>>>>> Thanks for the explanation. I like the fatory pattern to make the
>>>> member
>>>>>>> variables immutable and final.
>>>>>>>
>>>>>>> So +1 to the proposal.
>>>>>>>
>>>>>>> Best,
>>>>>>> Jark
>>>>>>>
>>>>>>> On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> I am fine with that.
>>>>>>>>
>>>>>>>> Much of the principles seem agreed upon. I understand the need to
>>>>>> support
>>>>>>>> code-generated extractors and we should support most of it
>>>>>>>> already (as
>>>>>>>> Aljoscha mentioned via the factories) can extend this if needed.
>>>>>>>>
>>>>>>>> I think that the factory approach supports code-generated
>>>>>>>> extractors
>>>> in
>>>>>> a
>>>>>>>> cleaner way even than an extractor with an open/init method.
>>>>>>>>
>>>>>>>>
>>>>>>>> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek
>>>>>>>> <aljoscha@apache.org
>>>>>>>> wrote:
>>>>>>>>
>>>>>>>>> We're slightly running out of time. I would propose we vote on
>>>>>>>>> the
>>>>>> basic
>>>>>>>>> principle and remain open to later additions. This feature is
>>>>>>>>> quite
>>>>>>>>> important to make the new Kafka Source that is developed as
>>>>>>>>> part of
>>>>>>>>> FLIP-27 useful. Otherwise we would have to use the legacy
>>>>>>>>> interfaces
>>>> in
>>>>>>>>> the newly added connector.
>>>>>>>>>
>>>>>>>>> I know that's a bit unorthodox but would everyone be OK with
>>>>>>>>> what's
>>>>>>>>> currently there and then we iterate?
>>>>>>>>>
>>>>>>>>> Best,
>>>>>>>>> Aljoscha
>>>>>>>>>
>>>>>>>>> On 11.05.20 13:57, Aljoscha Krettek wrote:
>>>>>>>>>> Ah, I meant to write this in my previous email, sorry about
>>>>>>>>>> that.
>>>>>>>>>>
>>>>>>>>>> The WatermarkStrategy, which is basically a factory for a
>>>>>>>>>> WatermarkGenerator is the replacement for the open() method.
>>>>>>>>>> This is
>>>>>>>> the
>>>>>>>>>> same strategy that was followed for StreamOperatorFactory,
>>>>>>>>>> which was
>>>>>>>>>> introduced to allow code generation in the Table API [1]. If
>>>>>>>>>> we need
>>>>>>>>>> metrics or other things we would add that as a parameter to the
>>>>>> factory
>>>>>>>>>> method. What do you think?
>>>>>>>>>>
>>>>>>>>>> Best,
>>>>>>>>>> Aljoscha
>>>>>>>>>>
>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
>>>>>>>>>>
>>>>>>>>>> On 10.05.20 05:07, Jark Wu wrote:
>>>>>>>>>>> Hi,
>>>>>>>>>>>
>>>>>>>>>>> Regarding to the `open()/close()`, I think it's necessary for
>>>>>>>>>>> Table&SQL to
>>>>>>>>>>> compile the generated code.
>>>>>>>>>>> In Table&SQL, the watermark strategy and event-timestamp is
>>>>>>>>>>> defined
>>>>>>>>> using
>>>>>>>>>>> SQL expressions, we will
>>>>>>>>>>> translate and generate Java code for the expressions. If we
>>>>>>>>>>> have
>>>>>>>>>>> `open()/close()`, we don't need lazy initialization.
>>>>>>>>>>> Besides that, I can see a need to report some metrics, e.g. the
>>>>>>>> current
>>>>>>>>>>> watermark, the dirty timestamps (null value), etc.
>>>>>>>>>>> So I think a simple `open()/close()` with a context which
>>>>>>>>>>> can get
>>>>>>>>>>> MetricGroup is nice and not complex for the first version.
>>>>>>>>>>>
>>>>>>>>>>> Best,
>>>>>>>>>>> Jark
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org>
>>>> wrote:
>>>>>>>>>>>> Thanks, Aljoscha, for picking this up.
>>>>>>>>>>>>
>>>>>>>>>>>> I agree with the approach of doing the here proposed set of
>>>> changes
>>>>>>>> for
>>>>>>>>>>>> now. It already makes things simpler and adds idleness support
>>>>>>>>>>>> everywhere.
>>>>>>>>>>>>
>>>>>>>>>>>> Rich functions and state always add complexity, let's do
>>>>>>>>>>>> this in a
>>>>>>>> next
>>>>>>>>>>>> step, if we have a really compelling case.
>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
>>>>>>>> aljoscha@apache.org>
>>>>>>>>>>>> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The
>>>>>> proposal
>>>>>>>>> is
>>>>>>>>>>>>> basically to turn emitting into a "flatMap", we give the
>>>>>>>>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and
>>>>>>>>>>>>> the WG
>>>>>>>> can
>>>>>>>>>>>>> decide whether to output a watermark or not and can also
>>>>>>>>>>>>> mark the
>>>>>>>>>>>>> output
>>>>>>>>>>>>> as idle. Changing the interface to return a Watermark (as the
>>>>>>>> previous
>>>>>>>>>>>>> watermark assigner interface did) would not allow that
>>>> flexibility.
>>>>>>>>>>>>> Regarding checkpointing the watermark and keeping track of
>>>>>>>>>>>>> the
>>>>>>>> minimum
>>>>>>>>>>>>> watermark, this would be the responsibility of the
>>>>>>>>>>>>> framework (or
>>>>>> the
>>>>>>>>>>>>> KafkaConsumer in the current implementation). The
>>>>>>>>>>>>> user-supplied
>>>> WG
>>>>>>>>> does
>>>>>>>>>>>>> not need to make sure the watermark doesn't regress.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Regarding making the WG a "rich function", I can see the
>>>> potential
>>>>>>>>>>>>> benefit but I also see a lot of pitfalls. For example, how
>>>>>>>>>>>>> should
>>>>>>>> the
>>>>>>>>>>>>> watermark state be handled in the case of scale-in? It
>>>>>>>>>>>>> could be
>>>>>> made
>>>>>>>>> to
>>>>>>>>>>>>> work in the Kafka case by attaching the state to the
>>>>>>>>>>>>> partition
>>>>>> state
>>>>>>>>>>>>> that we keep, but then we have potential backwards
>>>>>>>>>>>>> compatibility
>>>>>>>>>>>>> problems also for the WM state. Does the WG usually need
>>>>>>>>>>>>> to keep
>>>>>> the
>>>>>>>>>>>>> state or might it be enough if the state is transient,
>>>>>>>>>>>>> i.e. if
>>>> you
>>>>>>>>> have
>>>>>>>>>>>>> a restart the WG would loose its histogram but it would
>>>>>>>>>>>>> rebuild
>>>> it
>>>>>>>>>>>>> quickly and you would get back to the same steady state as
>>>> before.
>>>>>>>>>>>>> Best,
>>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>>
>>>>>>>>>>>>> On 27.04.20 12:12, David Anderson wrote:
>>>>>>>>>>>>>> Overall I like this proposal; thanks for bringing it
>>>>>>>>>>>>>> forward,
>>>>>>>>>>>>>> Aljoscha.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> I also like the idea of making the Watermark generator a
>>>>>>>>>>>>>> rich
>>>>>>>>> function
>>>>>>>>>>>> --
>>>>>>>>>>>>>> this should make it more straightforward to implement
>>>>>>>>>>>>>> smarter
>>>>>>>>>>>>>> watermark
>>>>>>>>>>>>>> generators. Eg, one that uses state to keep statistics
>>>>>>>>>>>>>> about the
>>>>>>>>>>>>>> actual
>>>>>>>>>>>>>> out-of-orderness, and uses those statistics to implement a
>>>>>> variable
>>>>>>>>>>>>> delay.
>>>>>>>>>>>>>> David
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
>>>>>>>> kkloudas@gmail.com>
>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>> Hi Aljoscha,
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Thanks for opening the discussion!
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> I have two comments on the FLIP:
>>>>>>>>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e.
>>>> open()/
>>>>>>>>>>>>>>> close(), probably with a Context as argument: I have not
>>>>>>>>>>>>>>> fully
>>>>>>>>>>>>>>> thought
>>>>>>>>>>>>>>> this through but I think that this is more aligned with the
>>>> rest
>>>>>>>> of
>>>>>>>>>>>>>>> our rich functions. In addition, it will allow, for
>>>>>>>>>>>>>>> example, to
>>>>>>>>>>>>>>> initialize the Watermark value, if we decide to
>>>>>>>>>>>>>>> checkpoint the
>>>>>>>>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL
>>>>>>>>>>>>>>> needs to
>>>> do
>>>>>>>>>>>>>>> anything in the open()).
>>>>>>>>>>>>>>> 2) aligned with the above, and with the case where we
>>>>>>>>>>>>>>> want to
>>>>>>>>>>>>>>> checkpoint the watermark in mind, I am wondering about
>>>>>>>>>>>>>>> how we
>>>>>>>> could
>>>>>>>>>>>>>>> implement this in the future. In the FLIP, it is
>>>>>>>>>>>>>>> proposed to
>>>>>>>> expose
>>>>>>>>>>>>>>> the WatermarkOutput in the methods of the
>>>>>>>>>>>>>>> WatermarkGenerator.
>>>>>>>> Given
>>>>>>>>>>>>>>> that there is the implicit contract that watermarks are
>>>>>>>>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will
>>>>>>>>>>>>>>> have
>>>> (I
>>>>>>>>>>>>>>> assume) a check that will compare the last emitted WM
>>>>>>>>>>>>>>> against
>>>> the
>>>>>>>>>>>>>>> provided one, and emit it only if it is >=. If not, then we
>>>> risk
>>>>>>>>>>>>>>> having the user shooting himself on the foot if he/she
>>>>>>>> accidentally
>>>>>>>>>>>>>>> forgets the check. Given that the WatermarkGenerator and
>>>>>>>>>>>>>>> its
>>>>>>>>>>>>>>> caller do
>>>>>>>>>>>>>>> not know if the watermark was finally emitted or not (the
>>>>>>>>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
>>>>>>>> responsible
>>>>>>>>>>>>>>> for checkpointing the WM?
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Given this, why not having the methods as:
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> public interface WatermarkGenerator<T> {
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
>>>>>>>>> WatermarkOutput
>>>>>>>>>>>>>>> output);
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
>>>>>>>>>>>>>>> }
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> and the caller will be the one enforcing any invariants,
>>>>>>>>>>>>>>> such
>>>> as
>>>>>>>>>>>>>>> non-decreasing watermarks. In this way, the caller can
>>>> checkpoint
>>>>>>>>>>>>>>> anything that is needed as it will have complete
>>>>>>>>>>>>>>> knowledge as
>>>> to
>>>>>>>> if
>>>>>>>>>>>>>>> the WM was emitted or not.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> What do you think?
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>>>> Kostas
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <
>>>> twalthr@apache.org
>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
>>>>>>>>> unification.
>>>>>>>>>>>> We
>>>>>>>>>>>>>>>> have considered this FLIP already in the interfaces for
>>>> FLIP-95
>>>>>>>> [1]
>>>>>>>>>>>> and
>>>>>>>>>>>>>>>> look forward to update to the new unified watermark
>>>>>>>>>>>>>>>> generators
>>>>>>>> once
>>>>>>>>>>>>>>>> FLIP-126 has been accepted.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Regards,
>>>>>>>>>>>>>>>> Timo
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
>>>>>>>>>>>>>>>>> Hi Everyone!
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> We would like to start a discussion on "FLIP-126:
>>>>>>>>>>>>>>>>> Unify (and
>>>>>>>>>>>> separate)
>>>>>>>>>>>>>>>>> Watermark Assigners" [1]. This work was started by
>>>>>>>>>>>>>>>>> Stephan in
>>>>>> an
>>>>>>>>>>>>>>>>> experimental branch. I expanded on that work to
>>>>>>>>>>>>>>>>> provide a PoC
>>>>>>>> for
>>>>>>>>>>>> the
>>>>>>>>>>>>>>>>> changes proposed in this FLIP: [2].
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Currently, we have two different flavours of Watermark
>>>>>>>>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
>>>>>>>>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
>>>>>>>>>>>>>>>>> from TimestampAssigner. This means that sources that
>>>>>>>>>>>>>>>>> want to
>>>>>>>>>>>>>>>>> support
>>>>>>>>>>>>>>>>> watermark assignment/extraction in the source need to
>>>>>>>>>>>>>>>>> support
>>>>>>>> two
>>>>>>>>>>>>>>>>> separate interfaces, we have two operator
>>>>>>>>>>>>>>>>> implementations for
>>>>>>>> the
>>>>>>>>>>>>>>>>> different flavours. Also, this makes features such as
>>>>>>>>>>>>>>>>> generic
>>>>>>>>>>>> support
>>>>>>>>>>>>>>>>> for idleness detection more complicated to implemented
>>>> because
>>>>>>>> we
>>>>>>>>>>>>> again
>>>>>>>>>>>>>>>>> have to support two types of watermark assigners.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> In this FLIP we propose two things:
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Unify the Watermark Assigners into one Interface
>>>>>>>>> WatermarkGenerator
>>>>>>>>>>>>>>>>> Separate this new interface from the TimestampAssigner
>>>>>>>>>>>>>>>>> The motivation for the first is to simplify future
>>>>>>>> implementations
>>>>>>>>>>>> and
>>>>>>>>>>>>>>>>> code duplication. The motivation for the second point is
>>>> again
>>>>>>>>> code
>>>>>>>>>>>>>>>>> deduplication, most assigners currently have to extend
>>>>>>>>>>>>>>>>> from
>>>>>> some
>>>>>>>>>>>> base
>>>>>>>>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or
>>>> users
>>>>>>>>>>>>>>>>> have
>>>>>>>>>>>>> to
>>>>>>>>>>>>>>>>> override an abstract method of the watermark assigner to
>>>>>> provide
>>>>>>>>>>>>>>>>> the
>>>>>>>>>>>>>>>>> timestamp extraction logic.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Additionally, we propose to add a generic wrapping
>>>>>>>>>>>> WatermarkGenerator
>>>>>>>>>>>>>>>>> that provides idleness detection, i.e. it can mark a
>>>>>>>>>>>> stream/partition
>>>>>>>>>>>>>>> as
>>>>>>>>>>>>>>>>> idle if no data arrives after a configured timeout.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> The "unify and separate" part refers to the fact that
>>>>>>>>>>>>>>>>> we want
>>>>>> to
>>>>>>>>>>>> unify
>>>>>>>>>>>>>>>>> punctuated and periodic assigners but at the same time
>>>>>>>>>>>>>>>>> split
>>>>>> the
>>>>>>>>>>>>>>>>> timestamp assigner from the watermark generator.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Please find more details in the FLIP [1]. Looking
>>>>>>>>>>>>>>>>> forward to
>>>>>>>>>>>>>>>>> your feedback.
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> Best,
>>>>>>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>> [1]
>>>>>>>>>>>>>>>>>
>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
>>>>
>>>>>>>>>>>>>>>>> [2]
>>>> https://github.com/aljoscha/flink/tree/stephan-event-time
>>>>>>
>>>>
>>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Aljoscha Krettek <al...@apache.org>.
Yes, I am also ok with a SerializableTimestampAssigner. This only looks
a bit clumsy in the API but as a user (that uses lambdas) you should not
see this. I pushed changes for this to my branch:
https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
And yes, recordTimestamp sounds good for the TimestampAssigner. I admit
I didn't read this well enough and only saw nativeTimestamp.
Best,
Aljoscha
On 12.05.20 17:16, Dawid Wysakowicz wrote:
> I have similar thoughts to @Stephan
>
> Ad. 1 I tried something like this on your branch:
>
> /**
> * Adds the given {@link TimestampAssigner} to this {@link
> WatermarkStrategies}. For top-level classes that implement both
> Serializable and TimestampAssigner
> */
> public <TA extends TimestampAssigner<T> & Serializable>
> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner) {
> checkNotNull(timestampAssigner, "timestampAssigner");
> this.timestampAssigner = timestampAssigner;
> return this;
> }
>
> @FunctionalInterface
> public interface SerializableTimestampAssigner<T> extends
> TimestampAssigner<T>, Serializable {
> }
>
> /**
> * Adds the given {@link TimestampAssigner} to this {@link
> WatermarkStrategies}.
> * Helper method for serializable lambdas.
> */
> public WatermarkStrategies<T>
> withTimestampAssigner(SerializableTimestampAssigner<T> timestampAssigner) {
> checkNotNull(timestampAssigner, "timestampAssigner");
> this.timestampAssigner = timestampAssigner;
> return this;
> }
>
> But I understand if that's too hacky. It's just a pity that we must
> enforce limitations on an interface that are not strictly necessary.
>
> Ad 2/3
>
> I am aware the watermark assigner/timestamp extractor can be applied
> further down the graph. Originally I also wanted to suggest
> sourceTimestamp and SourceTimestampAssigner, but then I realized it can
> be used also after the sources as you correctly pointed out. Even if the
> TimestampAssigner is used after the source there might be some
> native/record timestamp in the StreamRecord, that could've been
> extracted by previous assigner.
>
> Best,
>
> Dawid
>
> On 12/05/2020 16:47, Stephan Ewen wrote:
>> @Aljoscha
>>
>> About (1) could we have an interface SerializableTimestampAssigner that
>> simply mixes in the java.io.Serializable interface? Or will this be too
>> clumsy?
>>
>> About (3) RecordTimeStamp seems to fit both cases (in-source-record
>> timestamp, in stream-record timestamp).
>>
>> On Tue, May 12, 2020 at 4:12 PM Aljoscha Krettek <al...@apache.org>
>> wrote:
>>
>>> Definitely +1 to point 2) raised by Dawid. I'm not sure on points 1) and
>>> 3).
>>>
>>> 1) I can see the benefit of that but in reality most timestamp assigners
>>> will probably need to be Serializable. If you look at my (updated) POC
>>> branch [1] you can see how a TimestampAssigner would be specified on the
>>> WatermarkStrategies helper class: [2]. The signature of this would have
>>> to be changed to something like:
>>>
>>> public <TA extends TimestampAssigner<T> & Serializable>
>>> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner)
>>>
>>> Then, however, it would not be possible for users to specify a lambda or
>>> anonymous inner function for the TimestampAssigner like this:
>>>
>>> WatermarkStrategy<Long> testWmStrategy = WatermarkStrategies
>>> .forGenerator(new PeriodicTestWatermarkGenerator())
>>> .withTimestampAssigner((event, timestamp) -> event)
>>> .build();
>>>
>>> 3) This makes sense if we only allow WatermarkStrategies on sources,
>>> where the previous timestamp really is the "native" timestamp.
>>> Currently, we also allow setting watermark strategies at arbitrary
>>> points in the graph. I'm thinking we probably should only allow that in
>>> sources but it's not the reality currently. I'm not against renaming it,
>>> just voicing those thoughts.
>>>
>>> Best,
>>> Aljoscha
>>>
>>>
>>> [1] https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
>>> [2]
>>>
>>> https://github.com/aljoscha/flink/blob/flink-xxx-wm-generators-rebased/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java#L81
>>>
>>> On 12.05.20 15:48, Stephan Ewen wrote:
>>>> +1 to all of Dawid's suggestions, makes a lot of sense to me
>>>>
>>>> On Tue, May 12, 2020 at 2:32 PM Dawid Wysakowicz <dwysakowicz@apache.org
>>>>
>>>> wrote:
>>>>
>>>>> Hi Aljoscha,
>>>>>
>>>>> Sorry for adding comments during the vote, but I have some really minor
>>>>> suggestions that should not influence the voting thread imo.
>>>>>
>>>>> 1) Does it make sense to have the TimestampAssigner extend from Flink's
>>>>> Function? This implies it has to be serializable which with the factory
>>>>> pattern is not strictly necessary, right? BTW I really like that you
>>>>> suggested the FunctionInterface annotation there.
>>>>>
>>>>> 2) Could we rename the IdentityTimestampAssigner to e.g.
>>>>>
>>> RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
>>>>> Personally I found the IdentityTimestampAssigner a bit misleading as it
>>>>> usually mean a no-op. Which did not click for me, as I assumed it
>>>>> somehow returns the incoming record itself.
>>>>>
>>>>> 3) Could we rename the second parameter of TimestampAssigner#extract to
>>>>> e.g. recordTimestamp/nativeTimestamp? This is similar to the point
>>>>> above. This parameter was also a bit confusing for me as I thought at
>>>>> times its somehow related to
>>>>> TimerService#currentProcessingTimestamp()/currentWatermark() as the
>>>>> whole system currentTimestamp.
>>>>>
>>>>> Other than those three points I like the proposal and I was about to
>>>>> vote +1 if it was not for those three points.
>>>>>
>>>>> Best,
>>>>>
>>>>> Dawid
>>>>>
>>>>> On 11/05/2020 16:57, Jark Wu wrote:
>>>>>> Thanks for the explanation. I like the fatory pattern to make the
>>> member
>>>>>> variables immutable and final.
>>>>>>
>>>>>> So +1 to the proposal.
>>>>>>
>>>>>> Best,
>>>>>> Jark
>>>>>>
>>>>>> On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org> wrote:
>>>>>>
>>>>>>> I am fine with that.
>>>>>>>
>>>>>>> Much of the principles seem agreed upon. I understand the need to
>>>>> support
>>>>>>> code-generated extractors and we should support most of it already (as
>>>>>>> Aljoscha mentioned via the factories) can extend this if needed.
>>>>>>>
>>>>>>> I think that the factory approach supports code-generated extractors
>>> in
>>>>> a
>>>>>>> cleaner way even than an extractor with an open/init method.
>>>>>>>
>>>>>>>
>>>>>>> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek <aljoscha@apache.org
>>>>>>> wrote:
>>>>>>>
>>>>>>>> We're slightly running out of time. I would propose we vote on the
>>>>> basic
>>>>>>>> principle and remain open to later additions. This feature is quite
>>>>>>>> important to make the new Kafka Source that is developed as part of
>>>>>>>> FLIP-27 useful. Otherwise we would have to use the legacy interfaces
>>> in
>>>>>>>> the newly added connector.
>>>>>>>>
>>>>>>>> I know that's a bit unorthodox but would everyone be OK with what's
>>>>>>>> currently there and then we iterate?
>>>>>>>>
>>>>>>>> Best,
>>>>>>>> Aljoscha
>>>>>>>>
>>>>>>>> On 11.05.20 13:57, Aljoscha Krettek wrote:
>>>>>>>>> Ah, I meant to write this in my previous email, sorry about that.
>>>>>>>>>
>>>>>>>>> The WatermarkStrategy, which is basically a factory for a
>>>>>>>>> WatermarkGenerator is the replacement for the open() method. This is
>>>>>>> the
>>>>>>>>> same strategy that was followed for StreamOperatorFactory, which was
>>>>>>>>> introduced to allow code generation in the Table API [1]. If we need
>>>>>>>>> metrics or other things we would add that as a parameter to the
>>>>> factory
>>>>>>>>> method. What do you think?
>>>>>>>>>
>>>>>>>>> Best,
>>>>>>>>> Aljoscha
>>>>>>>>>
>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
>>>>>>>>>
>>>>>>>>> On 10.05.20 05:07, Jark Wu wrote:
>>>>>>>>>> Hi,
>>>>>>>>>>
>>>>>>>>>> Regarding to the `open()/close()`, I think it's necessary for
>>>>>>>>>> Table&SQL to
>>>>>>>>>> compile the generated code.
>>>>>>>>>> In Table&SQL, the watermark strategy and event-timestamp is defined
>>>>>>>> using
>>>>>>>>>> SQL expressions, we will
>>>>>>>>>> translate and generate Java code for the expressions. If we have
>>>>>>>>>> `open()/close()`, we don't need lazy initialization.
>>>>>>>>>> Besides that, I can see a need to report some metrics, e.g. the
>>>>>>> current
>>>>>>>>>> watermark, the dirty timestamps (null value), etc.
>>>>>>>>>> So I think a simple `open()/close()` with a context which can get
>>>>>>>>>> MetricGroup is nice and not complex for the first version.
>>>>>>>>>>
>>>>>>>>>> Best,
>>>>>>>>>> Jark
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org>
>>> wrote:
>>>>>>>>>>> Thanks, Aljoscha, for picking this up.
>>>>>>>>>>>
>>>>>>>>>>> I agree with the approach of doing the here proposed set of
>>> changes
>>>>>>> for
>>>>>>>>>>> now. It already makes things simpler and adds idleness support
>>>>>>>>>>> everywhere.
>>>>>>>>>>>
>>>>>>>>>>> Rich functions and state always add complexity, let's do this in a
>>>>>>> next
>>>>>>>>>>> step, if we have a really compelling case.
>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
>>>>>>> aljoscha@apache.org>
>>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The
>>>>> proposal
>>>>>>>> is
>>>>>>>>>>>> basically to turn emitting into a "flatMap", we give the
>>>>>>>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG
>>>>>>> can
>>>>>>>>>>>> decide whether to output a watermark or not and can also mark the
>>>>>>>>>>>> output
>>>>>>>>>>>> as idle. Changing the interface to return a Watermark (as the
>>>>>>> previous
>>>>>>>>>>>> watermark assigner interface did) would not allow that
>>> flexibility.
>>>>>>>>>>>> Regarding checkpointing the watermark and keeping track of the
>>>>>>> minimum
>>>>>>>>>>>> watermark, this would be the responsibility of the framework (or
>>>>> the
>>>>>>>>>>>> KafkaConsumer in the current implementation). The user-supplied
>>> WG
>>>>>>>> does
>>>>>>>>>>>> not need to make sure the watermark doesn't regress.
>>>>>>>>>>>>
>>>>>>>>>>>> Regarding making the WG a "rich function", I can see the
>>> potential
>>>>>>>>>>>> benefit but I also see a lot of pitfalls. For example, how should
>>>>>>> the
>>>>>>>>>>>> watermark state be handled in the case of scale-in? It could be
>>>>> made
>>>>>>>> to
>>>>>>>>>>>> work in the Kafka case by attaching the state to the partition
>>>>> state
>>>>>>>>>>>> that we keep, but then we have potential backwards compatibility
>>>>>>>>>>>> problems also for the WM state. Does the WG usually need to keep
>>>>> the
>>>>>>>>>>>> state or might it be enough if the state is transient, i.e. if
>>> you
>>>>>>>> have
>>>>>>>>>>>> a restart the WG would loose its histogram but it would rebuild
>>> it
>>>>>>>>>>>> quickly and you would get back to the same steady state as
>>> before.
>>>>>>>>>>>> Best,
>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>
>>>>>>>>>>>> On 27.04.20 12:12, David Anderson wrote:
>>>>>>>>>>>>> Overall I like this proposal; thanks for bringing it forward,
>>>>>>>>>>>>> Aljoscha.
>>>>>>>>>>>>>
>>>>>>>>>>>>> I also like the idea of making the Watermark generator a rich
>>>>>>>> function
>>>>>>>>>>> --
>>>>>>>>>>>>> this should make it more straightforward to implement smarter
>>>>>>>>>>>>> watermark
>>>>>>>>>>>>> generators. Eg, one that uses state to keep statistics about the
>>>>>>>>>>>>> actual
>>>>>>>>>>>>> out-of-orderness, and uses those statistics to implement a
>>>>> variable
>>>>>>>>>>>> delay.
>>>>>>>>>>>>> David
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
>>>>>>> kkloudas@gmail.com>
>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>> Hi Aljoscha,
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Thanks for opening the discussion!
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> I have two comments on the FLIP:
>>>>>>>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e.
>>> open()/
>>>>>>>>>>>>>> close(), probably with a Context as argument: I have not fully
>>>>>>>>>>>>>> thought
>>>>>>>>>>>>>> this through but I think that this is more aligned with the
>>> rest
>>>>>>> of
>>>>>>>>>>>>>> our rich functions. In addition, it will allow, for example, to
>>>>>>>>>>>>>> initialize the Watermark value, if we decide to checkpoint the
>>>>>>>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to
>>> do
>>>>>>>>>>>>>> anything in the open()).
>>>>>>>>>>>>>> 2) aligned with the above, and with the case where we want to
>>>>>>>>>>>>>> checkpoint the watermark in mind, I am wondering about how we
>>>>>>> could
>>>>>>>>>>>>>> implement this in the future. In the FLIP, it is proposed to
>>>>>>> expose
>>>>>>>>>>>>>> the WatermarkOutput in the methods of the WatermarkGenerator.
>>>>>>> Given
>>>>>>>>>>>>>> that there is the implicit contract that watermarks are
>>>>>>>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have
>>> (I
>>>>>>>>>>>>>> assume) a check that will compare the last emitted WM against
>>> the
>>>>>>>>>>>>>> provided one, and emit it only if it is >=. If not, then we
>>> risk
>>>>>>>>>>>>>> having the user shooting himself on the foot if he/she
>>>>>>> accidentally
>>>>>>>>>>>>>> forgets the check. Given that the WatermarkGenerator and its
>>>>>>>>>>>>>> caller do
>>>>>>>>>>>>>> not know if the watermark was finally emitted or not (the
>>>>>>>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
>>>>>>> responsible
>>>>>>>>>>>>>> for checkpointing the WM?
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Given this, why not having the methods as:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> public interface WatermarkGenerator<T> {
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
>>>>>>>> WatermarkOutput
>>>>>>>>>>>>>> output);
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
>>>>>>>>>>>>>> }
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> and the caller will be the one enforcing any invariants, such
>>> as
>>>>>>>>>>>>>> non-decreasing watermarks. In this way, the caller can
>>> checkpoint
>>>>>>>>>>>>>> anything that is needed as it will have complete knowledge as
>>> to
>>>>>>> if
>>>>>>>>>>>>>> the WM was emitted or not.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> What do you think?
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>>> Kostas
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <
>>> twalthr@apache.org
>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
>>>>>>>> unification.
>>>>>>>>>>> We
>>>>>>>>>>>>>>> have considered this FLIP already in the interfaces for
>>> FLIP-95
>>>>>>> [1]
>>>>>>>>>>> and
>>>>>>>>>>>>>>> look forward to update to the new unified watermark generators
>>>>>>> once
>>>>>>>>>>>>>>> FLIP-126 has been accepted.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Regards,
>>>>>>>>>>>>>>> Timo
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
>>>>>>>>>>>>>>>> Hi Everyone!
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
>>>>>>>>>>> separate)
>>>>>>>>>>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in
>>>>> an
>>>>>>>>>>>>>>>> experimental branch. I expanded on that work to provide a PoC
>>>>>>> for
>>>>>>>>>>> the
>>>>>>>>>>>>>>>> changes proposed in this FLIP: [2].
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Currently, we have two different flavours of Watermark
>>>>>>>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
>>>>>>>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
>>>>>>>>>>>>>>>> from TimestampAssigner. This means that sources that want to
>>>>>>>>>>>>>>>> support
>>>>>>>>>>>>>>>> watermark assignment/extraction in the source need to support
>>>>>>> two
>>>>>>>>>>>>>>>> separate interfaces, we have two operator implementations for
>>>>>>> the
>>>>>>>>>>>>>>>> different flavours. Also, this makes features such as generic
>>>>>>>>>>> support
>>>>>>>>>>>>>>>> for idleness detection more complicated to implemented
>>> because
>>>>>>> we
>>>>>>>>>>>> again
>>>>>>>>>>>>>>>> have to support two types of watermark assigners.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> In this FLIP we propose two things:
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Unify the Watermark Assigners into one Interface
>>>>>>>> WatermarkGenerator
>>>>>>>>>>>>>>>> Separate this new interface from the TimestampAssigner
>>>>>>>>>>>>>>>> The motivation for the first is to simplify future
>>>>>>> implementations
>>>>>>>>>>> and
>>>>>>>>>>>>>>>> code duplication. The motivation for the second point is
>>> again
>>>>>>>> code
>>>>>>>>>>>>>>>> deduplication, most assigners currently have to extend from
>>>>> some
>>>>>>>>>>> base
>>>>>>>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or
>>> users
>>>>>>>>>>>>>>>> have
>>>>>>>>>>>> to
>>>>>>>>>>>>>>>> override an abstract method of the watermark assigner to
>>>>> provide
>>>>>>>>>>>>>>>> the
>>>>>>>>>>>>>>>> timestamp extraction logic.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Additionally, we propose to add a generic wrapping
>>>>>>>>>>> WatermarkGenerator
>>>>>>>>>>>>>>>> that provides idleness detection, i.e. it can mark a
>>>>>>>>>>> stream/partition
>>>>>>>>>>>>>> as
>>>>>>>>>>>>>>>> idle if no data arrives after a configured timeout.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> The "unify and separate" part refers to the fact that we want
>>>>> to
>>>>>>>>>>> unify
>>>>>>>>>>>>>>>> punctuated and periodic assigners but at the same time split
>>>>> the
>>>>>>>>>>>>>>>> timestamp assigner from the watermark generator.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Please find more details in the FLIP [1]. Looking forward to
>>>>>>>>>>>>>>>> your feedback.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Best,
>>>>>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> [1]
>>>>>>>>>>>>>>>>
>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
>>>>>>>>>>>>>>>> [2]
>>> https://github.com/aljoscha/flink/tree/stephan-event-time
>>>>>
>>>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Dawid Wysakowicz <dw...@apache.org>.
I have similar thoughts to @Stephan
Ad. 1 I tried something like this on your branch:
/**
* Adds the given {@link TimestampAssigner} to this {@link
WatermarkStrategies}. For top-level classes that implement both
Serializable and TimestampAssigner
*/
public <TA extends TimestampAssigner<T> & Serializable>
WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner) {
checkNotNull(timestampAssigner, "timestampAssigner");
this.timestampAssigner = timestampAssigner;
return this;
}
@FunctionalInterface
public interface SerializableTimestampAssigner<T> extends
TimestampAssigner<T>, Serializable {
}
/**
* Adds the given {@link TimestampAssigner} to this {@link
WatermarkStrategies}.
* Helper method for serializable lambdas.
*/
public WatermarkStrategies<T>
withTimestampAssigner(SerializableTimestampAssigner<T> timestampAssigner) {
checkNotNull(timestampAssigner, "timestampAssigner");
this.timestampAssigner = timestampAssigner;
return this;
}
But I understand if that's too hacky. It's just a pity that we must
enforce limitations on an interface that are not strictly necessary.
Ad 2/3
I am aware the watermark assigner/timestamp extractor can be applied
further down the graph. Originally I also wanted to suggest
sourceTimestamp and SourceTimestampAssigner, but then I realized it can
be used also after the sources as you correctly pointed out. Even if the
TimestampAssigner is used after the source there might be some
native/record timestamp in the StreamRecord, that could've been
extracted by previous assigner.
Best,
Dawid
On 12/05/2020 16:47, Stephan Ewen wrote:
> @Aljoscha
>
> About (1) could we have an interface SerializableTimestampAssigner that
> simply mixes in the java.io.Serializable interface? Or will this be too
> clumsy?
>
> About (3) RecordTimeStamp seems to fit both cases (in-source-record
> timestamp, in stream-record timestamp).
>
> On Tue, May 12, 2020 at 4:12 PM Aljoscha Krettek <al...@apache.org>
> wrote:
>
>> Definitely +1 to point 2) raised by Dawid. I'm not sure on points 1) and
>> 3).
>>
>> 1) I can see the benefit of that but in reality most timestamp assigners
>> will probably need to be Serializable. If you look at my (updated) POC
>> branch [1] you can see how a TimestampAssigner would be specified on the
>> WatermarkStrategies helper class: [2]. The signature of this would have
>> to be changed to something like:
>>
>> public <TA extends TimestampAssigner<T> & Serializable>
>> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner)
>>
>> Then, however, it would not be possible for users to specify a lambda or
>> anonymous inner function for the TimestampAssigner like this:
>>
>> WatermarkStrategy<Long> testWmStrategy = WatermarkStrategies
>> .forGenerator(new PeriodicTestWatermarkGenerator())
>> .withTimestampAssigner((event, timestamp) -> event)
>> .build();
>>
>> 3) This makes sense if we only allow WatermarkStrategies on sources,
>> where the previous timestamp really is the "native" timestamp.
>> Currently, we also allow setting watermark strategies at arbitrary
>> points in the graph. I'm thinking we probably should only allow that in
>> sources but it's not the reality currently. I'm not against renaming it,
>> just voicing those thoughts.
>>
>> Best,
>> Aljoscha
>>
>>
>> [1] https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
>> [2]
>>
>> https://github.com/aljoscha/flink/blob/flink-xxx-wm-generators-rebased/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java#L81
>>
>> On 12.05.20 15:48, Stephan Ewen wrote:
>>> +1 to all of Dawid's suggestions, makes a lot of sense to me
>>>
>>> On Tue, May 12, 2020 at 2:32 PM Dawid Wysakowicz <dwysakowicz@apache.org
>>>
>>> wrote:
>>>
>>>> Hi Aljoscha,
>>>>
>>>> Sorry for adding comments during the vote, but I have some really minor
>>>> suggestions that should not influence the voting thread imo.
>>>>
>>>> 1) Does it make sense to have the TimestampAssigner extend from Flink's
>>>> Function? This implies it has to be serializable which with the factory
>>>> pattern is not strictly necessary, right? BTW I really like that you
>>>> suggested the FunctionInterface annotation there.
>>>>
>>>> 2) Could we rename the IdentityTimestampAssigner to e.g.
>>>>
>> RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
>>>> Personally I found the IdentityTimestampAssigner a bit misleading as it
>>>> usually mean a no-op. Which did not click for me, as I assumed it
>>>> somehow returns the incoming record itself.
>>>>
>>>> 3) Could we rename the second parameter of TimestampAssigner#extract to
>>>> e.g. recordTimestamp/nativeTimestamp? This is similar to the point
>>>> above. This parameter was also a bit confusing for me as I thought at
>>>> times its somehow related to
>>>> TimerService#currentProcessingTimestamp()/currentWatermark() as the
>>>> whole system currentTimestamp.
>>>>
>>>> Other than those three points I like the proposal and I was about to
>>>> vote +1 if it was not for those three points.
>>>>
>>>> Best,
>>>>
>>>> Dawid
>>>>
>>>> On 11/05/2020 16:57, Jark Wu wrote:
>>>>> Thanks for the explanation. I like the fatory pattern to make the
>> member
>>>>> variables immutable and final.
>>>>>
>>>>> So +1 to the proposal.
>>>>>
>>>>> Best,
>>>>> Jark
>>>>>
>>>>> On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org> wrote:
>>>>>
>>>>>> I am fine with that.
>>>>>>
>>>>>> Much of the principles seem agreed upon. I understand the need to
>>>> support
>>>>>> code-generated extractors and we should support most of it already (as
>>>>>> Aljoscha mentioned via the factories) can extend this if needed.
>>>>>>
>>>>>> I think that the factory approach supports code-generated extractors
>> in
>>>> a
>>>>>> cleaner way even than an extractor with an open/init method.
>>>>>>
>>>>>>
>>>>>> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek <aljoscha@apache.org
>>>>>> wrote:
>>>>>>
>>>>>>> We're slightly running out of time. I would propose we vote on the
>>>> basic
>>>>>>> principle and remain open to later additions. This feature is quite
>>>>>>> important to make the new Kafka Source that is developed as part of
>>>>>>> FLIP-27 useful. Otherwise we would have to use the legacy interfaces
>> in
>>>>>>> the newly added connector.
>>>>>>>
>>>>>>> I know that's a bit unorthodox but would everyone be OK with what's
>>>>>>> currently there and then we iterate?
>>>>>>>
>>>>>>> Best,
>>>>>>> Aljoscha
>>>>>>>
>>>>>>> On 11.05.20 13:57, Aljoscha Krettek wrote:
>>>>>>>> Ah, I meant to write this in my previous email, sorry about that.
>>>>>>>>
>>>>>>>> The WatermarkStrategy, which is basically a factory for a
>>>>>>>> WatermarkGenerator is the replacement for the open() method. This is
>>>>>> the
>>>>>>>> same strategy that was followed for StreamOperatorFactory, which was
>>>>>>>> introduced to allow code generation in the Table API [1]. If we need
>>>>>>>> metrics or other things we would add that as a parameter to the
>>>> factory
>>>>>>>> method. What do you think?
>>>>>>>>
>>>>>>>> Best,
>>>>>>>> Aljoscha
>>>>>>>>
>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
>>>>>>>>
>>>>>>>> On 10.05.20 05:07, Jark Wu wrote:
>>>>>>>>> Hi,
>>>>>>>>>
>>>>>>>>> Regarding to the `open()/close()`, I think it's necessary for
>>>>>>>>> Table&SQL to
>>>>>>>>> compile the generated code.
>>>>>>>>> In Table&SQL, the watermark strategy and event-timestamp is defined
>>>>>>> using
>>>>>>>>> SQL expressions, we will
>>>>>>>>> translate and generate Java code for the expressions. If we have
>>>>>>>>> `open()/close()`, we don't need lazy initialization.
>>>>>>>>> Besides that, I can see a need to report some metrics, e.g. the
>>>>>> current
>>>>>>>>> watermark, the dirty timestamps (null value), etc.
>>>>>>>>> So I think a simple `open()/close()` with a context which can get
>>>>>>>>> MetricGroup is nice and not complex for the first version.
>>>>>>>>>
>>>>>>>>> Best,
>>>>>>>>> Jark
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org>
>> wrote:
>>>>>>>>>> Thanks, Aljoscha, for picking this up.
>>>>>>>>>>
>>>>>>>>>> I agree with the approach of doing the here proposed set of
>> changes
>>>>>> for
>>>>>>>>>> now. It already makes things simpler and adds idleness support
>>>>>>>>>> everywhere.
>>>>>>>>>>
>>>>>>>>>> Rich functions and state always add complexity, let's do this in a
>>>>>> next
>>>>>>>>>> step, if we have a really compelling case.
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
>>>>>> aljoscha@apache.org>
>>>>>>>>>> wrote:
>>>>>>>>>>
>>>>>>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The
>>>> proposal
>>>>>>> is
>>>>>>>>>>> basically to turn emitting into a "flatMap", we give the
>>>>>>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG
>>>>>> can
>>>>>>>>>>> decide whether to output a watermark or not and can also mark the
>>>>>>>>>>> output
>>>>>>>>>>> as idle. Changing the interface to return a Watermark (as the
>>>>>> previous
>>>>>>>>>>> watermark assigner interface did) would not allow that
>> flexibility.
>>>>>>>>>>> Regarding checkpointing the watermark and keeping track of the
>>>>>> minimum
>>>>>>>>>>> watermark, this would be the responsibility of the framework (or
>>>> the
>>>>>>>>>>> KafkaConsumer in the current implementation). The user-supplied
>> WG
>>>>>>> does
>>>>>>>>>>> not need to make sure the watermark doesn't regress.
>>>>>>>>>>>
>>>>>>>>>>> Regarding making the WG a "rich function", I can see the
>> potential
>>>>>>>>>>> benefit but I also see a lot of pitfalls. For example, how should
>>>>>> the
>>>>>>>>>>> watermark state be handled in the case of scale-in? It could be
>>>> made
>>>>>>> to
>>>>>>>>>>> work in the Kafka case by attaching the state to the partition
>>>> state
>>>>>>>>>>> that we keep, but then we have potential backwards compatibility
>>>>>>>>>>> problems also for the WM state. Does the WG usually need to keep
>>>> the
>>>>>>>>>>> state or might it be enough if the state is transient, i.e. if
>> you
>>>>>>> have
>>>>>>>>>>> a restart the WG would loose its histogram but it would rebuild
>> it
>>>>>>>>>>> quickly and you would get back to the same steady state as
>> before.
>>>>>>>>>>> Best,
>>>>>>>>>>> Aljoscha
>>>>>>>>>>>
>>>>>>>>>>> On 27.04.20 12:12, David Anderson wrote:
>>>>>>>>>>>> Overall I like this proposal; thanks for bringing it forward,
>>>>>>>>>>>> Aljoscha.
>>>>>>>>>>>>
>>>>>>>>>>>> I also like the idea of making the Watermark generator a rich
>>>>>>> function
>>>>>>>>>> --
>>>>>>>>>>>> this should make it more straightforward to implement smarter
>>>>>>>>>>>> watermark
>>>>>>>>>>>> generators. Eg, one that uses state to keep statistics about the
>>>>>>>>>>>> actual
>>>>>>>>>>>> out-of-orderness, and uses those statistics to implement a
>>>> variable
>>>>>>>>>>> delay.
>>>>>>>>>>>> David
>>>>>>>>>>>>
>>>>>>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
>>>>>> kkloudas@gmail.com>
>>>>>>>>>>> wrote:
>>>>>>>>>>>>> Hi Aljoscha,
>>>>>>>>>>>>>
>>>>>>>>>>>>> Thanks for opening the discussion!
>>>>>>>>>>>>>
>>>>>>>>>>>>> I have two comments on the FLIP:
>>>>>>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e.
>> open()/
>>>>>>>>>>>>> close(), probably with a Context as argument: I have not fully
>>>>>>>>>>>>> thought
>>>>>>>>>>>>> this through but I think that this is more aligned with the
>> rest
>>>>>> of
>>>>>>>>>>>>> our rich functions. In addition, it will allow, for example, to
>>>>>>>>>>>>> initialize the Watermark value, if we decide to checkpoint the
>>>>>>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to
>> do
>>>>>>>>>>>>> anything in the open()).
>>>>>>>>>>>>> 2) aligned with the above, and with the case where we want to
>>>>>>>>>>>>> checkpoint the watermark in mind, I am wondering about how we
>>>>>> could
>>>>>>>>>>>>> implement this in the future. In the FLIP, it is proposed to
>>>>>> expose
>>>>>>>>>>>>> the WatermarkOutput in the methods of the WatermarkGenerator.
>>>>>> Given
>>>>>>>>>>>>> that there is the implicit contract that watermarks are
>>>>>>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have
>> (I
>>>>>>>>>>>>> assume) a check that will compare the last emitted WM against
>> the
>>>>>>>>>>>>> provided one, and emit it only if it is >=. If not, then we
>> risk
>>>>>>>>>>>>> having the user shooting himself on the foot if he/she
>>>>>> accidentally
>>>>>>>>>>>>> forgets the check. Given that the WatermarkGenerator and its
>>>>>>>>>>>>> caller do
>>>>>>>>>>>>> not know if the watermark was finally emitted or not (the
>>>>>>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
>>>>>> responsible
>>>>>>>>>>>>> for checkpointing the WM?
>>>>>>>>>>>>>
>>>>>>>>>>>>> Given this, why not having the methods as:
>>>>>>>>>>>>>
>>>>>>>>>>>>> public interface WatermarkGenerator<T> {
>>>>>>>>>>>>>
>>>>>>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
>>>>>>> WatermarkOutput
>>>>>>>>>>>>> output);
>>>>>>>>>>>>>
>>>>>>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
>>>>>>>>>>>>> }
>>>>>>>>>>>>>
>>>>>>>>>>>>> and the caller will be the one enforcing any invariants, such
>> as
>>>>>>>>>>>>> non-decreasing watermarks. In this way, the caller can
>> checkpoint
>>>>>>>>>>>>> anything that is needed as it will have complete knowledge as
>> to
>>>>>> if
>>>>>>>>>>>>> the WM was emitted or not.
>>>>>>>>>>>>>
>>>>>>>>>>>>> What do you think?
>>>>>>>>>>>>>
>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>> Kostas
>>>>>>>>>>>>>
>>>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <
>> twalthr@apache.org
>>>>>>>>>>> wrote:
>>>>>>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
>>>>>>> unification.
>>>>>>>>>> We
>>>>>>>>>>>>>> have considered this FLIP already in the interfaces for
>> FLIP-95
>>>>>> [1]
>>>>>>>>>> and
>>>>>>>>>>>>>> look forward to update to the new unified watermark generators
>>>>>> once
>>>>>>>>>>>>>> FLIP-126 has been accepted.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Regards,
>>>>>>>>>>>>>> Timo
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
>>>>>>>>>>>>>>> Hi Everyone!
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
>>>>>>>>>> separate)
>>>>>>>>>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in
>>>> an
>>>>>>>>>>>>>>> experimental branch. I expanded on that work to provide a PoC
>>>>>> for
>>>>>>>>>> the
>>>>>>>>>>>>>>> changes proposed in this FLIP: [2].
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Currently, we have two different flavours of Watermark
>>>>>>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
>>>>>>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
>>>>>>>>>>>>>>> from TimestampAssigner. This means that sources that want to
>>>>>>>>>>>>>>> support
>>>>>>>>>>>>>>> watermark assignment/extraction in the source need to support
>>>>>> two
>>>>>>>>>>>>>>> separate interfaces, we have two operator implementations for
>>>>>> the
>>>>>>>>>>>>>>> different flavours. Also, this makes features such as generic
>>>>>>>>>> support
>>>>>>>>>>>>>>> for idleness detection more complicated to implemented
>> because
>>>>>> we
>>>>>>>>>>> again
>>>>>>>>>>>>>>> have to support two types of watermark assigners.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> In this FLIP we propose two things:
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Unify the Watermark Assigners into one Interface
>>>>>>> WatermarkGenerator
>>>>>>>>>>>>>>> Separate this new interface from the TimestampAssigner
>>>>>>>>>>>>>>> The motivation for the first is to simplify future
>>>>>> implementations
>>>>>>>>>> and
>>>>>>>>>>>>>>> code duplication. The motivation for the second point is
>> again
>>>>>>> code
>>>>>>>>>>>>>>> deduplication, most assigners currently have to extend from
>>>> some
>>>>>>>>>> base
>>>>>>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or
>> users
>>>>>>>>>>>>>>> have
>>>>>>>>>>> to
>>>>>>>>>>>>>>> override an abstract method of the watermark assigner to
>>>> provide
>>>>>>>>>>>>>>> the
>>>>>>>>>>>>>>> timestamp extraction logic.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Additionally, we propose to add a generic wrapping
>>>>>>>>>> WatermarkGenerator
>>>>>>>>>>>>>>> that provides idleness detection, i.e. it can mark a
>>>>>>>>>> stream/partition
>>>>>>>>>>>>> as
>>>>>>>>>>>>>>> idle if no data arrives after a configured timeout.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> The "unify and separate" part refers to the fact that we want
>>>> to
>>>>>>>>>> unify
>>>>>>>>>>>>>>> punctuated and periodic assigners but at the same time split
>>>> the
>>>>>>>>>>>>>>> timestamp assigner from the watermark generator.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Please find more details in the FLIP [1]. Looking forward to
>>>>>>>>>>>>>>> your feedback.
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Best,
>>>>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> [1]
>>>>>>>>>>>>>>>
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
>>>>>>>>>>>>>>> [2]
>> https://github.com/aljoscha/flink/tree/stephan-event-time
>>>>
>>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Stephan Ewen <se...@apache.org>.
@Aljoscha
About (1) could we have an interface SerializableTimestampAssigner that
simply mixes in the java.io.Serializable interface? Or will this be too
clumsy?
About (3) RecordTimeStamp seems to fit both cases (in-source-record
timestamp, in stream-record timestamp).
On Tue, May 12, 2020 at 4:12 PM Aljoscha Krettek <al...@apache.org>
wrote:
> Definitely +1 to point 2) raised by Dawid. I'm not sure on points 1) and
> 3).
>
> 1) I can see the benefit of that but in reality most timestamp assigners
> will probably need to be Serializable. If you look at my (updated) POC
> branch [1] you can see how a TimestampAssigner would be specified on the
> WatermarkStrategies helper class: [2]. The signature of this would have
> to be changed to something like:
>
> public <TA extends TimestampAssigner<T> & Serializable>
> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner)
>
> Then, however, it would not be possible for users to specify a lambda or
> anonymous inner function for the TimestampAssigner like this:
>
> WatermarkStrategy<Long> testWmStrategy = WatermarkStrategies
> .forGenerator(new PeriodicTestWatermarkGenerator())
> .withTimestampAssigner((event, timestamp) -> event)
> .build();
>
> 3) This makes sense if we only allow WatermarkStrategies on sources,
> where the previous timestamp really is the "native" timestamp.
> Currently, we also allow setting watermark strategies at arbitrary
> points in the graph. I'm thinking we probably should only allow that in
> sources but it's not the reality currently. I'm not against renaming it,
> just voicing those thoughts.
>
> Best,
> Aljoscha
>
>
> [1] https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
> [2]
>
> https://github.com/aljoscha/flink/blob/flink-xxx-wm-generators-rebased/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java#L81
>
> On 12.05.20 15:48, Stephan Ewen wrote:
> > +1 to all of Dawid's suggestions, makes a lot of sense to me
> >
> > On Tue, May 12, 2020 at 2:32 PM Dawid Wysakowicz <dwysakowicz@apache.org
> >
> > wrote:
> >
> >> Hi Aljoscha,
> >>
> >> Sorry for adding comments during the vote, but I have some really minor
> >> suggestions that should not influence the voting thread imo.
> >>
> >> 1) Does it make sense to have the TimestampAssigner extend from Flink's
> >> Function? This implies it has to be serializable which with the factory
> >> pattern is not strictly necessary, right? BTW I really like that you
> >> suggested the FunctionInterface annotation there.
> >>
> >> 2) Could we rename the IdentityTimestampAssigner to e.g.
> >>
> RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
> >> Personally I found the IdentityTimestampAssigner a bit misleading as it
> >> usually mean a no-op. Which did not click for me, as I assumed it
> >> somehow returns the incoming record itself.
> >>
> >> 3) Could we rename the second parameter of TimestampAssigner#extract to
> >> e.g. recordTimestamp/nativeTimestamp? This is similar to the point
> >> above. This parameter was also a bit confusing for me as I thought at
> >> times its somehow related to
> >> TimerService#currentProcessingTimestamp()/currentWatermark() as the
> >> whole system currentTimestamp.
> >>
> >> Other than those three points I like the proposal and I was about to
> >> vote +1 if it was not for those three points.
> >>
> >> Best,
> >>
> >> Dawid
> >>
> >> On 11/05/2020 16:57, Jark Wu wrote:
> >>> Thanks for the explanation. I like the fatory pattern to make the
> member
> >>> variables immutable and final.
> >>>
> >>> So +1 to the proposal.
> >>>
> >>> Best,
> >>> Jark
> >>>
> >>> On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org> wrote:
> >>>
> >>>> I am fine with that.
> >>>>
> >>>> Much of the principles seem agreed upon. I understand the need to
> >> support
> >>>> code-generated extractors and we should support most of it already (as
> >>>> Aljoscha mentioned via the factories) can extend this if needed.
> >>>>
> >>>> I think that the factory approach supports code-generated extractors
> in
> >> a
> >>>> cleaner way even than an extractor with an open/init method.
> >>>>
> >>>>
> >>>> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek <aljoscha@apache.org
> >
> >>>> wrote:
> >>>>
> >>>>> We're slightly running out of time. I would propose we vote on the
> >> basic
> >>>>> principle and remain open to later additions. This feature is quite
> >>>>> important to make the new Kafka Source that is developed as part of
> >>>>> FLIP-27 useful. Otherwise we would have to use the legacy interfaces
> in
> >>>>> the newly added connector.
> >>>>>
> >>>>> I know that's a bit unorthodox but would everyone be OK with what's
> >>>>> currently there and then we iterate?
> >>>>>
> >>>>> Best,
> >>>>> Aljoscha
> >>>>>
> >>>>> On 11.05.20 13:57, Aljoscha Krettek wrote:
> >>>>>> Ah, I meant to write this in my previous email, sorry about that.
> >>>>>>
> >>>>>> The WatermarkStrategy, which is basically a factory for a
> >>>>>> WatermarkGenerator is the replacement for the open() method. This is
> >>>> the
> >>>>>> same strategy that was followed for StreamOperatorFactory, which was
> >>>>>> introduced to allow code generation in the Table API [1]. If we need
> >>>>>> metrics or other things we would add that as a parameter to the
> >> factory
> >>>>>> method. What do you think?
> >>>>>>
> >>>>>> Best,
> >>>>>> Aljoscha
> >>>>>>
> >>>>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
> >>>>>>
> >>>>>> On 10.05.20 05:07, Jark Wu wrote:
> >>>>>>> Hi,
> >>>>>>>
> >>>>>>> Regarding to the `open()/close()`, I think it's necessary for
> >>>>>>> Table&SQL to
> >>>>>>> compile the generated code.
> >>>>>>> In Table&SQL, the watermark strategy and event-timestamp is defined
> >>>>> using
> >>>>>>> SQL expressions, we will
> >>>>>>> translate and generate Java code for the expressions. If we have
> >>>>>>> `open()/close()`, we don't need lazy initialization.
> >>>>>>> Besides that, I can see a need to report some metrics, e.g. the
> >>>> current
> >>>>>>> watermark, the dirty timestamps (null value), etc.
> >>>>>>> So I think a simple `open()/close()` with a context which can get
> >>>>>>> MetricGroup is nice and not complex for the first version.
> >>>>>>>
> >>>>>>> Best,
> >>>>>>> Jark
> >>>>>>>
> >>>>>>>
> >>>>>>>
> >>>>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org>
> wrote:
> >>>>>>>
> >>>>>>>> Thanks, Aljoscha, for picking this up.
> >>>>>>>>
> >>>>>>>> I agree with the approach of doing the here proposed set of
> changes
> >>>> for
> >>>>>>>> now. It already makes things simpler and adds idleness support
> >>>>>>>> everywhere.
> >>>>>>>>
> >>>>>>>> Rich functions and state always add complexity, let's do this in a
> >>>> next
> >>>>>>>> step, if we have a really compelling case.
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
> >>>> aljoscha@apache.org>
> >>>>>>>> wrote:
> >>>>>>>>
> >>>>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The
> >> proposal
> >>>>> is
> >>>>>>>>> basically to turn emitting into a "flatMap", we give the
> >>>>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG
> >>>> can
> >>>>>>>>> decide whether to output a watermark or not and can also mark the
> >>>>>>>>> output
> >>>>>>>>> as idle. Changing the interface to return a Watermark (as the
> >>>> previous
> >>>>>>>>> watermark assigner interface did) would not allow that
> flexibility.
> >>>>>>>>>
> >>>>>>>>> Regarding checkpointing the watermark and keeping track of the
> >>>> minimum
> >>>>>>>>> watermark, this would be the responsibility of the framework (or
> >> the
> >>>>>>>>> KafkaConsumer in the current implementation). The user-supplied
> WG
> >>>>> does
> >>>>>>>>> not need to make sure the watermark doesn't regress.
> >>>>>>>>>
> >>>>>>>>> Regarding making the WG a "rich function", I can see the
> potential
> >>>>>>>>> benefit but I also see a lot of pitfalls. For example, how should
> >>>> the
> >>>>>>>>> watermark state be handled in the case of scale-in? It could be
> >> made
> >>>>> to
> >>>>>>>>> work in the Kafka case by attaching the state to the partition
> >> state
> >>>>>>>>> that we keep, but then we have potential backwards compatibility
> >>>>>>>>> problems also for the WM state. Does the WG usually need to keep
> >> the
> >>>>>>>>> state or might it be enough if the state is transient, i.e. if
> you
> >>>>> have
> >>>>>>>>> a restart the WG would loose its histogram but it would rebuild
> it
> >>>>>>>>> quickly and you would get back to the same steady state as
> before.
> >>>>>>>>>
> >>>>>>>>> Best,
> >>>>>>>>> Aljoscha
> >>>>>>>>>
> >>>>>>>>> On 27.04.20 12:12, David Anderson wrote:
> >>>>>>>>>> Overall I like this proposal; thanks for bringing it forward,
> >>>>>>>>>> Aljoscha.
> >>>>>>>>>>
> >>>>>>>>>> I also like the idea of making the Watermark generator a rich
> >>>>> function
> >>>>>>>> --
> >>>>>>>>>> this should make it more straightforward to implement smarter
> >>>>>>>>>> watermark
> >>>>>>>>>> generators. Eg, one that uses state to keep statistics about the
> >>>>>>>>>> actual
> >>>>>>>>>> out-of-orderness, and uses those statistics to implement a
> >> variable
> >>>>>>>>> delay.
> >>>>>>>>>> David
> >>>>>>>>>>
> >>>>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
> >>>> kkloudas@gmail.com>
> >>>>>>>>> wrote:
> >>>>>>>>>>> Hi Aljoscha,
> >>>>>>>>>>>
> >>>>>>>>>>> Thanks for opening the discussion!
> >>>>>>>>>>>
> >>>>>>>>>>> I have two comments on the FLIP:
> >>>>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e.
> open()/
> >>>>>>>>>>> close(), probably with a Context as argument: I have not fully
> >>>>>>>>>>> thought
> >>>>>>>>>>> this through but I think that this is more aligned with the
> rest
> >>>> of
> >>>>>>>>>>> our rich functions. In addition, it will allow, for example, to
> >>>>>>>>>>> initialize the Watermark value, if we decide to checkpoint the
> >>>>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to
> do
> >>>>>>>>>>> anything in the open()).
> >>>>>>>>>>> 2) aligned with the above, and with the case where we want to
> >>>>>>>>>>> checkpoint the watermark in mind, I am wondering about how we
> >>>> could
> >>>>>>>>>>> implement this in the future. In the FLIP, it is proposed to
> >>>> expose
> >>>>>>>>>>> the WatermarkOutput in the methods of the WatermarkGenerator.
> >>>> Given
> >>>>>>>>>>> that there is the implicit contract that watermarks are
> >>>>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have
> (I
> >>>>>>>>>>> assume) a check that will compare the last emitted WM against
> the
> >>>>>>>>>>> provided one, and emit it only if it is >=. If not, then we
> risk
> >>>>>>>>>>> having the user shooting himself on the foot if he/she
> >>>> accidentally
> >>>>>>>>>>> forgets the check. Given that the WatermarkGenerator and its
> >>>>>>>>>>> caller do
> >>>>>>>>>>> not know if the watermark was finally emitted or not (the
> >>>>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
> >>>> responsible
> >>>>>>>>>>> for checkpointing the WM?
> >>>>>>>>>>>
> >>>>>>>>>>> Given this, why not having the methods as:
> >>>>>>>>>>>
> >>>>>>>>>>> public interface WatermarkGenerator<T> {
> >>>>>>>>>>>
> >>>>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
> >>>>> WatermarkOutput
> >>>>>>>>>>> output);
> >>>>>>>>>>>
> >>>>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
> >>>>>>>>>>> }
> >>>>>>>>>>>
> >>>>>>>>>>> and the caller will be the one enforcing any invariants, such
> as
> >>>>>>>>>>> non-decreasing watermarks. In this way, the caller can
> checkpoint
> >>>>>>>>>>> anything that is needed as it will have complete knowledge as
> to
> >>>> if
> >>>>>>>>>>> the WM was emitted or not.
> >>>>>>>>>>>
> >>>>>>>>>>> What do you think?
> >>>>>>>>>>>
> >>>>>>>>>>> Cheers,
> >>>>>>>>>>> Kostas
> >>>>>>>>>>>
> >>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
> >>>>>>>>>>>
> >>>>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <
> twalthr@apache.org
> >>>
> >>>>>>>>> wrote:
> >>>>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
> >>>>> unification.
> >>>>>>>> We
> >>>>>>>>>>>> have considered this FLIP already in the interfaces for
> FLIP-95
> >>>> [1]
> >>>>>>>> and
> >>>>>>>>>>>> look forward to update to the new unified watermark generators
> >>>> once
> >>>>>>>>>>>> FLIP-126 has been accepted.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Regards,
> >>>>>>>>>>>> Timo
> >>>>>>>>>>>>
> >>>>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
> >>>>>>>>>>>>
> >>>>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
> >>>>>>>>>>>>> Hi Everyone!
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
> >>>>>>>> separate)
> >>>>>>>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in
> >> an
> >>>>>>>>>>>>> experimental branch. I expanded on that work to provide a PoC
> >>>> for
> >>>>>>>> the
> >>>>>>>>>>>>> changes proposed in this FLIP: [2].
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Currently, we have two different flavours of Watermark
> >>>>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
> >>>>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
> >>>>>>>>>>>>> from TimestampAssigner. This means that sources that want to
> >>>>>>>>>>>>> support
> >>>>>>>>>>>>> watermark assignment/extraction in the source need to support
> >>>> two
> >>>>>>>>>>>>> separate interfaces, we have two operator implementations for
> >>>> the
> >>>>>>>>>>>>> different flavours. Also, this makes features such as generic
> >>>>>>>> support
> >>>>>>>>>>>>> for idleness detection more complicated to implemented
> because
> >>>> we
> >>>>>>>>> again
> >>>>>>>>>>>>> have to support two types of watermark assigners.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> In this FLIP we propose two things:
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Unify the Watermark Assigners into one Interface
> >>>>> WatermarkGenerator
> >>>>>>>>>>>>> Separate this new interface from the TimestampAssigner
> >>>>>>>>>>>>> The motivation for the first is to simplify future
> >>>> implementations
> >>>>>>>> and
> >>>>>>>>>>>>> code duplication. The motivation for the second point is
> again
> >>>>> code
> >>>>>>>>>>>>> deduplication, most assigners currently have to extend from
> >> some
> >>>>>>>> base
> >>>>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or
> users
> >>>>>>>>>>>>> have
> >>>>>>>>> to
> >>>>>>>>>>>>> override an abstract method of the watermark assigner to
> >> provide
> >>>>>>>>>>>>> the
> >>>>>>>>>>>>> timestamp extraction logic.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Additionally, we propose to add a generic wrapping
> >>>>>>>> WatermarkGenerator
> >>>>>>>>>>>>> that provides idleness detection, i.e. it can mark a
> >>>>>>>> stream/partition
> >>>>>>>>>>> as
> >>>>>>>>>>>>> idle if no data arrives after a configured timeout.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> The "unify and separate" part refers to the fact that we want
> >> to
> >>>>>>>> unify
> >>>>>>>>>>>>> punctuated and periodic assigners but at the same time split
> >> the
> >>>>>>>>>>>>> timestamp assigner from the watermark generator.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Please find more details in the FLIP [1]. Looking forward to
> >>>>>>>>>>>>> your feedback.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> Best,
> >>>>>>>>>>>>> Aljoscha
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> [1]
> >>>>>>>>>>>>>
> >>>>
> >>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> [2]
> https://github.com/aljoscha/flink/tree/stephan-event-time
> >>>>>>>>>
> >>>>>
> >>
> >>
> >
>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Aljoscha Krettek <al...@apache.org>.
Definitely +1 to point 2) raised by Dawid. I'm not sure on points 1) and 3).
1) I can see the benefit of that but in reality most timestamp assigners
will probably need to be Serializable. If you look at my (updated) POC
branch [1] you can see how a TimestampAssigner would be specified on the
WatermarkStrategies helper class: [2]. The signature of this would have
to be changed to something like:
public <TA extends TimestampAssigner<T> & Serializable>
WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner)
Then, however, it would not be possible for users to specify a lambda or
anonymous inner function for the TimestampAssigner like this:
WatermarkStrategy<Long> testWmStrategy = WatermarkStrategies
.forGenerator(new PeriodicTestWatermarkGenerator())
.withTimestampAssigner((event, timestamp) -> event)
.build();
3) This makes sense if we only allow WatermarkStrategies on sources,
where the previous timestamp really is the "native" timestamp.
Currently, we also allow setting watermark strategies at arbitrary
points in the graph. I'm thinking we probably should only allow that in
sources but it's not the reality currently. I'm not against renaming it,
just voicing those thoughts.
Best,
Aljoscha
[1] https://github.com/aljoscha/flink/tree/flink-xxx-wm-generators-rebased
[2]
https://github.com/aljoscha/flink/blob/flink-xxx-wm-generators-rebased/flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java#L81
On 12.05.20 15:48, Stephan Ewen wrote:
> +1 to all of Dawid's suggestions, makes a lot of sense to me
>
> On Tue, May 12, 2020 at 2:32 PM Dawid Wysakowicz <dw...@apache.org>
> wrote:
>
>> Hi Aljoscha,
>>
>> Sorry for adding comments during the vote, but I have some really minor
>> suggestions that should not influence the voting thread imo.
>>
>> 1) Does it make sense to have the TimestampAssigner extend from Flink's
>> Function? This implies it has to be serializable which with the factory
>> pattern is not strictly necessary, right? BTW I really like that you
>> suggested the FunctionInterface annotation there.
>>
>> 2) Could we rename the IdentityTimestampAssigner to e.g.
>> RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
>> Personally I found the IdentityTimestampAssigner a bit misleading as it
>> usually mean a no-op. Which did not click for me, as I assumed it
>> somehow returns the incoming record itself.
>>
>> 3) Could we rename the second parameter of TimestampAssigner#extract to
>> e.g. recordTimestamp/nativeTimestamp? This is similar to the point
>> above. This parameter was also a bit confusing for me as I thought at
>> times its somehow related to
>> TimerService#currentProcessingTimestamp()/currentWatermark() as the
>> whole system currentTimestamp.
>>
>> Other than those three points I like the proposal and I was about to
>> vote +1 if it was not for those three points.
>>
>> Best,
>>
>> Dawid
>>
>> On 11/05/2020 16:57, Jark Wu wrote:
>>> Thanks for the explanation. I like the fatory pattern to make the member
>>> variables immutable and final.
>>>
>>> So +1 to the proposal.
>>>
>>> Best,
>>> Jark
>>>
>>> On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org> wrote:
>>>
>>>> I am fine with that.
>>>>
>>>> Much of the principles seem agreed upon. I understand the need to
>> support
>>>> code-generated extractors and we should support most of it already (as
>>>> Aljoscha mentioned via the factories) can extend this if needed.
>>>>
>>>> I think that the factory approach supports code-generated extractors in
>> a
>>>> cleaner way even than an extractor with an open/init method.
>>>>
>>>>
>>>> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek <al...@apache.org>
>>>> wrote:
>>>>
>>>>> We're slightly running out of time. I would propose we vote on the
>> basic
>>>>> principle and remain open to later additions. This feature is quite
>>>>> important to make the new Kafka Source that is developed as part of
>>>>> FLIP-27 useful. Otherwise we would have to use the legacy interfaces in
>>>>> the newly added connector.
>>>>>
>>>>> I know that's a bit unorthodox but would everyone be OK with what's
>>>>> currently there and then we iterate?
>>>>>
>>>>> Best,
>>>>> Aljoscha
>>>>>
>>>>> On 11.05.20 13:57, Aljoscha Krettek wrote:
>>>>>> Ah, I meant to write this in my previous email, sorry about that.
>>>>>>
>>>>>> The WatermarkStrategy, which is basically a factory for a
>>>>>> WatermarkGenerator is the replacement for the open() method. This is
>>>> the
>>>>>> same strategy that was followed for StreamOperatorFactory, which was
>>>>>> introduced to allow code generation in the Table API [1]. If we need
>>>>>> metrics or other things we would add that as a parameter to the
>> factory
>>>>>> method. What do you think?
>>>>>>
>>>>>> Best,
>>>>>> Aljoscha
>>>>>>
>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
>>>>>>
>>>>>> On 10.05.20 05:07, Jark Wu wrote:
>>>>>>> Hi,
>>>>>>>
>>>>>>> Regarding to the `open()/close()`, I think it's necessary for
>>>>>>> Table&SQL to
>>>>>>> compile the generated code.
>>>>>>> In Table&SQL, the watermark strategy and event-timestamp is defined
>>>>> using
>>>>>>> SQL expressions, we will
>>>>>>> translate and generate Java code for the expressions. If we have
>>>>>>> `open()/close()`, we don't need lazy initialization.
>>>>>>> Besides that, I can see a need to report some metrics, e.g. the
>>>> current
>>>>>>> watermark, the dirty timestamps (null value), etc.
>>>>>>> So I think a simple `open()/close()` with a context which can get
>>>>>>> MetricGroup is nice and not complex for the first version.
>>>>>>>
>>>>>>> Best,
>>>>>>> Jark
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org> wrote:
>>>>>>>
>>>>>>>> Thanks, Aljoscha, for picking this up.
>>>>>>>>
>>>>>>>> I agree with the approach of doing the here proposed set of changes
>>>> for
>>>>>>>> now. It already makes things simpler and adds idleness support
>>>>>>>> everywhere.
>>>>>>>>
>>>>>>>> Rich functions and state always add complexity, let's do this in a
>>>> next
>>>>>>>> step, if we have a really compelling case.
>>>>>>>>
>>>>>>>>
>>>>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
>>>> aljoscha@apache.org>
>>>>>>>> wrote:
>>>>>>>>
>>>>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The
>> proposal
>>>>> is
>>>>>>>>> basically to turn emitting into a "flatMap", we give the
>>>>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG
>>>> can
>>>>>>>>> decide whether to output a watermark or not and can also mark the
>>>>>>>>> output
>>>>>>>>> as idle. Changing the interface to return a Watermark (as the
>>>> previous
>>>>>>>>> watermark assigner interface did) would not allow that flexibility.
>>>>>>>>>
>>>>>>>>> Regarding checkpointing the watermark and keeping track of the
>>>> minimum
>>>>>>>>> watermark, this would be the responsibility of the framework (or
>> the
>>>>>>>>> KafkaConsumer in the current implementation). The user-supplied WG
>>>>> does
>>>>>>>>> not need to make sure the watermark doesn't regress.
>>>>>>>>>
>>>>>>>>> Regarding making the WG a "rich function", I can see the potential
>>>>>>>>> benefit but I also see a lot of pitfalls. For example, how should
>>>> the
>>>>>>>>> watermark state be handled in the case of scale-in? It could be
>> made
>>>>> to
>>>>>>>>> work in the Kafka case by attaching the state to the partition
>> state
>>>>>>>>> that we keep, but then we have potential backwards compatibility
>>>>>>>>> problems also for the WM state. Does the WG usually need to keep
>> the
>>>>>>>>> state or might it be enough if the state is transient, i.e. if you
>>>>> have
>>>>>>>>> a restart the WG would loose its histogram but it would rebuild it
>>>>>>>>> quickly and you would get back to the same steady state as before.
>>>>>>>>>
>>>>>>>>> Best,
>>>>>>>>> Aljoscha
>>>>>>>>>
>>>>>>>>> On 27.04.20 12:12, David Anderson wrote:
>>>>>>>>>> Overall I like this proposal; thanks for bringing it forward,
>>>>>>>>>> Aljoscha.
>>>>>>>>>>
>>>>>>>>>> I also like the idea of making the Watermark generator a rich
>>>>> function
>>>>>>>> --
>>>>>>>>>> this should make it more straightforward to implement smarter
>>>>>>>>>> watermark
>>>>>>>>>> generators. Eg, one that uses state to keep statistics about the
>>>>>>>>>> actual
>>>>>>>>>> out-of-orderness, and uses those statistics to implement a
>> variable
>>>>>>>>> delay.
>>>>>>>>>> David
>>>>>>>>>>
>>>>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
>>>> kkloudas@gmail.com>
>>>>>>>>> wrote:
>>>>>>>>>>> Hi Aljoscha,
>>>>>>>>>>>
>>>>>>>>>>> Thanks for opening the discussion!
>>>>>>>>>>>
>>>>>>>>>>> I have two comments on the FLIP:
>>>>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e. open()/
>>>>>>>>>>> close(), probably with a Context as argument: I have not fully
>>>>>>>>>>> thought
>>>>>>>>>>> this through but I think that this is more aligned with the rest
>>>> of
>>>>>>>>>>> our rich functions. In addition, it will allow, for example, to
>>>>>>>>>>> initialize the Watermark value, if we decide to checkpoint the
>>>>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to do
>>>>>>>>>>> anything in the open()).
>>>>>>>>>>> 2) aligned with the above, and with the case where we want to
>>>>>>>>>>> checkpoint the watermark in mind, I am wondering about how we
>>>> could
>>>>>>>>>>> implement this in the future. In the FLIP, it is proposed to
>>>> expose
>>>>>>>>>>> the WatermarkOutput in the methods of the WatermarkGenerator.
>>>> Given
>>>>>>>>>>> that there is the implicit contract that watermarks are
>>>>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
>>>>>>>>>>> assume) a check that will compare the last emitted WM against the
>>>>>>>>>>> provided one, and emit it only if it is >=. If not, then we risk
>>>>>>>>>>> having the user shooting himself on the foot if he/she
>>>> accidentally
>>>>>>>>>>> forgets the check. Given that the WatermarkGenerator and its
>>>>>>>>>>> caller do
>>>>>>>>>>> not know if the watermark was finally emitted or not (the
>>>>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
>>>> responsible
>>>>>>>>>>> for checkpointing the WM?
>>>>>>>>>>>
>>>>>>>>>>> Given this, why not having the methods as:
>>>>>>>>>>>
>>>>>>>>>>> public interface WatermarkGenerator<T> {
>>>>>>>>>>>
>>>>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
>>>>> WatermarkOutput
>>>>>>>>>>> output);
>>>>>>>>>>>
>>>>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
>>>>>>>>>>> }
>>>>>>>>>>>
>>>>>>>>>>> and the caller will be the one enforcing any invariants, such as
>>>>>>>>>>> non-decreasing watermarks. In this way, the caller can checkpoint
>>>>>>>>>>> anything that is needed as it will have complete knowledge as to
>>>> if
>>>>>>>>>>> the WM was emitted or not.
>>>>>>>>>>>
>>>>>>>>>>> What do you think?
>>>>>>>>>>>
>>>>>>>>>>> Cheers,
>>>>>>>>>>> Kostas
>>>>>>>>>>>
>>>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
>>>>>>>>>>>
>>>>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <twalthr@apache.org
>>>
>>>>>>>>> wrote:
>>>>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
>>>>> unification.
>>>>>>>> We
>>>>>>>>>>>> have considered this FLIP already in the interfaces for FLIP-95
>>>> [1]
>>>>>>>> and
>>>>>>>>>>>> look forward to update to the new unified watermark generators
>>>> once
>>>>>>>>>>>> FLIP-126 has been accepted.
>>>>>>>>>>>>
>>>>>>>>>>>> Regards,
>>>>>>>>>>>> Timo
>>>>>>>>>>>>
>>>>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
>>>>>>>>>>>>
>>>>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
>>>>>>>>>>>>> Hi Everyone!
>>>>>>>>>>>>>
>>>>>>>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
>>>>>>>> separate)
>>>>>>>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in
>> an
>>>>>>>>>>>>> experimental branch. I expanded on that work to provide a PoC
>>>> for
>>>>>>>> the
>>>>>>>>>>>>> changes proposed in this FLIP: [2].
>>>>>>>>>>>>>
>>>>>>>>>>>>> Currently, we have two different flavours of Watermark
>>>>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
>>>>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
>>>>>>>>>>>>> from TimestampAssigner. This means that sources that want to
>>>>>>>>>>>>> support
>>>>>>>>>>>>> watermark assignment/extraction in the source need to support
>>>> two
>>>>>>>>>>>>> separate interfaces, we have two operator implementations for
>>>> the
>>>>>>>>>>>>> different flavours. Also, this makes features such as generic
>>>>>>>> support
>>>>>>>>>>>>> for idleness detection more complicated to implemented because
>>>> we
>>>>>>>>> again
>>>>>>>>>>>>> have to support two types of watermark assigners.
>>>>>>>>>>>>>
>>>>>>>>>>>>> In this FLIP we propose two things:
>>>>>>>>>>>>>
>>>>>>>>>>>>> Unify the Watermark Assigners into one Interface
>>>>> WatermarkGenerator
>>>>>>>>>>>>> Separate this new interface from the TimestampAssigner
>>>>>>>>>>>>> The motivation for the first is to simplify future
>>>> implementations
>>>>>>>> and
>>>>>>>>>>>>> code duplication. The motivation for the second point is again
>>>>> code
>>>>>>>>>>>>> deduplication, most assigners currently have to extend from
>> some
>>>>>>>> base
>>>>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or users
>>>>>>>>>>>>> have
>>>>>>>>> to
>>>>>>>>>>>>> override an abstract method of the watermark assigner to
>> provide
>>>>>>>>>>>>> the
>>>>>>>>>>>>> timestamp extraction logic.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Additionally, we propose to add a generic wrapping
>>>>>>>> WatermarkGenerator
>>>>>>>>>>>>> that provides idleness detection, i.e. it can mark a
>>>>>>>> stream/partition
>>>>>>>>>>> as
>>>>>>>>>>>>> idle if no data arrives after a configured timeout.
>>>>>>>>>>>>>
>>>>>>>>>>>>> The "unify and separate" part refers to the fact that we want
>> to
>>>>>>>> unify
>>>>>>>>>>>>> punctuated and periodic assigners but at the same time split
>> the
>>>>>>>>>>>>> timestamp assigner from the watermark generator.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Please find more details in the FLIP [1]. Looking forward to
>>>>>>>>>>>>> your feedback.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Best,
>>>>>>>>>>>>> Aljoscha
>>>>>>>>>>>>>
>>>>>>>>>>>>> [1]
>>>>>>>>>>>>>
>>>>
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
>>>>>>>>>>>>>
>>>>>>>>>>>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
>>>>>>>>>
>>>>>
>>
>>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Stephan Ewen <se...@apache.org>.
+1 to all of Dawid's suggestions, makes a lot of sense to me
On Tue, May 12, 2020 at 2:32 PM Dawid Wysakowicz <dw...@apache.org>
wrote:
> Hi Aljoscha,
>
> Sorry for adding comments during the vote, but I have some really minor
> suggestions that should not influence the voting thread imo.
>
> 1) Does it make sense to have the TimestampAssigner extend from Flink's
> Function? This implies it has to be serializable which with the factory
> pattern is not strictly necessary, right? BTW I really like that you
> suggested the FunctionInterface annotation there.
>
> 2) Could we rename the IdentityTimestampAssigner to e.g.
> RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
> Personally I found the IdentityTimestampAssigner a bit misleading as it
> usually mean a no-op. Which did not click for me, as I assumed it
> somehow returns the incoming record itself.
>
> 3) Could we rename the second parameter of TimestampAssigner#extract to
> e.g. recordTimestamp/nativeTimestamp? This is similar to the point
> above. This parameter was also a bit confusing for me as I thought at
> times its somehow related to
> TimerService#currentProcessingTimestamp()/currentWatermark() as the
> whole system currentTimestamp.
>
> Other than those three points I like the proposal and I was about to
> vote +1 if it was not for those three points.
>
> Best,
>
> Dawid
>
> On 11/05/2020 16:57, Jark Wu wrote:
> > Thanks for the explanation. I like the fatory pattern to make the member
> > variables immutable and final.
> >
> > So +1 to the proposal.
> >
> > Best,
> > Jark
> >
> > On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org> wrote:
> >
> >> I am fine with that.
> >>
> >> Much of the principles seem agreed upon. I understand the need to
> support
> >> code-generated extractors and we should support most of it already (as
> >> Aljoscha mentioned via the factories) can extend this if needed.
> >>
> >> I think that the factory approach supports code-generated extractors in
> a
> >> cleaner way even than an extractor with an open/init method.
> >>
> >>
> >> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek <al...@apache.org>
> >> wrote:
> >>
> >>> We're slightly running out of time. I would propose we vote on the
> basic
> >>> principle and remain open to later additions. This feature is quite
> >>> important to make the new Kafka Source that is developed as part of
> >>> FLIP-27 useful. Otherwise we would have to use the legacy interfaces in
> >>> the newly added connector.
> >>>
> >>> I know that's a bit unorthodox but would everyone be OK with what's
> >>> currently there and then we iterate?
> >>>
> >>> Best,
> >>> Aljoscha
> >>>
> >>> On 11.05.20 13:57, Aljoscha Krettek wrote:
> >>>> Ah, I meant to write this in my previous email, sorry about that.
> >>>>
> >>>> The WatermarkStrategy, which is basically a factory for a
> >>>> WatermarkGenerator is the replacement for the open() method. This is
> >> the
> >>>> same strategy that was followed for StreamOperatorFactory, which was
> >>>> introduced to allow code generation in the Table API [1]. If we need
> >>>> metrics or other things we would add that as a parameter to the
> factory
> >>>> method. What do you think?
> >>>>
> >>>> Best,
> >>>> Aljoscha
> >>>>
> >>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
> >>>>
> >>>> On 10.05.20 05:07, Jark Wu wrote:
> >>>>> Hi,
> >>>>>
> >>>>> Regarding to the `open()/close()`, I think it's necessary for
> >>>>> Table&SQL to
> >>>>> compile the generated code.
> >>>>> In Table&SQL, the watermark strategy and event-timestamp is defined
> >>> using
> >>>>> SQL expressions, we will
> >>>>> translate and generate Java code for the expressions. If we have
> >>>>> `open()/close()`, we don't need lazy initialization.
> >>>>> Besides that, I can see a need to report some metrics, e.g. the
> >> current
> >>>>> watermark, the dirty timestamps (null value), etc.
> >>>>> So I think a simple `open()/close()` with a context which can get
> >>>>> MetricGroup is nice and not complex for the first version.
> >>>>>
> >>>>> Best,
> >>>>> Jark
> >>>>>
> >>>>>
> >>>>>
> >>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org> wrote:
> >>>>>
> >>>>>> Thanks, Aljoscha, for picking this up.
> >>>>>>
> >>>>>> I agree with the approach of doing the here proposed set of changes
> >> for
> >>>>>> now. It already makes things simpler and adds idleness support
> >>>>>> everywhere.
> >>>>>>
> >>>>>> Rich functions and state always add complexity, let's do this in a
> >> next
> >>>>>> step, if we have a really compelling case.
> >>>>>>
> >>>>>>
> >>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
> >> aljoscha@apache.org>
> >>>>>> wrote:
> >>>>>>
> >>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The
> proposal
> >>> is
> >>>>>>> basically to turn emitting into a "flatMap", we give the
> >>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG
> >> can
> >>>>>>> decide whether to output a watermark or not and can also mark the
> >>>>>>> output
> >>>>>>> as idle. Changing the interface to return a Watermark (as the
> >> previous
> >>>>>>> watermark assigner interface did) would not allow that flexibility.
> >>>>>>>
> >>>>>>> Regarding checkpointing the watermark and keeping track of the
> >> minimum
> >>>>>>> watermark, this would be the responsibility of the framework (or
> the
> >>>>>>> KafkaConsumer in the current implementation). The user-supplied WG
> >>> does
> >>>>>>> not need to make sure the watermark doesn't regress.
> >>>>>>>
> >>>>>>> Regarding making the WG a "rich function", I can see the potential
> >>>>>>> benefit but I also see a lot of pitfalls. For example, how should
> >> the
> >>>>>>> watermark state be handled in the case of scale-in? It could be
> made
> >>> to
> >>>>>>> work in the Kafka case by attaching the state to the partition
> state
> >>>>>>> that we keep, but then we have potential backwards compatibility
> >>>>>>> problems also for the WM state. Does the WG usually need to keep
> the
> >>>>>>> state or might it be enough if the state is transient, i.e. if you
> >>> have
> >>>>>>> a restart the WG would loose its histogram but it would rebuild it
> >>>>>>> quickly and you would get back to the same steady state as before.
> >>>>>>>
> >>>>>>> Best,
> >>>>>>> Aljoscha
> >>>>>>>
> >>>>>>> On 27.04.20 12:12, David Anderson wrote:
> >>>>>>>> Overall I like this proposal; thanks for bringing it forward,
> >>>>>>>> Aljoscha.
> >>>>>>>>
> >>>>>>>> I also like the idea of making the Watermark generator a rich
> >>> function
> >>>>>> --
> >>>>>>>> this should make it more straightforward to implement smarter
> >>>>>>>> watermark
> >>>>>>>> generators. Eg, one that uses state to keep statistics about the
> >>>>>>>> actual
> >>>>>>>> out-of-orderness, and uses those statistics to implement a
> variable
> >>>>>>> delay.
> >>>>>>>> David
> >>>>>>>>
> >>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
> >> kkloudas@gmail.com>
> >>>>>>> wrote:
> >>>>>>>>> Hi Aljoscha,
> >>>>>>>>>
> >>>>>>>>> Thanks for opening the discussion!
> >>>>>>>>>
> >>>>>>>>> I have two comments on the FLIP:
> >>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e. open()/
> >>>>>>>>> close(), probably with a Context as argument: I have not fully
> >>>>>>>>> thought
> >>>>>>>>> this through but I think that this is more aligned with the rest
> >> of
> >>>>>>>>> our rich functions. In addition, it will allow, for example, to
> >>>>>>>>> initialize the Watermark value, if we decide to checkpoint the
> >>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to do
> >>>>>>>>> anything in the open()).
> >>>>>>>>> 2) aligned with the above, and with the case where we want to
> >>>>>>>>> checkpoint the watermark in mind, I am wondering about how we
> >> could
> >>>>>>>>> implement this in the future. In the FLIP, it is proposed to
> >> expose
> >>>>>>>>> the WatermarkOutput in the methods of the WatermarkGenerator.
> >> Given
> >>>>>>>>> that there is the implicit contract that watermarks are
> >>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
> >>>>>>>>> assume) a check that will compare the last emitted WM against the
> >>>>>>>>> provided one, and emit it only if it is >=. If not, then we risk
> >>>>>>>>> having the user shooting himself on the foot if he/she
> >> accidentally
> >>>>>>>>> forgets the check. Given that the WatermarkGenerator and its
> >>>>>>>>> caller do
> >>>>>>>>> not know if the watermark was finally emitted or not (the
> >>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
> >> responsible
> >>>>>>>>> for checkpointing the WM?
> >>>>>>>>>
> >>>>>>>>> Given this, why not having the methods as:
> >>>>>>>>>
> >>>>>>>>> public interface WatermarkGenerator<T> {
> >>>>>>>>>
> >>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
> >>> WatermarkOutput
> >>>>>>>>> output);
> >>>>>>>>>
> >>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
> >>>>>>>>> }
> >>>>>>>>>
> >>>>>>>>> and the caller will be the one enforcing any invariants, such as
> >>>>>>>>> non-decreasing watermarks. In this way, the caller can checkpoint
> >>>>>>>>> anything that is needed as it will have complete knowledge as to
> >> if
> >>>>>>>>> the WM was emitted or not.
> >>>>>>>>>
> >>>>>>>>> What do you think?
> >>>>>>>>>
> >>>>>>>>> Cheers,
> >>>>>>>>> Kostas
> >>>>>>>>>
> >>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
> >>>>>>>>>
> >>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <twalthr@apache.org
> >
> >>>>>>> wrote:
> >>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
> >>> unification.
> >>>>>> We
> >>>>>>>>>> have considered this FLIP already in the interfaces for FLIP-95
> >> [1]
> >>>>>> and
> >>>>>>>>>> look forward to update to the new unified watermark generators
> >> once
> >>>>>>>>>> FLIP-126 has been accepted.
> >>>>>>>>>>
> >>>>>>>>>> Regards,
> >>>>>>>>>> Timo
> >>>>>>>>>>
> >>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
> >>>>>>>>>>
> >>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
> >>>>>>>>>>> Hi Everyone!
> >>>>>>>>>>>
> >>>>>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
> >>>>>> separate)
> >>>>>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in
> an
> >>>>>>>>>>> experimental branch. I expanded on that work to provide a PoC
> >> for
> >>>>>> the
> >>>>>>>>>>> changes proposed in this FLIP: [2].
> >>>>>>>>>>>
> >>>>>>>>>>> Currently, we have two different flavours of Watermark
> >>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
> >>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
> >>>>>>>>>>> from TimestampAssigner. This means that sources that want to
> >>>>>>>>>>> support
> >>>>>>>>>>> watermark assignment/extraction in the source need to support
> >> two
> >>>>>>>>>>> separate interfaces, we have two operator implementations for
> >> the
> >>>>>>>>>>> different flavours. Also, this makes features such as generic
> >>>>>> support
> >>>>>>>>>>> for idleness detection more complicated to implemented because
> >> we
> >>>>>>> again
> >>>>>>>>>>> have to support two types of watermark assigners.
> >>>>>>>>>>>
> >>>>>>>>>>> In this FLIP we propose two things:
> >>>>>>>>>>>
> >>>>>>>>>>> Unify the Watermark Assigners into one Interface
> >>> WatermarkGenerator
> >>>>>>>>>>> Separate this new interface from the TimestampAssigner
> >>>>>>>>>>> The motivation for the first is to simplify future
> >> implementations
> >>>>>> and
> >>>>>>>>>>> code duplication. The motivation for the second point is again
> >>> code
> >>>>>>>>>>> deduplication, most assigners currently have to extend from
> some
> >>>>>> base
> >>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or users
> >>>>>>>>>>> have
> >>>>>>> to
> >>>>>>>>>>> override an abstract method of the watermark assigner to
> provide
> >>>>>>>>>>> the
> >>>>>>>>>>> timestamp extraction logic.
> >>>>>>>>>>>
> >>>>>>>>>>> Additionally, we propose to add a generic wrapping
> >>>>>> WatermarkGenerator
> >>>>>>>>>>> that provides idleness detection, i.e. it can mark a
> >>>>>> stream/partition
> >>>>>>>>> as
> >>>>>>>>>>> idle if no data arrives after a configured timeout.
> >>>>>>>>>>>
> >>>>>>>>>>> The "unify and separate" part refers to the fact that we want
> to
> >>>>>> unify
> >>>>>>>>>>> punctuated and periodic assigners but at the same time split
> the
> >>>>>>>>>>> timestamp assigner from the watermark generator.
> >>>>>>>>>>>
> >>>>>>>>>>> Please find more details in the FLIP [1]. Looking forward to
> >>>>>>>>>>> your feedback.
> >>>>>>>>>>>
> >>>>>>>>>>> Best,
> >>>>>>>>>>> Aljoscha
> >>>>>>>>>>>
> >>>>>>>>>>> [1]
> >>>>>>>>>>>
> >>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
> >>>>>>>>>>>
> >>>>>>>>>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
> >>>>>>>
> >>>
>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Dawid Wysakowicz <dw...@apache.org>.
Hi Aljoscha,
Sorry for adding comments during the vote, but I have some really minor
suggestions that should not influence the voting thread imo.
1) Does it make sense to have the TimestampAssigner extend from Flink's
Function? This implies it has to be serializable which with the factory
pattern is not strictly necessary, right? BTW I really like that you
suggested the FunctionInterface annotation there.
2) Could we rename the IdentityTimestampAssigner to e.g.
RecordTimestampAssigner/SystemTimestampAssigner/NativeTimestampAssigner...
Personally I found the IdentityTimestampAssigner a bit misleading as it
usually mean a no-op. Which did not click for me, as I assumed it
somehow returns the incoming record itself.
3) Could we rename the second parameter of TimestampAssigner#extract to
e.g. recordTimestamp/nativeTimestamp? This is similar to the point
above. This parameter was also a bit confusing for me as I thought at
times its somehow related to
TimerService#currentProcessingTimestamp()/currentWatermark() as the
whole system currentTimestamp.
Other than those three points I like the proposal and I was about to
vote +1 if it was not for those three points.
Best,
Dawid
On 11/05/2020 16:57, Jark Wu wrote:
> Thanks for the explanation. I like the fatory pattern to make the member
> variables immutable and final.
>
> So +1 to the proposal.
>
> Best,
> Jark
>
> On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org> wrote:
>
>> I am fine with that.
>>
>> Much of the principles seem agreed upon. I understand the need to support
>> code-generated extractors and we should support most of it already (as
>> Aljoscha mentioned via the factories) can extend this if needed.
>>
>> I think that the factory approach supports code-generated extractors in a
>> cleaner way even than an extractor with an open/init method.
>>
>>
>> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek <al...@apache.org>
>> wrote:
>>
>>> We're slightly running out of time. I would propose we vote on the basic
>>> principle and remain open to later additions. This feature is quite
>>> important to make the new Kafka Source that is developed as part of
>>> FLIP-27 useful. Otherwise we would have to use the legacy interfaces in
>>> the newly added connector.
>>>
>>> I know that's a bit unorthodox but would everyone be OK with what's
>>> currently there and then we iterate?
>>>
>>> Best,
>>> Aljoscha
>>>
>>> On 11.05.20 13:57, Aljoscha Krettek wrote:
>>>> Ah, I meant to write this in my previous email, sorry about that.
>>>>
>>>> The WatermarkStrategy, which is basically a factory for a
>>>> WatermarkGenerator is the replacement for the open() method. This is
>> the
>>>> same strategy that was followed for StreamOperatorFactory, which was
>>>> introduced to allow code generation in the Table API [1]. If we need
>>>> metrics or other things we would add that as a parameter to the factory
>>>> method. What do you think?
>>>>
>>>> Best,
>>>> Aljoscha
>>>>
>>>> [1] https://issues.apache.org/jira/browse/FLINK-11974
>>>>
>>>> On 10.05.20 05:07, Jark Wu wrote:
>>>>> Hi,
>>>>>
>>>>> Regarding to the `open()/close()`, I think it's necessary for
>>>>> Table&SQL to
>>>>> compile the generated code.
>>>>> In Table&SQL, the watermark strategy and event-timestamp is defined
>>> using
>>>>> SQL expressions, we will
>>>>> translate and generate Java code for the expressions. If we have
>>>>> `open()/close()`, we don't need lazy initialization.
>>>>> Besides that, I can see a need to report some metrics, e.g. the
>> current
>>>>> watermark, the dirty timestamps (null value), etc.
>>>>> So I think a simple `open()/close()` with a context which can get
>>>>> MetricGroup is nice and not complex for the first version.
>>>>>
>>>>> Best,
>>>>> Jark
>>>>>
>>>>>
>>>>>
>>>>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org> wrote:
>>>>>
>>>>>> Thanks, Aljoscha, for picking this up.
>>>>>>
>>>>>> I agree with the approach of doing the here proposed set of changes
>> for
>>>>>> now. It already makes things simpler and adds idleness support
>>>>>> everywhere.
>>>>>>
>>>>>> Rich functions and state always add complexity, let's do this in a
>> next
>>>>>> step, if we have a really compelling case.
>>>>>>
>>>>>>
>>>>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
>> aljoscha@apache.org>
>>>>>> wrote:
>>>>>>
>>>>>>> Regarding the WatermarkGenerator (WG) interface itself. The proposal
>>> is
>>>>>>> basically to turn emitting into a "flatMap", we give the
>>>>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG
>> can
>>>>>>> decide whether to output a watermark or not and can also mark the
>>>>>>> output
>>>>>>> as idle. Changing the interface to return a Watermark (as the
>> previous
>>>>>>> watermark assigner interface did) would not allow that flexibility.
>>>>>>>
>>>>>>> Regarding checkpointing the watermark and keeping track of the
>> minimum
>>>>>>> watermark, this would be the responsibility of the framework (or the
>>>>>>> KafkaConsumer in the current implementation). The user-supplied WG
>>> does
>>>>>>> not need to make sure the watermark doesn't regress.
>>>>>>>
>>>>>>> Regarding making the WG a "rich function", I can see the potential
>>>>>>> benefit but I also see a lot of pitfalls. For example, how should
>> the
>>>>>>> watermark state be handled in the case of scale-in? It could be made
>>> to
>>>>>>> work in the Kafka case by attaching the state to the partition state
>>>>>>> that we keep, but then we have potential backwards compatibility
>>>>>>> problems also for the WM state. Does the WG usually need to keep the
>>>>>>> state or might it be enough if the state is transient, i.e. if you
>>> have
>>>>>>> a restart the WG would loose its histogram but it would rebuild it
>>>>>>> quickly and you would get back to the same steady state as before.
>>>>>>>
>>>>>>> Best,
>>>>>>> Aljoscha
>>>>>>>
>>>>>>> On 27.04.20 12:12, David Anderson wrote:
>>>>>>>> Overall I like this proposal; thanks for bringing it forward,
>>>>>>>> Aljoscha.
>>>>>>>>
>>>>>>>> I also like the idea of making the Watermark generator a rich
>>> function
>>>>>> --
>>>>>>>> this should make it more straightforward to implement smarter
>>>>>>>> watermark
>>>>>>>> generators. Eg, one that uses state to keep statistics about the
>>>>>>>> actual
>>>>>>>> out-of-orderness, and uses those statistics to implement a variable
>>>>>>> delay.
>>>>>>>> David
>>>>>>>>
>>>>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
>> kkloudas@gmail.com>
>>>>>>> wrote:
>>>>>>>>> Hi Aljoscha,
>>>>>>>>>
>>>>>>>>> Thanks for opening the discussion!
>>>>>>>>>
>>>>>>>>> I have two comments on the FLIP:
>>>>>>>>> 1) we could add lifecycle methods to the Generator, i.e. open()/
>>>>>>>>> close(), probably with a Context as argument: I have not fully
>>>>>>>>> thought
>>>>>>>>> this through but I think that this is more aligned with the rest
>> of
>>>>>>>>> our rich functions. In addition, it will allow, for example, to
>>>>>>>>> initialize the Watermark value, if we decide to checkpoint the
>>>>>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to do
>>>>>>>>> anything in the open()).
>>>>>>>>> 2) aligned with the above, and with the case where we want to
>>>>>>>>> checkpoint the watermark in mind, I am wondering about how we
>> could
>>>>>>>>> implement this in the future. In the FLIP, it is proposed to
>> expose
>>>>>>>>> the WatermarkOutput in the methods of the WatermarkGenerator.
>> Given
>>>>>>>>> that there is the implicit contract that watermarks are
>>>>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
>>>>>>>>> assume) a check that will compare the last emitted WM against the
>>>>>>>>> provided one, and emit it only if it is >=. If not, then we risk
>>>>>>>>> having the user shooting himself on the foot if he/she
>> accidentally
>>>>>>>>> forgets the check. Given that the WatermarkGenerator and its
>>>>>>>>> caller do
>>>>>>>>> not know if the watermark was finally emitted or not (the
>>>>>>>>> WatermarkOutput#emitWatermark returns void), who will be
>> responsible
>>>>>>>>> for checkpointing the WM?
>>>>>>>>>
>>>>>>>>> Given this, why not having the methods as:
>>>>>>>>>
>>>>>>>>> public interface WatermarkGenerator<T> {
>>>>>>>>>
>>>>>>>>> Watermark onEvent(T event, long eventTimestamp,
>>> WatermarkOutput
>>>>>>>>> output);
>>>>>>>>>
>>>>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
>>>>>>>>> }
>>>>>>>>>
>>>>>>>>> and the caller will be the one enforcing any invariants, such as
>>>>>>>>> non-decreasing watermarks. In this way, the caller can checkpoint
>>>>>>>>> anything that is needed as it will have complete knowledge as to
>> if
>>>>>>>>> the WM was emitted or not.
>>>>>>>>>
>>>>>>>>> What do you think?
>>>>>>>>>
>>>>>>>>> Cheers,
>>>>>>>>> Kostas
>>>>>>>>>
>>>>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
>>>>>>>>>
>>>>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <tw...@apache.org>
>>>>>>> wrote:
>>>>>>>>>> Thanks for the proposal Aljoscha. This is a very useful
>>> unification.
>>>>>> We
>>>>>>>>>> have considered this FLIP already in the interfaces for FLIP-95
>> [1]
>>>>>> and
>>>>>>>>>> look forward to update to the new unified watermark generators
>> once
>>>>>>>>>> FLIP-126 has been accepted.
>>>>>>>>>>
>>>>>>>>>> Regards,
>>>>>>>>>> Timo
>>>>>>>>>>
>>>>>>>>>> [1] https://github.com/apache/flink/pull/11692
>>>>>>>>>>
>>>>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
>>>>>>>>>>> Hi Everyone!
>>>>>>>>>>>
>>>>>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
>>>>>> separate)
>>>>>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in an
>>>>>>>>>>> experimental branch. I expanded on that work to provide a PoC
>> for
>>>>>> the
>>>>>>>>>>> changes proposed in this FLIP: [2].
>>>>>>>>>>>
>>>>>>>>>>> Currently, we have two different flavours of Watermark
>>>>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
>>>>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
>>>>>>>>>>> from TimestampAssigner. This means that sources that want to
>>>>>>>>>>> support
>>>>>>>>>>> watermark assignment/extraction in the source need to support
>> two
>>>>>>>>>>> separate interfaces, we have two operator implementations for
>> the
>>>>>>>>>>> different flavours. Also, this makes features such as generic
>>>>>> support
>>>>>>>>>>> for idleness detection more complicated to implemented because
>> we
>>>>>>> again
>>>>>>>>>>> have to support two types of watermark assigners.
>>>>>>>>>>>
>>>>>>>>>>> In this FLIP we propose two things:
>>>>>>>>>>>
>>>>>>>>>>> Unify the Watermark Assigners into one Interface
>>> WatermarkGenerator
>>>>>>>>>>> Separate this new interface from the TimestampAssigner
>>>>>>>>>>> The motivation for the first is to simplify future
>> implementations
>>>>>> and
>>>>>>>>>>> code duplication. The motivation for the second point is again
>>> code
>>>>>>>>>>> deduplication, most assigners currently have to extend from some
>>>>>> base
>>>>>>>>>>> timestamp extractor or duplicate the extraction logic, or users
>>>>>>>>>>> have
>>>>>>> to
>>>>>>>>>>> override an abstract method of the watermark assigner to provide
>>>>>>>>>>> the
>>>>>>>>>>> timestamp extraction logic.
>>>>>>>>>>>
>>>>>>>>>>> Additionally, we propose to add a generic wrapping
>>>>>> WatermarkGenerator
>>>>>>>>>>> that provides idleness detection, i.e. it can mark a
>>>>>> stream/partition
>>>>>>>>> as
>>>>>>>>>>> idle if no data arrives after a configured timeout.
>>>>>>>>>>>
>>>>>>>>>>> The "unify and separate" part refers to the fact that we want to
>>>>>> unify
>>>>>>>>>>> punctuated and periodic assigners but at the same time split the
>>>>>>>>>>> timestamp assigner from the watermark generator.
>>>>>>>>>>>
>>>>>>>>>>> Please find more details in the FLIP [1]. Looking forward to
>>>>>>>>>>> your feedback.
>>>>>>>>>>>
>>>>>>>>>>> Best,
>>>>>>>>>>> Aljoscha
>>>>>>>>>>>
>>>>>>>>>>> [1]
>>>>>>>>>>>
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
>>>>>>>>>>>
>>>>>>>>>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
>>>>>>>
>>>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Jark Wu <im...@gmail.com>.
Thanks for the explanation. I like the fatory pattern to make the member
variables immutable and final.
So +1 to the proposal.
Best,
Jark
On Mon, 11 May 2020 at 22:01, Stephan Ewen <se...@apache.org> wrote:
> I am fine with that.
>
> Much of the principles seem agreed upon. I understand the need to support
> code-generated extractors and we should support most of it already (as
> Aljoscha mentioned via the factories) can extend this if needed.
>
> I think that the factory approach supports code-generated extractors in a
> cleaner way even than an extractor with an open/init method.
>
>
> On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek <al...@apache.org>
> wrote:
>
> > We're slightly running out of time. I would propose we vote on the basic
> > principle and remain open to later additions. This feature is quite
> > important to make the new Kafka Source that is developed as part of
> > FLIP-27 useful. Otherwise we would have to use the legacy interfaces in
> > the newly added connector.
> >
> > I know that's a bit unorthodox but would everyone be OK with what's
> > currently there and then we iterate?
> >
> > Best,
> > Aljoscha
> >
> > On 11.05.20 13:57, Aljoscha Krettek wrote:
> > > Ah, I meant to write this in my previous email, sorry about that.
> > >
> > > The WatermarkStrategy, which is basically a factory for a
> > > WatermarkGenerator is the replacement for the open() method. This is
> the
> > > same strategy that was followed for StreamOperatorFactory, which was
> > > introduced to allow code generation in the Table API [1]. If we need
> > > metrics or other things we would add that as a parameter to the factory
> > > method. What do you think?
> > >
> > > Best,
> > > Aljoscha
> > >
> > > [1] https://issues.apache.org/jira/browse/FLINK-11974
> > >
> > > On 10.05.20 05:07, Jark Wu wrote:
> > >> Hi,
> > >>
> > >> Regarding to the `open()/close()`, I think it's necessary for
> > >> Table&SQL to
> > >> compile the generated code.
> > >> In Table&SQL, the watermark strategy and event-timestamp is defined
> > using
> > >> SQL expressions, we will
> > >> translate and generate Java code for the expressions. If we have
> > >> `open()/close()`, we don't need lazy initialization.
> > >> Besides that, I can see a need to report some metrics, e.g. the
> current
> > >> watermark, the dirty timestamps (null value), etc.
> > >> So I think a simple `open()/close()` with a context which can get
> > >> MetricGroup is nice and not complex for the first version.
> > >>
> > >> Best,
> > >> Jark
> > >>
> > >>
> > >>
> > >> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org> wrote:
> > >>
> > >>> Thanks, Aljoscha, for picking this up.
> > >>>
> > >>> I agree with the approach of doing the here proposed set of changes
> for
> > >>> now. It already makes things simpler and adds idleness support
> > >>> everywhere.
> > >>>
> > >>> Rich functions and state always add complexity, let's do this in a
> next
> > >>> step, if we have a really compelling case.
> > >>>
> > >>>
> > >>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <
> aljoscha@apache.org>
> > >>> wrote:
> > >>>
> > >>>> Regarding the WatermarkGenerator (WG) interface itself. The proposal
> > is
> > >>>> basically to turn emitting into a "flatMap", we give the
> > >>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG
> can
> > >>>> decide whether to output a watermark or not and can also mark the
> > >>>> output
> > >>>> as idle. Changing the interface to return a Watermark (as the
> previous
> > >>>> watermark assigner interface did) would not allow that flexibility.
> > >>>>
> > >>>> Regarding checkpointing the watermark and keeping track of the
> minimum
> > >>>> watermark, this would be the responsibility of the framework (or the
> > >>>> KafkaConsumer in the current implementation). The user-supplied WG
> > does
> > >>>> not need to make sure the watermark doesn't regress.
> > >>>>
> > >>>> Regarding making the WG a "rich function", I can see the potential
> > >>>> benefit but I also see a lot of pitfalls. For example, how should
> the
> > >>>> watermark state be handled in the case of scale-in? It could be made
> > to
> > >>>> work in the Kafka case by attaching the state to the partition state
> > >>>> that we keep, but then we have potential backwards compatibility
> > >>>> problems also for the WM state. Does the WG usually need to keep the
> > >>>> state or might it be enough if the state is transient, i.e. if you
> > have
> > >>>> a restart the WG would loose its histogram but it would rebuild it
> > >>>> quickly and you would get back to the same steady state as before.
> > >>>>
> > >>>> Best,
> > >>>> Aljoscha
> > >>>>
> > >>>> On 27.04.20 12:12, David Anderson wrote:
> > >>>>> Overall I like this proposal; thanks for bringing it forward,
> > >>>>> Aljoscha.
> > >>>>>
> > >>>>> I also like the idea of making the Watermark generator a rich
> > function
> > >>> --
> > >>>>> this should make it more straightforward to implement smarter
> > >>>>> watermark
> > >>>>> generators. Eg, one that uses state to keep statistics about the
> > >>>>> actual
> > >>>>> out-of-orderness, and uses those statistics to implement a variable
> > >>>> delay.
> > >>>>>
> > >>>>> David
> > >>>>>
> > >>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <
> kkloudas@gmail.com>
> > >>>> wrote:
> > >>>>>
> > >>>>>> Hi Aljoscha,
> > >>>>>>
> > >>>>>> Thanks for opening the discussion!
> > >>>>>>
> > >>>>>> I have two comments on the FLIP:
> > >>>>>> 1) we could add lifecycle methods to the Generator, i.e. open()/
> > >>>>>> close(), probably with a Context as argument: I have not fully
> > >>>>>> thought
> > >>>>>> this through but I think that this is more aligned with the rest
> of
> > >>>>>> our rich functions. In addition, it will allow, for example, to
> > >>>>>> initialize the Watermark value, if we decide to checkpoint the
> > >>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to do
> > >>>>>> anything in the open()).
> > >>>>>> 2) aligned with the above, and with the case where we want to
> > >>>>>> checkpoint the watermark in mind, I am wondering about how we
> could
> > >>>>>> implement this in the future. In the FLIP, it is proposed to
> expose
> > >>>>>> the WatermarkOutput in the methods of the WatermarkGenerator.
> Given
> > >>>>>> that there is the implicit contract that watermarks are
> > >>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
> > >>>>>> assume) a check that will compare the last emitted WM against the
> > >>>>>> provided one, and emit it only if it is >=. If not, then we risk
> > >>>>>> having the user shooting himself on the foot if he/she
> accidentally
> > >>>>>> forgets the check. Given that the WatermarkGenerator and its
> > >>>>>> caller do
> > >>>>>> not know if the watermark was finally emitted or not (the
> > >>>>>> WatermarkOutput#emitWatermark returns void), who will be
> responsible
> > >>>>>> for checkpointing the WM?
> > >>>>>>
> > >>>>>> Given this, why not having the methods as:
> > >>>>>>
> > >>>>>> public interface WatermarkGenerator<T> {
> > >>>>>>
> > >>>>>> Watermark onEvent(T event, long eventTimestamp,
> > WatermarkOutput
> > >>>>>> output);
> > >>>>>>
> > >>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
> > >>>>>> }
> > >>>>>>
> > >>>>>> and the caller will be the one enforcing any invariants, such as
> > >>>>>> non-decreasing watermarks. In this way, the caller can checkpoint
> > >>>>>> anything that is needed as it will have complete knowledge as to
> if
> > >>>>>> the WM was emitted or not.
> > >>>>>>
> > >>>>>> What do you think?
> > >>>>>>
> > >>>>>> Cheers,
> > >>>>>> Kostas
> > >>>>>>
> > >>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
> > >>>>>>
> > >>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <tw...@apache.org>
> > >>>> wrote:
> > >>>>>>>
> > >>>>>>> Thanks for the proposal Aljoscha. This is a very useful
> > unification.
> > >>> We
> > >>>>>>> have considered this FLIP already in the interfaces for FLIP-95
> [1]
> > >>> and
> > >>>>>>> look forward to update to the new unified watermark generators
> once
> > >>>>>>> FLIP-126 has been accepted.
> > >>>>>>>
> > >>>>>>> Regards,
> > >>>>>>> Timo
> > >>>>>>>
> > >>>>>>> [1] https://github.com/apache/flink/pull/11692
> > >>>>>>>
> > >>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
> > >>>>>>>> Hi Everyone!
> > >>>>>>>>
> > >>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
> > >>> separate)
> > >>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in an
> > >>>>>>>> experimental branch. I expanded on that work to provide a PoC
> for
> > >>> the
> > >>>>>>>> changes proposed in this FLIP: [2].
> > >>>>>>>>
> > >>>>>>>> Currently, we have two different flavours of Watermark
> > >>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
> > >>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
> > >>>>>>>> from TimestampAssigner. This means that sources that want to
> > >>>>>>>> support
> > >>>>>>>> watermark assignment/extraction in the source need to support
> two
> > >>>>>>>> separate interfaces, we have two operator implementations for
> the
> > >>>>>>>> different flavours. Also, this makes features such as generic
> > >>> support
> > >>>>>>>> for idleness detection more complicated to implemented because
> we
> > >>>> again
> > >>>>>>>> have to support two types of watermark assigners.
> > >>>>>>>>
> > >>>>>>>> In this FLIP we propose two things:
> > >>>>>>>>
> > >>>>>>>> Unify the Watermark Assigners into one Interface
> > WatermarkGenerator
> > >>>>>>>> Separate this new interface from the TimestampAssigner
> > >>>>>>>> The motivation for the first is to simplify future
> implementations
> > >>> and
> > >>>>>>>> code duplication. The motivation for the second point is again
> > code
> > >>>>>>>> deduplication, most assigners currently have to extend from some
> > >>> base
> > >>>>>>>> timestamp extractor or duplicate the extraction logic, or users
> > >>>>>>>> have
> > >>>> to
> > >>>>>>>> override an abstract method of the watermark assigner to provide
> > >>>>>>>> the
> > >>>>>>>> timestamp extraction logic.
> > >>>>>>>>
> > >>>>>>>> Additionally, we propose to add a generic wrapping
> > >>> WatermarkGenerator
> > >>>>>>>> that provides idleness detection, i.e. it can mark a
> > >>> stream/partition
> > >>>>>> as
> > >>>>>>>> idle if no data arrives after a configured timeout.
> > >>>>>>>>
> > >>>>>>>> The "unify and separate" part refers to the fact that we want to
> > >>> unify
> > >>>>>>>> punctuated and periodic assigners but at the same time split the
> > >>>>>>>> timestamp assigner from the watermark generator.
> > >>>>>>>>
> > >>>>>>>> Please find more details in the FLIP [1]. Looking forward to
> > >>>>>>>> your feedback.
> > >>>>>>>>
> > >>>>>>>> Best,
> > >>>>>>>> Aljoscha
> > >>>>>>>>
> > >>>>>>>> [1]
> > >>>>>>>>
> > >>>>>>
> > >>>>
> > >>>
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
> > >>>
> > >>>>>>>>
> > >>>>>>>>
> > >>>>>>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
> > >>>>>>>
> > >>>>>>
> > >>>>>
> > >>>>
> > >>>>
> > >>>
> > >>
> > >
> >
> >
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Stephan Ewen <se...@apache.org>.
I am fine with that.
Much of the principles seem agreed upon. I understand the need to support
code-generated extractors and we should support most of it already (as
Aljoscha mentioned via the factories) can extend this if needed.
I think that the factory approach supports code-generated extractors in a
cleaner way even than an extractor with an open/init method.
On Mon, May 11, 2020 at 3:38 PM Aljoscha Krettek <al...@apache.org>
wrote:
> We're slightly running out of time. I would propose we vote on the basic
> principle and remain open to later additions. This feature is quite
> important to make the new Kafka Source that is developed as part of
> FLIP-27 useful. Otherwise we would have to use the legacy interfaces in
> the newly added connector.
>
> I know that's a bit unorthodox but would everyone be OK with what's
> currently there and then we iterate?
>
> Best,
> Aljoscha
>
> On 11.05.20 13:57, Aljoscha Krettek wrote:
> > Ah, I meant to write this in my previous email, sorry about that.
> >
> > The WatermarkStrategy, which is basically a factory for a
> > WatermarkGenerator is the replacement for the open() method. This is the
> > same strategy that was followed for StreamOperatorFactory, which was
> > introduced to allow code generation in the Table API [1]. If we need
> > metrics or other things we would add that as a parameter to the factory
> > method. What do you think?
> >
> > Best,
> > Aljoscha
> >
> > [1] https://issues.apache.org/jira/browse/FLINK-11974
> >
> > On 10.05.20 05:07, Jark Wu wrote:
> >> Hi,
> >>
> >> Regarding to the `open()/close()`, I think it's necessary for
> >> Table&SQL to
> >> compile the generated code.
> >> In Table&SQL, the watermark strategy and event-timestamp is defined
> using
> >> SQL expressions, we will
> >> translate and generate Java code for the expressions. If we have
> >> `open()/close()`, we don't need lazy initialization.
> >> Besides that, I can see a need to report some metrics, e.g. the current
> >> watermark, the dirty timestamps (null value), etc.
> >> So I think a simple `open()/close()` with a context which can get
> >> MetricGroup is nice and not complex for the first version.
> >>
> >> Best,
> >> Jark
> >>
> >>
> >>
> >> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org> wrote:
> >>
> >>> Thanks, Aljoscha, for picking this up.
> >>>
> >>> I agree with the approach of doing the here proposed set of changes for
> >>> now. It already makes things simpler and adds idleness support
> >>> everywhere.
> >>>
> >>> Rich functions and state always add complexity, let's do this in a next
> >>> step, if we have a really compelling case.
> >>>
> >>>
> >>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <al...@apache.org>
> >>> wrote:
> >>>
> >>>> Regarding the WatermarkGenerator (WG) interface itself. The proposal
> is
> >>>> basically to turn emitting into a "flatMap", we give the
> >>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG can
> >>>> decide whether to output a watermark or not and can also mark the
> >>>> output
> >>>> as idle. Changing the interface to return a Watermark (as the previous
> >>>> watermark assigner interface did) would not allow that flexibility.
> >>>>
> >>>> Regarding checkpointing the watermark and keeping track of the minimum
> >>>> watermark, this would be the responsibility of the framework (or the
> >>>> KafkaConsumer in the current implementation). The user-supplied WG
> does
> >>>> not need to make sure the watermark doesn't regress.
> >>>>
> >>>> Regarding making the WG a "rich function", I can see the potential
> >>>> benefit but I also see a lot of pitfalls. For example, how should the
> >>>> watermark state be handled in the case of scale-in? It could be made
> to
> >>>> work in the Kafka case by attaching the state to the partition state
> >>>> that we keep, but then we have potential backwards compatibility
> >>>> problems also for the WM state. Does the WG usually need to keep the
> >>>> state or might it be enough if the state is transient, i.e. if you
> have
> >>>> a restart the WG would loose its histogram but it would rebuild it
> >>>> quickly and you would get back to the same steady state as before.
> >>>>
> >>>> Best,
> >>>> Aljoscha
> >>>>
> >>>> On 27.04.20 12:12, David Anderson wrote:
> >>>>> Overall I like this proposal; thanks for bringing it forward,
> >>>>> Aljoscha.
> >>>>>
> >>>>> I also like the idea of making the Watermark generator a rich
> function
> >>> --
> >>>>> this should make it more straightforward to implement smarter
> >>>>> watermark
> >>>>> generators. Eg, one that uses state to keep statistics about the
> >>>>> actual
> >>>>> out-of-orderness, and uses those statistics to implement a variable
> >>>> delay.
> >>>>>
> >>>>> David
> >>>>>
> >>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <kk...@gmail.com>
> >>>> wrote:
> >>>>>
> >>>>>> Hi Aljoscha,
> >>>>>>
> >>>>>> Thanks for opening the discussion!
> >>>>>>
> >>>>>> I have two comments on the FLIP:
> >>>>>> 1) we could add lifecycle methods to the Generator, i.e. open()/
> >>>>>> close(), probably with a Context as argument: I have not fully
> >>>>>> thought
> >>>>>> this through but I think that this is more aligned with the rest of
> >>>>>> our rich functions. In addition, it will allow, for example, to
> >>>>>> initialize the Watermark value, if we decide to checkpoint the
> >>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to do
> >>>>>> anything in the open()).
> >>>>>> 2) aligned with the above, and with the case where we want to
> >>>>>> checkpoint the watermark in mind, I am wondering about how we could
> >>>>>> implement this in the future. In the FLIP, it is proposed to expose
> >>>>>> the WatermarkOutput in the methods of the WatermarkGenerator. Given
> >>>>>> that there is the implicit contract that watermarks are
> >>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
> >>>>>> assume) a check that will compare the last emitted WM against the
> >>>>>> provided one, and emit it only if it is >=. If not, then we risk
> >>>>>> having the user shooting himself on the foot if he/she accidentally
> >>>>>> forgets the check. Given that the WatermarkGenerator and its
> >>>>>> caller do
> >>>>>> not know if the watermark was finally emitted or not (the
> >>>>>> WatermarkOutput#emitWatermark returns void), who will be responsible
> >>>>>> for checkpointing the WM?
> >>>>>>
> >>>>>> Given this, why not having the methods as:
> >>>>>>
> >>>>>> public interface WatermarkGenerator<T> {
> >>>>>>
> >>>>>> Watermark onEvent(T event, long eventTimestamp,
> WatermarkOutput
> >>>>>> output);
> >>>>>>
> >>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
> >>>>>> }
> >>>>>>
> >>>>>> and the caller will be the one enforcing any invariants, such as
> >>>>>> non-decreasing watermarks. In this way, the caller can checkpoint
> >>>>>> anything that is needed as it will have complete knowledge as to if
> >>>>>> the WM was emitted or not.
> >>>>>>
> >>>>>> What do you think?
> >>>>>>
> >>>>>> Cheers,
> >>>>>> Kostas
> >>>>>>
> >>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
> >>>>>>
> >>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <tw...@apache.org>
> >>>> wrote:
> >>>>>>>
> >>>>>>> Thanks for the proposal Aljoscha. This is a very useful
> unification.
> >>> We
> >>>>>>> have considered this FLIP already in the interfaces for FLIP-95 [1]
> >>> and
> >>>>>>> look forward to update to the new unified watermark generators once
> >>>>>>> FLIP-126 has been accepted.
> >>>>>>>
> >>>>>>> Regards,
> >>>>>>> Timo
> >>>>>>>
> >>>>>>> [1] https://github.com/apache/flink/pull/11692
> >>>>>>>
> >>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
> >>>>>>>> Hi Everyone!
> >>>>>>>>
> >>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
> >>> separate)
> >>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in an
> >>>>>>>> experimental branch. I expanded on that work to provide a PoC for
> >>> the
> >>>>>>>> changes proposed in this FLIP: [2].
> >>>>>>>>
> >>>>>>>> Currently, we have two different flavours of Watermark
> >>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
> >>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
> >>>>>>>> from TimestampAssigner. This means that sources that want to
> >>>>>>>> support
> >>>>>>>> watermark assignment/extraction in the source need to support two
> >>>>>>>> separate interfaces, we have two operator implementations for the
> >>>>>>>> different flavours. Also, this makes features such as generic
> >>> support
> >>>>>>>> for idleness detection more complicated to implemented because we
> >>>> again
> >>>>>>>> have to support two types of watermark assigners.
> >>>>>>>>
> >>>>>>>> In this FLIP we propose two things:
> >>>>>>>>
> >>>>>>>> Unify the Watermark Assigners into one Interface
> WatermarkGenerator
> >>>>>>>> Separate this new interface from the TimestampAssigner
> >>>>>>>> The motivation for the first is to simplify future implementations
> >>> and
> >>>>>>>> code duplication. The motivation for the second point is again
> code
> >>>>>>>> deduplication, most assigners currently have to extend from some
> >>> base
> >>>>>>>> timestamp extractor or duplicate the extraction logic, or users
> >>>>>>>> have
> >>>> to
> >>>>>>>> override an abstract method of the watermark assigner to provide
> >>>>>>>> the
> >>>>>>>> timestamp extraction logic.
> >>>>>>>>
> >>>>>>>> Additionally, we propose to add a generic wrapping
> >>> WatermarkGenerator
> >>>>>>>> that provides idleness detection, i.e. it can mark a
> >>> stream/partition
> >>>>>> as
> >>>>>>>> idle if no data arrives after a configured timeout.
> >>>>>>>>
> >>>>>>>> The "unify and separate" part refers to the fact that we want to
> >>> unify
> >>>>>>>> punctuated and periodic assigners but at the same time split the
> >>>>>>>> timestamp assigner from the watermark generator.
> >>>>>>>>
> >>>>>>>> Please find more details in the FLIP [1]. Looking forward to
> >>>>>>>> your feedback.
> >>>>>>>>
> >>>>>>>> Best,
> >>>>>>>> Aljoscha
> >>>>>>>>
> >>>>>>>> [1]
> >>>>>>>>
> >>>>>>
> >>>>
> >>>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
> >>>
> >>>>>>>>
> >>>>>>>>
> >>>>>>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
> >>>>>>>
> >>>>>>
> >>>>>
> >>>>
> >>>>
> >>>
> >>
> >
>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Aljoscha Krettek <al...@apache.org>.
We're slightly running out of time. I would propose we vote on the basic
principle and remain open to later additions. This feature is quite
important to make the new Kafka Source that is developed as part of
FLIP-27 useful. Otherwise we would have to use the legacy interfaces in
the newly added connector.
I know that's a bit unorthodox but would everyone be OK with what's
currently there and then we iterate?
Best,
Aljoscha
On 11.05.20 13:57, Aljoscha Krettek wrote:
> Ah, I meant to write this in my previous email, sorry about that.
>
> The WatermarkStrategy, which is basically a factory for a
> WatermarkGenerator is the replacement for the open() method. This is the
> same strategy that was followed for StreamOperatorFactory, which was
> introduced to allow code generation in the Table API [1]. If we need
> metrics or other things we would add that as a parameter to the factory
> method. What do you think?
>
> Best,
> Aljoscha
>
> [1] https://issues.apache.org/jira/browse/FLINK-11974
>
> On 10.05.20 05:07, Jark Wu wrote:
>> Hi,
>>
>> Regarding to the `open()/close()`, I think it's necessary for
>> Table&SQL to
>> compile the generated code.
>> In Table&SQL, the watermark strategy and event-timestamp is defined using
>> SQL expressions, we will
>> translate and generate Java code for the expressions. If we have
>> `open()/close()`, we don't need lazy initialization.
>> Besides that, I can see a need to report some metrics, e.g. the current
>> watermark, the dirty timestamps (null value), etc.
>> So I think a simple `open()/close()` with a context which can get
>> MetricGroup is nice and not complex for the first version.
>>
>> Best,
>> Jark
>>
>>
>>
>> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org> wrote:
>>
>>> Thanks, Aljoscha, for picking this up.
>>>
>>> I agree with the approach of doing the here proposed set of changes for
>>> now. It already makes things simpler and adds idleness support
>>> everywhere.
>>>
>>> Rich functions and state always add complexity, let's do this in a next
>>> step, if we have a really compelling case.
>>>
>>>
>>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <al...@apache.org>
>>> wrote:
>>>
>>>> Regarding the WatermarkGenerator (WG) interface itself. The proposal is
>>>> basically to turn emitting into a "flatMap", we give the
>>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG can
>>>> decide whether to output a watermark or not and can also mark the
>>>> output
>>>> as idle. Changing the interface to return a Watermark (as the previous
>>>> watermark assigner interface did) would not allow that flexibility.
>>>>
>>>> Regarding checkpointing the watermark and keeping track of the minimum
>>>> watermark, this would be the responsibility of the framework (or the
>>>> KafkaConsumer in the current implementation). The user-supplied WG does
>>>> not need to make sure the watermark doesn't regress.
>>>>
>>>> Regarding making the WG a "rich function", I can see the potential
>>>> benefit but I also see a lot of pitfalls. For example, how should the
>>>> watermark state be handled in the case of scale-in? It could be made to
>>>> work in the Kafka case by attaching the state to the partition state
>>>> that we keep, but then we have potential backwards compatibility
>>>> problems also for the WM state. Does the WG usually need to keep the
>>>> state or might it be enough if the state is transient, i.e. if you have
>>>> a restart the WG would loose its histogram but it would rebuild it
>>>> quickly and you would get back to the same steady state as before.
>>>>
>>>> Best,
>>>> Aljoscha
>>>>
>>>> On 27.04.20 12:12, David Anderson wrote:
>>>>> Overall I like this proposal; thanks for bringing it forward,
>>>>> Aljoscha.
>>>>>
>>>>> I also like the idea of making the Watermark generator a rich function
>>> --
>>>>> this should make it more straightforward to implement smarter
>>>>> watermark
>>>>> generators. Eg, one that uses state to keep statistics about the
>>>>> actual
>>>>> out-of-orderness, and uses those statistics to implement a variable
>>>> delay.
>>>>>
>>>>> David
>>>>>
>>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <kk...@gmail.com>
>>>> wrote:
>>>>>
>>>>>> Hi Aljoscha,
>>>>>>
>>>>>> Thanks for opening the discussion!
>>>>>>
>>>>>> I have two comments on the FLIP:
>>>>>> 1) we could add lifecycle methods to the Generator, i.e. open()/
>>>>>> close(), probably with a Context as argument: I have not fully
>>>>>> thought
>>>>>> this through but I think that this is more aligned with the rest of
>>>>>> our rich functions. In addition, it will allow, for example, to
>>>>>> initialize the Watermark value, if we decide to checkpoint the
>>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to do
>>>>>> anything in the open()).
>>>>>> 2) aligned with the above, and with the case where we want to
>>>>>> checkpoint the watermark in mind, I am wondering about how we could
>>>>>> implement this in the future. In the FLIP, it is proposed to expose
>>>>>> the WatermarkOutput in the methods of the WatermarkGenerator. Given
>>>>>> that there is the implicit contract that watermarks are
>>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
>>>>>> assume) a check that will compare the last emitted WM against the
>>>>>> provided one, and emit it only if it is >=. If not, then we risk
>>>>>> having the user shooting himself on the foot if he/she accidentally
>>>>>> forgets the check. Given that the WatermarkGenerator and its
>>>>>> caller do
>>>>>> not know if the watermark was finally emitted or not (the
>>>>>> WatermarkOutput#emitWatermark returns void), who will be responsible
>>>>>> for checkpointing the WM?
>>>>>>
>>>>>> Given this, why not having the methods as:
>>>>>>
>>>>>> public interface WatermarkGenerator<T> {
>>>>>>
>>>>>> Watermark onEvent(T event, long eventTimestamp, WatermarkOutput
>>>>>> output);
>>>>>>
>>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
>>>>>> }
>>>>>>
>>>>>> and the caller will be the one enforcing any invariants, such as
>>>>>> non-decreasing watermarks. In this way, the caller can checkpoint
>>>>>> anything that is needed as it will have complete knowledge as to if
>>>>>> the WM was emitted or not.
>>>>>>
>>>>>> What do you think?
>>>>>>
>>>>>> Cheers,
>>>>>> Kostas
>>>>>>
>>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
>>>>>>
>>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <tw...@apache.org>
>>>> wrote:
>>>>>>>
>>>>>>> Thanks for the proposal Aljoscha. This is a very useful unification.
>>> We
>>>>>>> have considered this FLIP already in the interfaces for FLIP-95 [1]
>>> and
>>>>>>> look forward to update to the new unified watermark generators once
>>>>>>> FLIP-126 has been accepted.
>>>>>>>
>>>>>>> Regards,
>>>>>>> Timo
>>>>>>>
>>>>>>> [1] https://github.com/apache/flink/pull/11692
>>>>>>>
>>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
>>>>>>>> Hi Everyone!
>>>>>>>>
>>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
>>> separate)
>>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in an
>>>>>>>> experimental branch. I expanded on that work to provide a PoC for
>>> the
>>>>>>>> changes proposed in this FLIP: [2].
>>>>>>>>
>>>>>>>> Currently, we have two different flavours of Watermark
>>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
>>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
>>>>>>>> from TimestampAssigner. This means that sources that want to
>>>>>>>> support
>>>>>>>> watermark assignment/extraction in the source need to support two
>>>>>>>> separate interfaces, we have two operator implementations for the
>>>>>>>> different flavours. Also, this makes features such as generic
>>> support
>>>>>>>> for idleness detection more complicated to implemented because we
>>>> again
>>>>>>>> have to support two types of watermark assigners.
>>>>>>>>
>>>>>>>> In this FLIP we propose two things:
>>>>>>>>
>>>>>>>> Unify the Watermark Assigners into one Interface WatermarkGenerator
>>>>>>>> Separate this new interface from the TimestampAssigner
>>>>>>>> The motivation for the first is to simplify future implementations
>>> and
>>>>>>>> code duplication. The motivation for the second point is again code
>>>>>>>> deduplication, most assigners currently have to extend from some
>>> base
>>>>>>>> timestamp extractor or duplicate the extraction logic, or users
>>>>>>>> have
>>>> to
>>>>>>>> override an abstract method of the watermark assigner to provide
>>>>>>>> the
>>>>>>>> timestamp extraction logic.
>>>>>>>>
>>>>>>>> Additionally, we propose to add a generic wrapping
>>> WatermarkGenerator
>>>>>>>> that provides idleness detection, i.e. it can mark a
>>> stream/partition
>>>>>> as
>>>>>>>> idle if no data arrives after a configured timeout.
>>>>>>>>
>>>>>>>> The "unify and separate" part refers to the fact that we want to
>>> unify
>>>>>>>> punctuated and periodic assigners but at the same time split the
>>>>>>>> timestamp assigner from the watermark generator.
>>>>>>>>
>>>>>>>> Please find more details in the FLIP [1]. Looking forward to
>>>>>>>> your feedback.
>>>>>>>>
>>>>>>>> Best,
>>>>>>>> Aljoscha
>>>>>>>>
>>>>>>>> [1]
>>>>>>>>
>>>>>>
>>>>
>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
>>>
>>>>>>>>
>>>>>>>>
>>>>>>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
>>>>>>>
>>>>>>
>>>>>
>>>>
>>>>
>>>
>>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Aljoscha Krettek <al...@apache.org>.
Ah, I meant to write this in my previous email, sorry about that.
The WatermarkStrategy, which is basically a factory for a
WatermarkGenerator is the replacement for the open() method. This is the
same strategy that was followed for StreamOperatorFactory, which was
introduced to allow code generation in the Table API [1]. If we need
metrics or other things we would add that as a parameter to the factory
method. What do you think?
Best,
Aljoscha
[1] https://issues.apache.org/jira/browse/FLINK-11974
On 10.05.20 05:07, Jark Wu wrote:
> Hi,
>
> Regarding to the `open()/close()`, I think it's necessary for Table&SQL to
> compile the generated code.
> In Table&SQL, the watermark strategy and event-timestamp is defined using
> SQL expressions, we will
> translate and generate Java code for the expressions. If we have
> `open()/close()`, we don't need lazy initialization.
> Besides that, I can see a need to report some metrics, e.g. the current
> watermark, the dirty timestamps (null value), etc.
> So I think a simple `open()/close()` with a context which can get
> MetricGroup is nice and not complex for the first version.
>
> Best,
> Jark
>
>
>
> On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org> wrote:
>
>> Thanks, Aljoscha, for picking this up.
>>
>> I agree with the approach of doing the here proposed set of changes for
>> now. It already makes things simpler and adds idleness support everywhere.
>>
>> Rich functions and state always add complexity, let's do this in a next
>> step, if we have a really compelling case.
>>
>>
>> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <al...@apache.org>
>> wrote:
>>
>>> Regarding the WatermarkGenerator (WG) interface itself. The proposal is
>>> basically to turn emitting into a "flatMap", we give the
>>> WatermarkGenerator a "collector" (the WatermarkOutput) and the WG can
>>> decide whether to output a watermark or not and can also mark the output
>>> as idle. Changing the interface to return a Watermark (as the previous
>>> watermark assigner interface did) would not allow that flexibility.
>>>
>>> Regarding checkpointing the watermark and keeping track of the minimum
>>> watermark, this would be the responsibility of the framework (or the
>>> KafkaConsumer in the current implementation). The user-supplied WG does
>>> not need to make sure the watermark doesn't regress.
>>>
>>> Regarding making the WG a "rich function", I can see the potential
>>> benefit but I also see a lot of pitfalls. For example, how should the
>>> watermark state be handled in the case of scale-in? It could be made to
>>> work in the Kafka case by attaching the state to the partition state
>>> that we keep, but then we have potential backwards compatibility
>>> problems also for the WM state. Does the WG usually need to keep the
>>> state or might it be enough if the state is transient, i.e. if you have
>>> a restart the WG would loose its histogram but it would rebuild it
>>> quickly and you would get back to the same steady state as before.
>>>
>>> Best,
>>> Aljoscha
>>>
>>> On 27.04.20 12:12, David Anderson wrote:
>>>> Overall I like this proposal; thanks for bringing it forward, Aljoscha.
>>>>
>>>> I also like the idea of making the Watermark generator a rich function
>> --
>>>> this should make it more straightforward to implement smarter watermark
>>>> generators. Eg, one that uses state to keep statistics about the actual
>>>> out-of-orderness, and uses those statistics to implement a variable
>>> delay.
>>>>
>>>> David
>>>>
>>>> On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <kk...@gmail.com>
>>> wrote:
>>>>
>>>>> Hi Aljoscha,
>>>>>
>>>>> Thanks for opening the discussion!
>>>>>
>>>>> I have two comments on the FLIP:
>>>>> 1) we could add lifecycle methods to the Generator, i.e. open()/
>>>>> close(), probably with a Context as argument: I have not fully thought
>>>>> this through but I think that this is more aligned with the rest of
>>>>> our rich functions. In addition, it will allow, for example, to
>>>>> initialize the Watermark value, if we decide to checkpoint the
>>>>> watermark (see [1]) (I also do not know if Table/SQL needs to do
>>>>> anything in the open()).
>>>>> 2) aligned with the above, and with the case where we want to
>>>>> checkpoint the watermark in mind, I am wondering about how we could
>>>>> implement this in the future. In the FLIP, it is proposed to expose
>>>>> the WatermarkOutput in the methods of the WatermarkGenerator. Given
>>>>> that there is the implicit contract that watermarks are
>>>>> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
>>>>> assume) a check that will compare the last emitted WM against the
>>>>> provided one, and emit it only if it is >=. If not, then we risk
>>>>> having the user shooting himself on the foot if he/she accidentally
>>>>> forgets the check. Given that the WatermarkGenerator and its caller do
>>>>> not know if the watermark was finally emitted or not (the
>>>>> WatermarkOutput#emitWatermark returns void), who will be responsible
>>>>> for checkpointing the WM?
>>>>>
>>>>> Given this, why not having the methods as:
>>>>>
>>>>> public interface WatermarkGenerator<T> {
>>>>>
>>>>> Watermark onEvent(T event, long eventTimestamp, WatermarkOutput
>>>>> output);
>>>>>
>>>>> Watermark onPeriodicEmit(WatermarkOutput output);
>>>>> }
>>>>>
>>>>> and the caller will be the one enforcing any invariants, such as
>>>>> non-decreasing watermarks. In this way, the caller can checkpoint
>>>>> anything that is needed as it will have complete knowledge as to if
>>>>> the WM was emitted or not.
>>>>>
>>>>> What do you think?
>>>>>
>>>>> Cheers,
>>>>> Kostas
>>>>>
>>>>> [1] https://issues.apache.org/jira/browse/FLINK-5601
>>>>>
>>>>> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <tw...@apache.org>
>>> wrote:
>>>>>>
>>>>>> Thanks for the proposal Aljoscha. This is a very useful unification.
>> We
>>>>>> have considered this FLIP already in the interfaces for FLIP-95 [1]
>> and
>>>>>> look forward to update to the new unified watermark generators once
>>>>>> FLIP-126 has been accepted.
>>>>>>
>>>>>> Regards,
>>>>>> Timo
>>>>>>
>>>>>> [1] https://github.com/apache/flink/pull/11692
>>>>>>
>>>>>> On 20.04.20 18:10, Aljoscha Krettek wrote:
>>>>>>> Hi Everyone!
>>>>>>>
>>>>>>> We would like to start a discussion on "FLIP-126: Unify (and
>> separate)
>>>>>>> Watermark Assigners" [1]. This work was started by Stephan in an
>>>>>>> experimental branch. I expanded on that work to provide a PoC for
>> the
>>>>>>> changes proposed in this FLIP: [2].
>>>>>>>
>>>>>>> Currently, we have two different flavours of Watermark
>>>>>>> Assigners: AssignerWithPunctuatedWatermarks
>>>>>>> and AssignerWithPeriodicWatermarks. Both of them extend
>>>>>>> from TimestampAssigner. This means that sources that want to support
>>>>>>> watermark assignment/extraction in the source need to support two
>>>>>>> separate interfaces, we have two operator implementations for the
>>>>>>> different flavours. Also, this makes features such as generic
>> support
>>>>>>> for idleness detection more complicated to implemented because we
>>> again
>>>>>>> have to support two types of watermark assigners.
>>>>>>>
>>>>>>> In this FLIP we propose two things:
>>>>>>>
>>>>>>> Unify the Watermark Assigners into one Interface WatermarkGenerator
>>>>>>> Separate this new interface from the TimestampAssigner
>>>>>>> The motivation for the first is to simplify future implementations
>> and
>>>>>>> code duplication. The motivation for the second point is again code
>>>>>>> deduplication, most assigners currently have to extend from some
>> base
>>>>>>> timestamp extractor or duplicate the extraction logic, or users have
>>> to
>>>>>>> override an abstract method of the watermark assigner to provide the
>>>>>>> timestamp extraction logic.
>>>>>>>
>>>>>>> Additionally, we propose to add a generic wrapping
>> WatermarkGenerator
>>>>>>> that provides idleness detection, i.e. it can mark a
>> stream/partition
>>>>> as
>>>>>>> idle if no data arrives after a configured timeout.
>>>>>>>
>>>>>>> The "unify and separate" part refers to the fact that we want to
>> unify
>>>>>>> punctuated and periodic assigners but at the same time split the
>>>>>>> timestamp assigner from the watermark generator.
>>>>>>>
>>>>>>> Please find more details in the FLIP [1]. Looking forward to
>>>>>>> your feedback.
>>>>>>>
>>>>>>> Best,
>>>>>>> Aljoscha
>>>>>>>
>>>>>>> [1]
>>>>>>>
>>>>>
>>>
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
>>>>>>>
>>>>>>>
>>>>>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
>>>>>>
>>>>>
>>>>
>>>
>>>
>>
>
Re: [DISCUSS] FLIP-126: Unify (and separate) Watermark Assigners
Posted by Jark Wu <im...@gmail.com>.
Hi,
Regarding to the `open()/close()`, I think it's necessary for Table&SQL to
compile the generated code.
In Table&SQL, the watermark strategy and event-timestamp is defined using
SQL expressions, we will
translate and generate Java code for the expressions. If we have
`open()/close()`, we don't need lazy initialization.
Besides that, I can see a need to report some metrics, e.g. the current
watermark, the dirty timestamps (null value), etc.
So I think a simple `open()/close()` with a context which can get
MetricGroup is nice and not complex for the first version.
Best,
Jark
On Sun, 10 May 2020 at 00:50, Stephan Ewen <se...@apache.org> wrote:
> Thanks, Aljoscha, for picking this up.
>
> I agree with the approach of doing the here proposed set of changes for
> now. It already makes things simpler and adds idleness support everywhere.
>
> Rich functions and state always add complexity, let's do this in a next
> step, if we have a really compelling case.
>
>
> On Wed, Apr 29, 2020 at 7:24 PM Aljoscha Krettek <al...@apache.org>
> wrote:
>
> > Regarding the WatermarkGenerator (WG) interface itself. The proposal is
> > basically to turn emitting into a "flatMap", we give the
> > WatermarkGenerator a "collector" (the WatermarkOutput) and the WG can
> > decide whether to output a watermark or not and can also mark the output
> > as idle. Changing the interface to return a Watermark (as the previous
> > watermark assigner interface did) would not allow that flexibility.
> >
> > Regarding checkpointing the watermark and keeping track of the minimum
> > watermark, this would be the responsibility of the framework (or the
> > KafkaConsumer in the current implementation). The user-supplied WG does
> > not need to make sure the watermark doesn't regress.
> >
> > Regarding making the WG a "rich function", I can see the potential
> > benefit but I also see a lot of pitfalls. For example, how should the
> > watermark state be handled in the case of scale-in? It could be made to
> > work in the Kafka case by attaching the state to the partition state
> > that we keep, but then we have potential backwards compatibility
> > problems also for the WM state. Does the WG usually need to keep the
> > state or might it be enough if the state is transient, i.e. if you have
> > a restart the WG would loose its histogram but it would rebuild it
> > quickly and you would get back to the same steady state as before.
> >
> > Best,
> > Aljoscha
> >
> > On 27.04.20 12:12, David Anderson wrote:
> > > Overall I like this proposal; thanks for bringing it forward, Aljoscha.
> > >
> > > I also like the idea of making the Watermark generator a rich function
> --
> > > this should make it more straightforward to implement smarter watermark
> > > generators. Eg, one that uses state to keep statistics about the actual
> > > out-of-orderness, and uses those statistics to implement a variable
> > delay.
> > >
> > > David
> > >
> > > On Mon, Apr 27, 2020 at 11:44 AM Kostas Kloudas <kk...@gmail.com>
> > wrote:
> > >
> > >> Hi Aljoscha,
> > >>
> > >> Thanks for opening the discussion!
> > >>
> > >> I have two comments on the FLIP:
> > >> 1) we could add lifecycle methods to the Generator, i.e. open()/
> > >> close(), probably with a Context as argument: I have not fully thought
> > >> this through but I think that this is more aligned with the rest of
> > >> our rich functions. In addition, it will allow, for example, to
> > >> initialize the Watermark value, if we decide to checkpoint the
> > >> watermark (see [1]) (I also do not know if Table/SQL needs to do
> > >> anything in the open()).
> > >> 2) aligned with the above, and with the case where we want to
> > >> checkpoint the watermark in mind, I am wondering about how we could
> > >> implement this in the future. In the FLIP, it is proposed to expose
> > >> the WatermarkOutput in the methods of the WatermarkGenerator. Given
> > >> that there is the implicit contract that watermarks are
> > >> non-decreasing, the WatermarkOutput#emitWatermark() will have (I
> > >> assume) a check that will compare the last emitted WM against the
> > >> provided one, and emit it only if it is >=. If not, then we risk
> > >> having the user shooting himself on the foot if he/she accidentally
> > >> forgets the check. Given that the WatermarkGenerator and its caller do
> > >> not know if the watermark was finally emitted or not (the
> > >> WatermarkOutput#emitWatermark returns void), who will be responsible
> > >> for checkpointing the WM?
> > >>
> > >> Given this, why not having the methods as:
> > >>
> > >> public interface WatermarkGenerator<T> {
> > >>
> > >> Watermark onEvent(T event, long eventTimestamp, WatermarkOutput
> > >> output);
> > >>
> > >> Watermark onPeriodicEmit(WatermarkOutput output);
> > >> }
> > >>
> > >> and the caller will be the one enforcing any invariants, such as
> > >> non-decreasing watermarks. In this way, the caller can checkpoint
> > >> anything that is needed as it will have complete knowledge as to if
> > >> the WM was emitted or not.
> > >>
> > >> What do you think?
> > >>
> > >> Cheers,
> > >> Kostas
> > >>
> > >> [1] https://issues.apache.org/jira/browse/FLINK-5601
> > >>
> > >> On Tue, Apr 21, 2020 at 2:25 PM Timo Walther <tw...@apache.org>
> > wrote:
> > >>>
> > >>> Thanks for the proposal Aljoscha. This is a very useful unification.
> We
> > >>> have considered this FLIP already in the interfaces for FLIP-95 [1]
> and
> > >>> look forward to update to the new unified watermark generators once
> > >>> FLIP-126 has been accepted.
> > >>>
> > >>> Regards,
> > >>> Timo
> > >>>
> > >>> [1] https://github.com/apache/flink/pull/11692
> > >>>
> > >>> On 20.04.20 18:10, Aljoscha Krettek wrote:
> > >>>> Hi Everyone!
> > >>>>
> > >>>> We would like to start a discussion on "FLIP-126: Unify (and
> separate)
> > >>>> Watermark Assigners" [1]. This work was started by Stephan in an
> > >>>> experimental branch. I expanded on that work to provide a PoC for
> the
> > >>>> changes proposed in this FLIP: [2].
> > >>>>
> > >>>> Currently, we have two different flavours of Watermark
> > >>>> Assigners: AssignerWithPunctuatedWatermarks
> > >>>> and AssignerWithPeriodicWatermarks. Both of them extend
> > >>>> from TimestampAssigner. This means that sources that want to support
> > >>>> watermark assignment/extraction in the source need to support two
> > >>>> separate interfaces, we have two operator implementations for the
> > >>>> different flavours. Also, this makes features such as generic
> support
> > >>>> for idleness detection more complicated to implemented because we
> > again
> > >>>> have to support two types of watermark assigners.
> > >>>>
> > >>>> In this FLIP we propose two things:
> > >>>>
> > >>>> Unify the Watermark Assigners into one Interface WatermarkGenerator
> > >>>> Separate this new interface from the TimestampAssigner
> > >>>> The motivation for the first is to simplify future implementations
> and
> > >>>> code duplication. The motivation for the second point is again code
> > >>>> deduplication, most assigners currently have to extend from some
> base
> > >>>> timestamp extractor or duplicate the extraction logic, or users have
> > to
> > >>>> override an abstract method of the watermark assigner to provide the
> > >>>> timestamp extraction logic.
> > >>>>
> > >>>> Additionally, we propose to add a generic wrapping
> WatermarkGenerator
> > >>>> that provides idleness detection, i.e. it can mark a
> stream/partition
> > >> as
> > >>>> idle if no data arrives after a configured timeout.
> > >>>>
> > >>>> The "unify and separate" part refers to the fact that we want to
> unify
> > >>>> punctuated and periodic assigners but at the same time split the
> > >>>> timestamp assigner from the watermark generator.
> > >>>>
> > >>>> Please find more details in the FLIP [1]. Looking forward to
> > >>>> your feedback.
> > >>>>
> > >>>> Best,
> > >>>> Aljoscha
> > >>>>
> > >>>> [1]
> > >>>>
> > >>
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-126%3A+Unify+%28and+separate%29+Watermark+Assigners
> > >>>>
> > >>>>
> > >>>> [2] https://github.com/aljoscha/flink/tree/stephan-event-time
> > >>>
> > >>
> > >
> >
> >
>