You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by Michal Borowiecki <mi...@openbet.com> on 2017/06/02 16:33:38 UTC
Re: [DISCUSS] KIP-138: Change punctuate semantics
Hi Matthias,
Apologies, somehow I totally missed this email earlier.
Wrt ValueTransformer, I added it to the the list of deprecated methods
(PR is up to date).
Wrt Cancellable vs Cancelable:
I'm not fluent enough to have spotted this nuance, but having googled
for it, you are right.
On the other hand however, the precedent seems to have been set by
java.util.concurrent.Cancellable and akka for instance followed that
with akka.actor.Cancellable.
Given established heritage in computing context, I'd err on the side of
consistency with prior practice.
Unless anyone has strong opinions on this matter?
Thanks,
Michal
On 04/05/17 20:43, Matthias J. Sax wrote:
> Hi,
>
> thanks for updating the KIP. Looks good to me overall.
>
> I think adding `Cancellable` (or should it be `Cancelable` to follow
> American English?) is a clean solution, in contrast to the proposed
> alternative.
>
> One minor comment: can you add `ValueTransformer#punctuate()` to the
> list of deprecated methods?
>
>
> -Matthias
>
>
>
> On 5/4/17 1:41 AM, Michal Borowiecki wrote:
>> Further in this direction I've updated the main proposal to incorporate
>> the Cancellable return type for ProcessorContext.schedule and the
>> guidance on how to implement "hybrid" punctuation with the proposed 2
>> PunctuationTypes.
>>
>> I look forward to more comments whether the Cancallable return type is
>> an agreeable solution and it's precise definition.
>>
>> I shall move all alternatives other than the main proposal into the
>> Rejected Alternatives section and if I hear any objections, I'll move
>> those back up and we'll discuss further.
>>
>>
>> Looking forward to all comments and suggestions.
>>
>>
>> Thanks,
>>
>> Michal
>>
>>
>> On 01/05/17 18:23, Michal Borowiecki wrote:
>>> Hi all,
>>>
>>> As promised, here is my take at how one could implement the previously
>>> discussed hybrid semantics using the 2 PunctuationType callbacks (one
>>> for STREAM_TIME and one for SYSTEM_TIME).
>>>
>>> However, there's a twist.
>>>
>>> Since currently calling context.schedule() adds a new
>>> PunctuationSchedule and does not overwrite the previous one, a slight
>>> change would be required:
>>>
>>> a) either that PuncuationSchedules are cancellable
>>>
>>> b) or that calling schedule() ||overwrites(cancels) the previous one
>>> with the given |PunctuationType |(but that's not how it works currently)
>>>
>>>
>>> Below is an example assuming approach a) is implemented by having
>>> schedule return Cancellable instead of void.
>>>
>>> |ProcessorContext context;|
>>> |long| |streamTimeInterval = ...;|
>>> |long| |systemTimeUpperBound = ...; ||//e.g. systemTimeUpperBound =
>>> streamTimeInterval + some tolerance|
>>> |Cancellable streamTimeSchedule;|
>>> |Cancellable systemTimeSchedule;|
>>> |long| |lastStreamTimePunctation = -||1||;|
>>> | |
>>> |public| |void| |init(ProcessorContext context){|
>>> | ||this||.context = context;|
>>> | ||streamTimeSchedule =
>>> context.schedule(PunctuationType.STREAM_TIME,
>>> streamTimeInterval, ||this||::streamTimePunctuate);|
>>> | ||systemTimeSchedule =
>>> context.schedule(PunctuationType.SYSTEM_TIME,
>>> systemTimeUpperBound, ||this||::systemTimePunctuate); |
>>> |}|
>>> | |
>>> |public| |void| |streamTimePunctuate(||long| |streamTime){|
>>> | ||periodicBusiness(streamTime);|
>>>
>>> | ||systemTimeSchedule.cancel();|
>>> | ||systemTimeSchedule =
>>> context.schedule(PunctuationType.SYSTEM_TIME,
>>> systemTimeUpperBound, ||this||::systemTimePunctuate);|
>>> |}|
>>> | |
>>> |public| |void| |systemTimePunctuate(||long| |systemTime){|
>>> | ||periodicBusiness(context.timestamp());|
>>>
>>> | ||streamTimeSchedule.cancel();|
>>> | ||streamTimeSchedule =
>>> context.schedule(PunctuationType.STREAM_TIME,
>>> streamTimeInterval, ||this||::streamTimePunctuate);|
>>> |}|
>>> | |
>>> |public| |void| |periodicBusiness(||long| |streamTime){|
>>> | ||// guard against streamTime == -1, easy enough.|
>>> | ||// if you need system time instead, just use
>>> System.currentTimeMillis()|
>>> | |
>>> | ||// do something businessy here|
>>> |}|
>>>
>>> Where Cancellable is either an interface containing just a single void
>>> cancel() method or also boolean isCancelled() like here
>>> <http://doc.akka.io/japi/akka/2.5.0/akka/actor/Cancellable.html>.
>>>
>>>
>>> Please let your opinions known whether we should proceed in this
>>> direction or leave "hybrid" considerations out of scope.
>>>
>>> Looking forward to hearing your thoughts.
>>>
>>> Thanks,
>>> Michal
>>>
>>> On 30/04/17 20:07, Michal Borowiecki wrote:
>>>> Hi Matthias,
>>>>
>>>> I'd like to start moving the discarded ideas into Rejected
>>>> Alternatives section. Before I do, I want to tidy them up, ensure
>>>> they've each been given proper treatment.
>>>>
>>>> To that end let me go back to one of your earlier comments about the
>>>> original suggestion (A) to put that to bed.
>>>>
>>>>
>>>> On 04/04/17 06:44, Matthias J. Sax wrote:
>>>>> (A) You argue, that users can still "punctuate" on event-time via
>>>>> process(), but I am not sure if this is possible. Note, that users only
>>>>> get record timestamps via context.timestamp(). Thus, users would need to
>>>>> track the time progress per partition (based on the partitions they
>>>>> obverse via context.partition(). (This alone puts a huge burden on the
>>>>> user by itself.) However, users are not notified at startup what
>>>>> partitions are assigned, and user are not notified when partitions get
>>>>> revoked. Because this information is not available, it's not possible to
>>>>> "manually advance" stream-time, and thus event-time punctuation within
>>>>> process() seems not to be possible -- or do you see a way to get it
>>>>> done? And even if, it might still be too clumsy to use.
>>>> I might have missed something but I'm guessing your worry about users
>>>> having to track time progress /per partition/ comes from the what the
>>>> stream-time does currently.
>>>> But I'm not sure that those semantics of stream-time are ideal for
>>>> users of punctuate.
>>>> That is, if stream-time punctuate didn't exist and users had to use
>>>> process(), would they actually want to use the current semantics of
>>>> stream time?
>>>>
>>>> As a reminder stream time, in all its glory, is (not exactly
>>>> actually, but when trying to be absolutely precise here I spotted
>>>> KAFKA-5144 <https://issues.apache.org/jira/browse/KAFKA-5144> so I
>>>> think this approximation suffices to illustrate the point):
>>>>
>>>> a minimum across all input partitions of (
>>>> if(msgs never received by partition) -1;
>>>> else {
>>>> a non-descending-minimum of ( the per-batch minimum msg timestamp)
>>>> }
>>>> )
>>>>
>>>> Would that really be clear enough to the users of punctuate? Do they
>>>> care for such a convoluted notion of time? I see how this can be
>>>> useful for StreamTask to pick the next partition to take a record
>>>> from but for punctuate?
>>>> If users had to implement punctuation with process(), is that what
>>>> they would have chosen as their notion of time?
>>>> I'd argue not.
>>>>
>>>> None of the processors implementing the rich windowing/join
>>>> operations in the DSL use punctuate.
>>>> Let's take the KStreamKStreamJoinProcessor as an example, in it's
>>>> process() method it simply uses context().timestamp(), which, since
>>>> it's called from process, returns simply, per javadoc:
>>>> If it is triggered while processing a record streamed from the source
>>>> processor, timestamp is defined as the timestamp of the current input
>>>> record;
>>>> So they don't use that convoluted formula for stream-time. Instead,
>>>> they only care about the timestamp of the current record. I think
>>>> that having users track just that wouldn't be that much of a burden.
>>>> I don't think they need to care about which partitions got assigned
>>>> or not. And StreamTask would still be picking records first from the
>>>> partition having the lowest timestamp to try to "synchronize" the
>>>> streams as it does now.
>>>>
>>>> What users would have to do in their Processor implementations is
>>>> somewhere along the lines of:
>>>>
>>>> long lastPunctuationTime = 0;
>>>> long interval = <some-number>; //millis
>>>>
>>>> @Override
>>>> public void process(K key, V value){
>>>> while (ctx.timestamp() >= lastPunctuationTime + interval){
>>>> punctuate(ctx.timestamp());
>>>> lastPunctuationTime += interval;// I'm not sure of the merit
>>>> of this vs lastPunctuationTime = ctx.timestamp(); but that's what
>>>> PunctuationQueue does currently
>>>> }
>>>> // do some other business logic here
>>>> }
>>>>
>>>> Looking forward to your thoughts.
>>>>
>>>> Cheers,
>>>> Michal
>>>>
>>>> --
>>>> Signature
>>>> <http://www.openbet.com/> Michal Borowiecki
>>>> Senior Software Engineer L4
>>>> T: +44 208 742 1600
>>>>
>>>>
>>>> +44 203 249 8448
>>>>
>>>>
>>>>
>>>> E: michal.borowiecki@openbet.com
>>>> W: www.openbet.com <http://www.openbet.com/>
>>>>
>>>>
>>>> OpenBet Ltd
>>>>
>>>> Chiswick Park Building 9
>>>>
>>>> 566 Chiswick High Rd
>>>>
>>>> London
>>>>
>>>> W4 5XT
>>>>
>>>> UK
>>>>
>>>>
>>>> <https://www.openbet.com/email_promo>
>>>>
>>>> This message is confidential and intended only for the addressee. If
>>>> you have received this message in error, please immediately notify
>>>> the postmaster@openbet.com <ma...@openbet.com> and delete
>>>> it from your system as well as any copies. The content of e-mails as
>>>> well as traffic data may be monitored by OpenBet for employment and
>>>> security purposes. To protect the environment please do not print
>>>> this e-mail unless necessary. OpenBet Ltd. Registered Office:
>>>> Chiswick Park Building 9, 566 Chiswick High Road, London, W4 5XT,
>>>> United Kingdom. A company registered in England and Wales. Registered
>>>> no. 3134634. VAT no. GB927523612
>>>>
>>> --
>>> Signature
>>> <http://www.openbet.com/> Michal Borowiecki
>>> Senior Software Engineer L4
>>> T: +44 208 742 1600
>>>
>>>
>>> +44 203 249 8448
>>>
>>>
>>>
>>> E: michal.borowiecki@openbet.com
>>> W: www.openbet.com <http://www.openbet.com/>
>>>
>>>
>>> OpenBet Ltd
>>>
>>> Chiswick Park Building 9
>>>
>>> 566 Chiswick High Rd
>>>
>>> London
>>>
>>> W4 5XT
>>>
>>> UK
>>>
>>>
>>> <https://www.openbet.com/email_promo>
>>>
>>> This message is confidential and intended only for the addressee. If
>>> you have received this message in error, please immediately notify the
>>> postmaster@openbet.com <ma...@openbet.com> and delete it
>>> from your system as well as any copies. The content of e-mails as well
>>> as traffic data may be monitored by OpenBet for employment and
>>> security purposes. To protect the environment please do not print this
>>> e-mail unless necessary. OpenBet Ltd. Registered Office: Chiswick Park
>>> Building 9, 566 Chiswick High Road, London, W4 5XT, United Kingdom. A
>>> company registered in England and Wales. Registered no. 3134634. VAT
>>> no. GB927523612
>>>
>> --
>> Signature
>> <http://www.openbet.com/> Michal Borowiecki
>> Senior Software Engineer L4
>> T: +44 208 742 1600
>>
>>
>> +44 203 249 8448
>>
>>
>>
>> E: michal.borowiecki@openbet.com
>> W: www.openbet.com <http://www.openbet.com/>
>>
>>
>> OpenBet Ltd
>>
>> Chiswick Park Building 9
>>
>> 566 Chiswick High Rd
>>
>> London
>>
>> W4 5XT
>>
>> UK
>>
>>
>> <https://www.openbet.com/email_promo>
>>
>> This message is confidential and intended only for the addressee. If you
>> have received this message in error, please immediately notify the
>> postmaster@openbet.com <ma...@openbet.com> and delete it
>> from your system as well as any copies. The content of e-mails as well
>> as traffic data may be monitored by OpenBet for employment and security
>> purposes. To protect the environment please do not print this e-mail
>> unless necessary. OpenBet Ltd. Registered Office: Chiswick Park Building
>> 9, 566 Chiswick High Road, London, W4 5XT, United Kingdom. A company
>> registered in England and Wales. Registered no. 3134634. VAT no.
>> GB927523612
>>
--
Signature
<http://www.openbet.com/> Michal Borowiecki
Senior Software Engineer L4
T: +44 208 742 1600
+44 203 249 8448
E: michal.borowiecki@openbet.com
W: www.openbet.com <http://www.openbet.com/>
OpenBet Ltd
Chiswick Park Building 9
566 Chiswick High Rd
London
W4 5XT
UK
<https://www.openbet.com/email_promo>
This message is confidential and intended only for the addressee. If you
have received this message in error, please immediately notify the
postmaster@openbet.com <ma...@openbet.com> and delete it
from your system as well as any copies. The content of e-mails as well
as traffic data may be monitored by OpenBet for employment and security
purposes. To protect the environment please do not print this e-mail
unless necessary. OpenBet Ltd. Registered Office: Chiswick Park Building
9, 566 Chiswick High Road, London, W4 5XT, United Kingdom. A company
registered in England and Wales. Registered no. 3134634. VAT no.
GB927523612