Thanks John. SGTM.
On 10/1/20 2:50 PM, John Roesler wrote: > Hello again, all, > > I'm sorry to make another tweak to this KIP, but during the > implementation of the design we've just agreed on, I > realized that Processors would almost never need to > reference the RecordMetadata. Therefore, I'm proposing to > streamline the API by moving the Optional<RecordMetadata> to > the new ProcessorContext as a method, rather than making it > a method argument to Processor#process. > > The change is visible here: > https://cwiki.apache.org/confluence/pages/diffpagesbyversion.action?pageId=118172121&selectedPageVersions=16&selectedPageVersions=15 > > All of the same semantics and considerations we discussed > still apply, it's just that Processor implementers won't > have to think about it unless they actually _need_ the > topic/partition/offset information from the RecordMetadata. > > Also, the PR for this part of the KIP is now available here: > https://github.com/apache/kafka/pull/9361 > > I know it's a bit on the heavy side; I've annotated the PR > to try and ease the reviewer's job. I'd greatly appreciate > it if anyone can take the time to review. > > Thanks, > -John > > On Wed, 2020-09-30 at 10:16 -0500, John Roesler wrote: >> Thanks, Matthias! >> >> I can certainly document it. I didn't bother because the old >> Processor, Supplier, and Context will themselves be >> deprecated, so any method that handles them won't be able to >> avoid the deprecation warning. Nevertheless, it doesn't hurt >> just to explicitly deprecated those methods. >> >> Thanks, >> -John >> >> On Wed, 2020-09-30 at 08:10 -0700, Matthias J. Sax wrote: >>> Thanks John. I like the proposal. >>> >>> Btw: I was just going over the KIP and realized that we add new methods >>> to `StreamBuilder`, `Topology`, and `KStream` that take the new >>> `ProcessorSupplier` class -- should we also deprecate the corresponding >>> existing ones that take the old `ProcessorSupplier`? >>> >>> >>> -Matthias >>> >>> >>> On 9/30/20 7:46 AM, John Roesler wrote: >>>> Thanks Paul and Sophie, >>>> >>>> Your feedback certainly underscores the need to be explicit >>>> in the javadoc about why that parameter is Optional. Getting >>>> this kind of feedback before the release is exactly the kind >>>> of outcome we hope to get from the KIP process! >>>> >>>> Thanks, >>>> -John >>>> >>>> On Tue, 2020-09-29 at 22:32 -0500, Paul Whalen wrote: >>>>> John, I totally agree that adding a method to Processor is cumbersome and >>>>> not a good path. I was imagining maybe a separate interface that could be >>>>> used in the appropriate context, but I don't think that makes too much >>>>> sense - it's just too far away from what Kafka Streams is. I was >>>>> originally more interested in the "why" Optional than the "how" (I think >>>>> my >>>>> original reply overplayed the "optional as an argument" concern). But >>>>> you've convinced me that there is a perfectly legitimate "why". We should >>>>> make sure that it's clear why it's Optional, but I suppose that goes >>>>> without saying. It's a nice opportunity to make the API reflect more what >>>>> is actually going on under the hood. >>>>> >>>>> Thanks! >>>>> Paul >>>>> >>>>> On Tue, Sep 29, 2020 at 10:05 PM Sophie Blee-Goldman <sop...@confluent.io> >>>>> wrote: >>>>> >>>>>> FWIW, while I'm really not a fan of Optional in general, I agree that its >>>>>> usage >>>>>> here seems appropriate. Even for those rare software developers who >>>>>> carefully >>>>>> read all the docs several times over, I think it wouldn't be too hard to >>>>>> miss a >>>>>> note about the RecordMetadata possibly being null. >>>>>> >>>>>> Especially because it's not that obvious why at first glance, and takes a >>>>>> bit of >>>>>> thinking to realize that records originating from a Punctuator wouldn't >>>>>> have a >>>>>> "current record". This is something that has definitely confused users >>>>>> today. >>>>>> >>>>>> It's on us to improve the education here -- and an >>>>>> Optional<RecordMetadata> >>>>>> would naturally raise awareness of this subtlety >>>>>> >>>>>> On Tue, Sep 29, 2020 at 7:40 PM Sophie Blee-Goldman <sop...@confluent.io> >>>>>> wrote: >>>>>> >>>>>>> Does my reply address your concerns? >>>>>>> >>>>>>> >>>>>>> Yes; also, I definitely misread part of the proposal earlier and thought >>>>>>> you had put >>>>>>> the timestamp field in RecordMetadata. Sorry for not giving things a >>>>>>> closer look >>>>>>> before responding! I'm not sure my original message made much sense >>>>>>> given >>>>>>> the misunderstanding, but thanks for responding anyway :P >>>>>>> >>>>>>> Having given the proposal a second pass, I agree, it's very elegant. +1 >>>>>>> >>>>>>> On Tue, Sep 29, 2020 at 6:50 PM John Roesler <vvcep...@apache.org> >>>>>> wrote: >>>>>>>> Thanks for the reply, Sophie, >>>>>>>> >>>>>>>> I think I may have summarized too much in my prior reply. >>>>>>>> >>>>>>>> In the currently proposed KIP, any caller of forward() must >>>>>>>> supply a Record, which consists of: >>>>>>>> * key >>>>>>>> * value >>>>>>>> * timestamp >>>>>>>> * headers (with a convenience constructor that sets empty >>>>>>>> headers) >>>>>>>> >>>>>>>> These aren't what I was referring to as potentially being >>>>>>>> undefined downstream, since thanks to the introduction of >>>>>>>> Record, they are, as you're advocating, required to be >>>>>>>> defined everywhere, even when forwarding from a punctuator. >>>>>>>> >>>>>>>> So to be clear, the intent of this change is actually to >>>>>>>> _enforce_ that timestamp would never be undefined (which it >>>>>>>> currently can be). Also, since punctuators _are_ going to >>>>>>>> have to "make up" a timestamp going forward, we should note >>>>>>>> that the "punctuate" method currently passes in a good >>>>>>>> timestamp that they can use: for system-time punctuations, >>>>>>>> they receive the current system time, and for stream-time >>>>>>>> punctuations, they get the current stream time. >>>>>>>> >>>>>>>> The potentially undefined RecordMetadata only contains these >>>>>>>> fields: >>>>>>>> * topic >>>>>>>> * partition >>>>>>>> * offset >>>>>>>> >>>>>>>> These fields aren't required (or even used) in a Sink, and >>>>>>>> it doesn't seem like they would be important to many >>>>>>>> applications. Furthermore, it doesn't _seem_ like you'd even >>>>>>>> want to set these fields. They seem purely informational and >>>>>>>> only useful in the context when you are actually processing >>>>>>>> a real input record. It doesn't sound like you were asking >>>>>>>> for it, but just to put it on the record, I think if we were >>>>>>>> to require values for the metadata from punctuators, people >>>>>>>> would mostly just make up their own dummy values, to no >>>>>>>> one's benefit. >>>>>>>> >>>>>>>> I should also note that with the current >>>>>>>> Record/RecordMetadata split, we will have the freedom to >>>>>>>> move fields into the Record class (or even add new fields) >>>>>>>> if we want them to become "data" as opposed to "metadata" in >>>>>>>> the future. >>>>>>>> >>>>>>>> Thanks for your reply; I was similarly floored when I >>>>>>>> realized the true nature of the current situation. Does my >>>>>>>> reply address your concerns? >>>>>>>> >>>>>>>> Thanks, >>>>>>>> -John >>>>>>>> >>>>>>>> On Tue, 2020-09-29 at 18:34 -0700, Sophie Blee-Goldman >>>>>>>> wrote: >>>>>>>>>> However, the record metadata is only defined when the parent >>>>>> forwards >>>>>>>>>> while processing a >>>>>>>>> >>>>>>>>> real record, not when it calls forward from the punctuator >>>>>>>>> >>>>>>>>> >>>>>>>>> Can we take a step back for a second...why wouldn't you be required to >>>>>>>> set >>>>>>>>> the RecordContext >>>>>>>>> yourself when calling forward from a Punctuator? I think I agree with >>>>>>>> Paul >>>>>>>>> here, it seems kind of >>>>>>>>> absurd not to enforce that the RecordContext be present inside the >>>>>>>>> process() method. >>>>>>>>> >>>>>>>>> The original problem with Punctuators, as I understood it, was that >>>>>> all >>>>>>>> of >>>>>>>>> the RecordContext >>>>>>>>> fields were exposed automatically to both the Processor and any >>>>>>>> Punctuator, >>>>>>>>> due to being >>>>>>>>> direct methods on the ProcessorContext. We can't control which >>>>>>>>> ProcessorContext methods >>>>>>>>> someone will call from with a Punctuator vs from a Processor. The best >>>>>>>> we >>>>>>>>> could do was >>>>>>>>> set these "nonsense" fields to null when inside a Punctuator, or set >>>>>>>> them >>>>>>>>> to some dummy >>>>>>>>> values as you pointed out. >>>>>>>>> >>>>>>>>> But then you proposed the solution of a separate RecordContext which >>>>>> is >>>>>>>> not >>>>>>>>> attached to the >>>>>>>>> ProcessorContext at all. This seemed to solve the above problem very >>>>>>>>> neatly: we only pass >>>>>>>>> in the RecordContext to the process() method, so we don't have to >>>>>> worry >>>>>>>>> about people trying >>>>>>>>> to access these fields from within a Punctuator. The fields aren't >>>>>>>>> accessible unless they're >>>>>>>>> defined. >>>>>>>>> >>>>>>>>> So what happens when someone wants to forward something from within a >>>>>>>>> Punctuator? I >>>>>>>>> don't think it's reasonable to let the timestamp field be undefined, >>>>>>>> ever. >>>>>>>>> What if the Punctuator >>>>>>>>> forwards directly to a sink, or directly to some windowing logic. Are >>>>>> we >>>>>>>>> supposed to add >>>>>>>>> handling for the RecordContext == null case to every processor? Or are >>>>>>>> we >>>>>>>>> just going to >>>>>>>>> assume the implicit restriction that users will only forward records >>>>>>>> from a >>>>>>>>> Punctuator to >>>>>>>>> downstream processors that know how to handle and/or set the >>>>>>>> RecordContext >>>>>>>>> if it's >>>>>>>>> undefined. That seems to throw away a lot of the awesome safety added >>>>>> in >>>>>>>>> this KIP >>>>>>>>> >>>>>>>>> Apologies for the rant. But I feel pretty strongly that allowing to >>>>>>>> forward >>>>>>>>> records from a >>>>>>>>> Punctuator without a defined RecordContext would be asking for >>>>>> trouble. >>>>>>>>> Imo, if you >>>>>>>>> want to forward from a Punctuator, you need to store the info you need >>>>>>>> in >>>>>>>>> order to >>>>>>>>> set the timestamp, or make one up yourself >>>>>>>>> >>>>>>>>> (the one alternative I can think of here is that maybe we could pass >>>>>> in >>>>>>>> the >>>>>>>>> current >>>>>>>>> partition time, so users can at least put in a reasonable estimate for >>>>>>>> the >>>>>>>>> timestamp >>>>>>>>> that won't cause it to get dropped and won't potentially lurch the >>>>>>>>> streamtime far into >>>>>>>>> the future. This would be similar to what we do in the >>>>>>>> TimestampExtractor) >>>>>>>>> On Tue, Sep 29, 2020 at 6:06 PM John Roesler <vvcep...@apache.org> >>>>>>>> wrote: >>>>>>>>>> Oh, I guess one other thing I should have mentioned is that I’ve >>>>>>>> recently >>>>>>>>>> discovered that in cases where the context is undefined, we >>>>>> currently >>>>>>>> just >>>>>>>>>> fill in dummy values for the context. So there’s a good chance that >>>>>>>> real >>>>>>>>>> applications in use are depending on undefined context without even >>>>>>>>>> realizing it. What I’m hoping to do is just make the situation >>>>>>>> explicit and >>>>>>>>>> get rid of the dummy values. >>>>>>>>>> >>>>>>>>>> Thanks, >>>>>>>>>> John >>>>>>>>>> >>>>>>>>>> On Tue, Sep 29, 2020, at 20:01, John Roesler wrote: >>>>>>>>>>> Thanks for the review, Paul! >>>>>>>>>>> >>>>>>>>>>> I had read some of that debate before. There seems to be some >>>>>>>> subtext >>>>>>>>>>> there, because they advise against using Optional in cases like >>>>>>>> this, >>>>>>>>>>> but there doesn’t seem to be a specific reason why it’s >>>>>>>> inappropriate. >>>>>>>>>>> I got the impression they were just afraid that people would go >>>>>>>>>>> overboard and make everything Optional. >>>>>>>>>>> >>>>>>>>>>> I could also make two methods, but it seemed like it might be an >>>>>>>>>>> unfortunate way to handle the issue, since Processor is just >>>>>> about a >>>>>>>>>>> Function as-is, but the two-method approach would require people >>>>>> to >>>>>>>>>>> implement both methods. >>>>>>>>>>> >>>>>>>>>>> To your question, this is something that’s only recently became >>>>>>>> clear >>>>>>>>>>> to me. Imagine you have a parent processor that calls forward both >>>>>>>> from >>>>>>>>>>> process and a punctuator. The child will have process() invoked in >>>>>>>> both >>>>>>>>>>> cases, and won’t be able to distinguish them. However, the record >>>>>>>>>>> metadata is only defined when the parent forwards while >>>>>> processing a >>>>>>>>>>> real record, not when it calls forward from the punctuator. >>>>>>>>>>> >>>>>>>>>>> This is why I wanted to make the metadata Optional, to advertise >>>>>>>> that >>>>>>>>>>> the metadata might be undefined if any ancestor processor ever >>>>>> calls >>>>>>>>>>> forward from a punctuator. We could remove the Optional and >>>>>> instead >>>>>>>>>>> just document that the argument might be null. >>>>>>>>>>> >>>>>>>>>>> With that context in place, what’s your take? >>>>>>>>>>> >>>>>>>>>>> Thanks, >>>>>>>>>>> John >>>>>>>>>>> >>>>>>>>>>> On Tue, Sep 29, 2020, at 19:09, Paul Whalen wrote: >>>>>>>>>>>> Looks pretty good to me, though the Processor#process(Record, >>>>>>>>>>>> Optional<RecordMetadata>) signature caught my eye. There's some >>>>>>>>>> debate >>>>>>>>>>>> ( >>>>>>>>>>>> >>>>>> https://stackoverflow.com/questions/31922866/why-should-java-8s-optional-not-be-used-in-arguments >>>>>>>>>> ) >>>>>>>>>>>> about whether to use Optionals in arguments, and while that's a >>>>>>>> bit of >>>>>>>>>> a >>>>>>>>>>>> religious debate in the abstract, it did make me wonder whether >>>>>> it >>>>>>>>>> makes >>>>>>>>>>>> sense in this specific case. When is it actually not present? >>>>>> I >>>>>>>> was >>>>>>>>>>>> under >>>>>>>>>>>> the impression that we should always have access to it in >>>>>>>> process(), >>>>>>>>>> and >>>>>>>>>>>> that the concern about metadata being undefined was about having >>>>>>>>>> access >>>>>>>>>>>> to >>>>>>>>>>>> record metadata in the ProcessorContext held for use inside a >>>>>>>>>>>> Punctuator. >>>>>>>>>>>> >>>>>>>>>>>> If that's not the case and it is truly optional in process(), is >>>>>>>> there >>>>>>>>>> an >>>>>>>>>>>> opportunity for an alternate interface for the cases when we >>>>>>>> don't get >>>>>>>>>> it, >>>>>>>>>>>> rather than force the branching on implementers of the >>>>>> interface? >>>>>>>>>>>> Apologies if I've missed something, I took a look at the PR and >>>>>> I >>>>>>>>>> didn't >>>>>>>>>>>> see any spots where I thought it would be empty. Perhaps an >>>>>>>> example >>>>>>>>>> of a >>>>>>>>>>>> Punctuator using (and not using) the new API would clear things >>>>>>>> up. >>>>>>>>>>>> Best, >>>>>>>>>>>> Paul >>>>>>>>>>>> >>>>>>>>>>>> On Tue, Sep 29, 2020 at 4:10 PM John Roesler < >>>>>> vvcep...@apache.org >>>>>>>>>> wrote: >>>>>>>>>>>>> Hello again, all, >>>>>>>>>>>>> >>>>>>>>>>>>> Thanks for the latest round of discussion. I've taken the >>>>>>>>>>>>> recent feedback and come up with an updated KIP that seems >>>>>>>>>>>>> actually quite a bit nicer than the prior proposal. >>>>>>>>>>>>> >>>>>>>>>>>>> The specific diff on the KIP is here: >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>> https://cwiki.apache.org/confluence/pages/diffpagesbyversion.action?pageId=118172121&selectedPageVersions=15&selectedPageVersions=14 >>>>>>>>>>>>> These changes are implemented in this POC PR: >>>>>>>>>>>>> https://github.com/apache/kafka/pull/9346 >>>>>>>>>>>>> >>>>>>>>>>>>> The basic idea is that, building on the recent conversaion, >>>>>>>>>>>>> we would transition away from the current API where we get >>>>>>>>>>>>> only key/value in the process() method and other "data" >>>>>>>>>>>>> comes in the ProcessorContext along with the "metadata". >>>>>>>>>>>>> >>>>>>>>>>>>> Instead, we formalize what is "data" and what is "metadata", >>>>>>>>>>>>> and pass it all in to the process method: >>>>>>>>>>>>> Processor#process(Record, Optional<RecordMetadata>) >>>>>>>>>>>>> >>>>>>>>>>>>> Also, you forward the whole data class instead of mutating >>>>>>>>>>>>> the ProcessorContext fields and also calling forward: >>>>>>>>>>>>> ProcessorContext#forward(Record) >>>>>>>>>>>>> >>>>>>>>>>>>> The Record class itself ships with methods like >>>>>>>>>>>>> record#withValue(NewV newValue) >>>>>>>>>>>>> that make a shallow copy of the input Record, enabling >>>>>>>>>>>>> Processors to safely handle the record without polluting the >>>>>>>>>>>>> context of their parents and siblings. >>>>>>>>>>>>> >>>>>>>>>>>>> This proposal has a number of key benefits: >>>>>>>>>>>>> * As we've discovered in KAFKA-9584, it's unsafe to mutate >>>>>>>>>>>>> the Headers via the ProcessorContext. This proposal offers a >>>>>>>>>>>>> way to safely forward changes only to downstream processors. >>>>>>>>>>>>> * The new API has symmetry (each processor's input is the >>>>>>>>>>>>> output of its parent processor) >>>>>>>>>>>>> * The API makes clear that the record metadata isn't always >>>>>>>>>>>>> defined (for example, in a punctuation, there is no current >>>>>>>>>>>>> topic/partition/offset) >>>>>>>>>>>>> * The API enables punctuators to forward well defined >>>>>>>>>>>>> headers downstream, which is currently not possible. >>>>>>>>>>>>> >>>>>>>>>>>>> Unless their are objections, I'll go ahead and re-finalize >>>>>>>>>>>>> this KIP and update that PR to a mergeable state. >>>>>>>>>>>>> >>>>>>>>>>>>> Thanks, all, >>>>>>>>>>>>> -John >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> On Thu, 2020-09-24 at 09:41 -0700, Matthias J. Sax wrote: >>>>>>>>>>>>>> Interesting proposal. However, I am not totally convinced, >>>>>>>> because >>>>>>>>>> I see >>>>>>>>>>>>>> a fundamental difference between "data" and "metadata". >>>>>>>>>>>>>> >>>>>>>>>>>>>> Topic/partition/offset are "metadata" in the strong sense >>>>>> and >>>>>>>> they >>>>>>>>>> are >>>>>>>>>>>>>> immutable. >>>>>>>>>>>>>> >>>>>>>>>>>>>> On the other hand there is "primary" data like key and >>>>>> value, >>>>>>>> as >>>>>>>>>> well as >>>>>>>>>>>>>> "secondary" data like timestamp and headers. The issue seems >>>>>>>> that >>>>>>>>>> we >>>>>>>>>>>>>> treat "secondary data" more like metadata atm? >>>>>>>>>>>>>> >>>>>>>>>>>>>> Thus, promoting timestamp and headers into a first class >>>>>>>> citizen >>>>>>>>>> roll >>>>>>>>>>>>>> make sense to me (my original proposal about `RecordContext` >>>>>>>> would >>>>>>>>>> still >>>>>>>>>>>>>> fall short with this regard). However, putting both (data >>>>>> and >>>>>>>>>> metadata) >>>>>>>>>>>>>> into a `Record` abstraction might go too far? >>>>>>>>>>>>>> >>>>>>>>>>>>>> I am also a little bit concerned about `Record.copy()` >>>>>>>> because it >>>>>>>>>> might >>>>>>>>>>>>>> be a trap: Users might assume it does a full deep copy of >>>>>> the >>>>>>>>>> record, >>>>>>>>>>>>>> however, it would not. It would only create a new `Record` >>>>>>>> object >>>>>>>>>> as >>>>>>>>>>>>>> wrapper that points to the same key/value/header objects as >>>>>>>> the >>>>>>>>>> input >>>>>>>>>>>>>> record. >>>>>>>>>>>>>> >>>>>>>>>>>>>> With the current `context.forward(key, value)` we don't have >>>>>>>> this >>>>>>>>>> "deep >>>>>>>>>>>>>> copy" issue -- it's pretty clear what is happening. >>>>>>>>>>>>>> >>>>>>>>>>>>>> Instead of `To.all().withTimestamp()` we could also add >>>>>>>>>>>>>> `context.forward(key, value, timestamp)` etc (just wondering >>>>>>>> about >>>>>>>>>> the >>>>>>>>>>>>>> exposition in overload)? >>>>>>>>>>>>>> >>>>>>>>>>>>>> Also, `Record.withValue` etc sounds odd? Should a record not >>>>>>>> be >>>>>>>>>>>>>> immutable? So, we could have something like >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>> `RecordFactory.withKeyValue(...).withTimestamp(...).withHeaders(...).build()`. >>>>>>>>>>>>>> But it looks rather verbose? >>>>>>>>>>>>>> >>>>>>>>>>>>>> The other question is of course, to what extend to we want >>>>>> to >>>>>>>> keep >>>>>>>>>> the >>>>>>>>>>>>>> distinction between "primary" and "secondary" data? To me, >>>>>>>> it's a >>>>>>>>>>>>>> question of easy of use? >>>>>>>>>>>>>> >>>>>>>>>>>>>> Just putting all this out to move the discussion forward. >>>>>>>> Don't >>>>>>>>>> have a >>>>>>>>>>>>>> concrete proposal atm. >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> -Matthias >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> On 9/14/20 9:24 AM, John Roesler wrote: >>>>>>>>>>>>>>> Thanks for this thought, Matthias! >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> To be honest, it's bugged me quite a bit that _all_ the >>>>>>>>>>>>>>> record information hasn't been an argument to `process`. I >>>>>>>>>>>>>>> suppose I was trying to be conservative in this proposal, >>>>>>>>>>>>>>> but then again, if we're adding new Processor and >>>>>>>>>>>>>>> ProcessorContext interfaces, then this is the time to make >>>>>>>>>>>>>>> such a change. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> To be unambiguous, I think this is what we're talking >>>>>> about: >>>>>>>>>>>>>>> ProcessorContext: >>>>>>>>>>>>>>> * applicationId >>>>>>>>>>>>>>> * taskId >>>>>>>>>>>>>>> * appConfigs >>>>>>>>>>>>>>> * appConfigsWithPrefix >>>>>>>>>>>>>>> * keySerde >>>>>>>>>>>>>>> * valueSerde >>>>>>>>>>>>>>> * stateDir >>>>>>>>>>>>>>> * metrics >>>>>>>>>>>>>>> * schedule >>>>>>>>>>>>>>> * commit >>>>>>>>>>>>>>> * forward >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> StateStoreContext: >>>>>>>>>>>>>>> * applicationId >>>>>>>>>>>>>>> * taskId >>>>>>>>>>>>>>> * appConfigs >>>>>>>>>>>>>>> * appConfigsWithPrefix >>>>>>>>>>>>>>> * keySerde >>>>>>>>>>>>>>> * valueSerde >>>>>>>>>>>>>>> * stateDir >>>>>>>>>>>>>>> * metrics >>>>>>>>>>>>>>> * register >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> RecordContext >>>>>>>>>>>>>>> * topic >>>>>>>>>>>>>>> * partition >>>>>>>>>>>>>>> * offset >>>>>>>>>>>>>>> * timestamp >>>>>>>>>>>>>>> * headers >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Your proposal sounds good to me as-is. Just to cover the >>>>>>>>>>>>>>> bases, though, I'm wondering if we should push the idea >>>>>> just >>>>>>>>>>>>>>> a little farther. Instead of decomposing >>>>>> key,value,context, >>>>>>>>>>>>>>> we could just keep them all in one object, like this: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Record: >>>>>>>>>>>>>>> * key >>>>>>>>>>>>>>> * value >>>>>>>>>>>>>>> * topic >>>>>>>>>>>>>>> * partition >>>>>>>>>>>>>>> * offset >>>>>>>>>>>>>>> * timestamp >>>>>>>>>>>>>>> * headers >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Then, we could have: >>>>>>>>>>>>>>> Processor#process(Record) >>>>>>>>>>>>>>> ProcessorContext#forward(Record, To) >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Viewed from this perspective, a record has three >>>>>> properties >>>>>>>>>>>>>>> that people may specify in their processors: key, value, >>>>>> and >>>>>>>>>>>>>>> timestamp. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> We could deprecate `To#withTimestamp` and enable people to >>>>>>>>>>>>>>> specify the timestamp along with the key and value when >>>>>> they >>>>>>>>>>>>>>> forward a record. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> E.g., >>>>>>>>>>>>>>> RecordBuilder toForward = RecordBuilder.copy(record) >>>>>>>>>>>>>>> toForward.withKey(newKey) >>>>>>>>>>>>>>> toForward.withValue(newValue) >>>>>>>>>>>>>>> toForward.withTimestamp(newTimestamp) >>>>>>>>>>>>>>> Record newRecord = toForward.build() >>>>>>>>>>>>>>> context.forward(newRecord, To.child("child1")) >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> Or, the more compact common case: >>>>>>>>>>>>>>> current: >>>>>>>>>>>>>>> context.forward(key, "newValue") >>>>>>>>>>>>>>> proposed: >>>>>>>>>>>>>>> >>>>>> context.forward(copy(record).withValue("newValue").build()) >>>>>>>>>>>>>>> It's slightly more verbose, but also more extensible. This >>>>>>>>>>>>>>> would give us a clean path to add header support in PAPI >>>>>> as >>>>>>>>>>>>>>> well, simply by adding `withHeaders` in RecordBuilder. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> It's also more symmetrical, since the recipient of >>>>>> `forward` >>>>>>>>>>>>>>> would just get the sent `Record`. Whereas today, the >>>>>> sender >>>>>>>>>>>>>>> puts the timestamp in `To`, but the recipient gets in in >>>>>> its >>>>>>>>>>>>>>> own `ProcessorContext`. >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> WDYT? >>>>>>>>>>>>>>> -John >>>>>>>>>>>>>>> >>>>>>>>>>>>>>> On Fri, 2020-09-11 at 12:30 -0700, Matthias J. Sax wrote: >>>>>>>>>>>>>>>> I think separating the different contexts make sense. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> In fact, we could even go one step further and remove >>>>>> the >>>>>>>>>> record >>>>>>>>>>>>> context >>>>>>>>>>>>>>>> from the processor context completely and we add a third >>>>>>>>>> parameter to >>>>>>>>>>>>>>>> `process(key, value, recordContext)`. This would make it >>>>>>>> clear >>>>>>>>>> that >>>>>>>>>>>>> the >>>>>>>>>>>>>>>> context is for the input record only and it's not >>>>>>>> possible to >>>>>>>>>> pass >>>>>>>>>>>>> it to >>>>>>>>>>>>>>>> a `punctuate` callback. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> For the stores and changelogging: I think there are two >>>>>>>> cases. >>>>>>>>>> (1) >>>>>>>>>>>>> You >>>>>>>>>>>>>>>> use a plain key-value store. For this case, it seems you >>>>>>>> do >>>>>>>>>> not care >>>>>>>>>>>>>>>> about the timestamp and thus does not care what >>>>>> timestamp >>>>>>>> is >>>>>>>>>> set in >>>>>>>>>>>>> the >>>>>>>>>>>>>>>> changelog records. (We can set anything we want, as it's >>>>>>>> not >>>>>>>>>>>>> relevant at >>>>>>>>>>>>>>>> all -- the timestamp is ignored on read anyway.) (2) The >>>>>>>> other >>>>>>>>>> case >>>>>>>>>>>>> is, >>>>>>>>>>>>>>>> that one does care about timestamps, and for this case >>>>>>>> should >>>>>>>>>> use >>>>>>>>>>>>>>>> TimestampedKeyValueStore. The passed timestamp will be >>>>>>>> set on >>>>>>>>>> the >>>>>>>>>>>>>>>> changelog records for this case. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> Thus, for both cases, accessing the record context does >>>>>>>> not >>>>>>>>>> seems to >>>>>>>>>>>>> be >>>>>>>>>>>>>>>> a requirement. And providing access to the processor >>>>>>>> context >>>>>>>>>> to, eg., >>>>>>>>>>>>>>>> `forward()` or similar seems safe. >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> -Matthias >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> On 9/10/20 7:25 PM, John Roesler wrote: >>>>>>>>>>>>>>>>> Thanks for the reply, Paul! >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> I certainly intend to make sure that the changelogging >>>>>>>> layer >>>>>>>>>>>>>>>>> continues to work the way it does now, by hook or by >>>>>>>> crook. >>>>>>>>>>>>>>>>> I think the easiest path for me is to just "cheat" and >>>>>>>> get >>>>>>>>>>>>>>>>> the real ProcessorContext into the ChangeLoggingStore >>>>>>>>>>>>>>>>> implementation somehow. I'll tag you on the PR when I >>>>>>>> create >>>>>>>>>>>>>>>>> it, so you have an opportunity to express a preference >>>>>>>> about >>>>>>>>>>>>>>>>> the implementation choice, and maybe even compile/test >>>>>>>>>>>>>>>>> against it to make sure your stuff still works. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Regarding this: >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> we have an interest in making a state store with a >>>>>>>> richer >>>>>>>>>>>>>>>>>> way of querying its data (like perhaps getting all >>>>>>>> values >>>>>>>>>>>>>>>>>> associated with a secondary key), while still >>>>>>>> ultimately >>>>>>>>>>>>>>>>>> writing to the changelog topic for later >>>>>> restoration. >>>>>>>>>>>>>>>>> This is very intriguing to me. On the side, I've been >>>>>>>>>>>>>>>>> preparing a couple of ideas related to this topic. I >>>>>>>> don't >>>>>>>>>>>>>>>>> think I have a coherent enough thought to even express >>>>>>>> it in >>>>>>>>>>>>>>>>> a Jira right now, but when I do, I'll tag you on it >>>>>>>> also to >>>>>>>>>>>>>>>>> see what you think. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> Whenever you're ready to share the usability >>>>>> improvement >>>>>>>>>>>>>>>>> ideas, I'm very interested to see what you've come up >>>>>>>> with. >>>>>>>>>>>>>>>>> Thanks, >>>>>>>>>>>>>>>>> -John >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> On Thu, 2020-09-10 at 21:02 -0500, Paul Whalen wrote: >>>>>>>>>>>>>>>>>>> when you use a HashMap or RocksDB or other "state >>>>>>>>>> stores", you >>>>>>>>>>>>> don't >>>>>>>>>>>>>>>>>>> expect them to automatically know extra stuff >>>>>> about >>>>>>>> the >>>>>>>>>> record >>>>>>>>>>>>> you're >>>>>>>>>>>>>>>>>>> storing. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> So, I don't think there is any reason we *can't* >>>>>>>> retain the >>>>>>>>>>>>> record context >>>>>>>>>>>>>>>>>>> in the StateStoreContext, and if any users came >>>>>>>> along >>>>>>>>>> with a >>>>>>>>>>>>> clear use case >>>>>>>>>>>>>>>>>>> I'd find that convincing. >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> I agree with the principle of being conservative >>>>>> with >>>>>>>> the >>>>>>>>>>>>> StateStoreContext >>>>>>>>>>>>>>>>>> API. Regarding user expectations or a clear use >>>>>>>> case, the >>>>>>>>>> only >>>>>>>>>>>>>>>>>> counterpoint I would offer is that we sort of have >>>>>>>> that >>>>>>>>>> use case >>>>>>>>>>>>> already, >>>>>>>>>>>>>>>>>> which is the example I gave of the change logging >>>>>>>> store >>>>>>>>>> using the >>>>>>>>>>>>>>>>>> timestamp. I am curious if this functionality will >>>>>> be >>>>>>>>>> retained >>>>>>>>>>>>> when using >>>>>>>>>>>>>>>>>> built in state stores, or will a low-level processor >>>>>>>> get a >>>>>>>>>>>>> KeyValueStore >>>>>>>>>>>>>>>>>> that no longer writes to the changelog topic with >>>>>> the >>>>>>>>>> record's >>>>>>>>>>>>> timestamp. >>>>>>>>>>>>>>>>>> While I personally don't care much about that >>>>>>>> functionality >>>>>>>>>>>>> specifically, I >>>>>>>>>>>>>>>>>> have a general desire for custom state stores to >>>>>>>> easily do >>>>>>>>>> the >>>>>>>>>>>>> things that >>>>>>>>>>>>>>>>>> built in state stores do. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> It genuinely did not occur to me that users might be >>>>>>>>>> looking up >>>>>>>>>>>>> and/or >>>>>>>>>>>>>>>>>>> updating records of other keys from within a >>>>>>>> Processor. >>>>>>>>>>>>>>>>>> I'm glad you said this Sophie, because it gives me >>>>>> an >>>>>>>>>>>>> opportunity to say >>>>>>>>>>>>>>>>>> that this is actually a *huge* use case for my team. >>>>>>>> The >>>>>>>>>> state >>>>>>>>>>>>> store >>>>>>>>>>>>>>>>>> usability improvements I was referring to in my >>>>>>>> previous >>>>>>>>>> message >>>>>>>>>>>>> were about >>>>>>>>>>>>>>>>>> enabling the user to write custom stores while still >>>>>>>> easily >>>>>>>>>>>>> hooking into >>>>>>>>>>>>>>>>>> the ability to write to a changelog topic. I think >>>>>>>> that is >>>>>>>>>>>>> technically >>>>>>>>>>>>>>>>>> possible now, but I don't think it's trivial. >>>>>>>>>> Specifically, we >>>>>>>>>>>>> have an >>>>>>>>>>>>>>>>>> interest in making a state store with a richer way >>>>>> of >>>>>>>>>> querying >>>>>>>>>>>>> its data >>>>>>>>>>>>>>>>>> (like perhaps getting all values associated with a >>>>>>>>>> secondary >>>>>>>>>>>>> key), while >>>>>>>>>>>>>>>>>> still ultimately writing to the changelog topic for >>>>>>>> later >>>>>>>>>>>>> restoration. >>>>>>>>>>>>>>>>>> We recognize that this use case throws away some of >>>>>>>> what >>>>>>>>>> kafka >>>>>>>>>>>>> streams >>>>>>>>>>>>>>>>>> (especially the DSL) is good at - easy >>>>>>>> parallelizability by >>>>>>>>>>>>> partitioning >>>>>>>>>>>>>>>>>> all processing by key - and that our business logic >>>>>>>> would >>>>>>>>>>>>> completely fall >>>>>>>>>>>>>>>>>> apart if we were consuming from multi-partition >>>>>>>> topics with >>>>>>>>>>>>> multiple >>>>>>>>>>>>>>>>>> consumers. But we have found that using the low >>>>>> level >>>>>>>>>> processor >>>>>>>>>>>>> API is >>>>>>>>>>>>>>>>>> good for the very simple stream processing >>>>>> primitives >>>>>>>> it >>>>>>>>>>>>> provides: handling >>>>>>>>>>>>>>>>>> the plumbing of consuming from multiple kafka topics >>>>>>>> and >>>>>>>>>>>>> potentially >>>>>>>>>>>>>>>>>> updating persistent local state in a reliable way. >>>>>>>> That in >>>>>>>>>>>>> itself has >>>>>>>>>>>>>>>>>> proven to be a worthwhile programming model. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Since I got off track a bit, let me summarize: I >>>>>> don't >>>>>>>>>>>>> particularly care >>>>>>>>>>>>>>>>>> about the record context being available to state >>>>>>>> store >>>>>>>>>>>>> implementations, >>>>>>>>>>>>>>>>>> and I think this KIP is headed in the right >>>>>> direction >>>>>>>> in >>>>>>>>>> that >>>>>>>>>>>>> regard. But >>>>>>>>>>>>>>>>>> more generally, I wanted to express the importance >>>>>> of >>>>>>>>>>>>> maintaining a >>>>>>>>>>>>>>>>>> powerful and flexible StateStore interface. >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Thanks! >>>>>>>>>>>>>>>>>> Paul >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> On Thu, Sep 10, 2020 at 6:11 PM Sophie Blee-Goldman >>>>>> < >>>>>>>>>>>>> sop...@confluent.io> >>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> Aha, I did misinterpret the example in your >>>>>> previous >>>>>>>>>> response >>>>>>>>>>>>> regarding the >>>>>>>>>>>>>>>>>>> range query after all. I thought you just meant a >>>>>>>>>> time-range >>>>>>>>>>>>> query inside a >>>>>>>>>>>>>>>>>>> punctuator. It genuinely did not occur to me that >>>>>>>> users >>>>>>>>>> might >>>>>>>>>>>>> be looking up >>>>>>>>>>>>>>>>>>> and/or updating records of other keys from within >>>>>> a >>>>>>>>>> Processor. >>>>>>>>>>>>> Sorry for >>>>>>>>>>>>>>>>>>> being closed minded >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> I won't drag out this discussion any further by >>>>>>>> asking >>>>>>>>>> whether >>>>>>>>>>>>> that might >>>>>>>>>>>>>>>>>>> be >>>>>>>>>>>>>>>>>>> a valid use case or just a lurking bug in itself >>>>>> :) >>>>>>>>>>>>>>>>>>> Thanks for humoring me. The current proposal for >>>>>>>> KIP-478 >>>>>>>>>>>>> sounds good to me >>>>>>>>>>>>>>>>>>> On Thu, Sep 10, 2020 at 3:43 PM John Roesler < >>>>>>>>>>>>> vvcep...@apache.org> wrote: >>>>>>>>>>>>>>>>>>>> Ah, thanks Sophie, >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> I'm sorry for misinterpreting your resonse. Yes, >>>>>>>> we >>>>>>>>>>>>>>>>>>>> absolutely can and should clear the context >>>>>> before >>>>>>>>>>>>>>>>>>>> punctuating. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> My secondary concern is maybe more far-fetched. >>>>>> I >>>>>>>> was >>>>>>>>>>>>>>>>>>>> thinking that inside process(key,value), a >>>>>>>> Processor >>>>>>>>>> might >>>>>>>>>>>>>>>>>>>> do a get/put of a _different_ key. Consider, for >>>>>>>>>> example, >>>>>>>>>>>>>>>>>>>> the way that Suppress processors work. When they >>>>>>>> get a >>>>>>>>>>>>>>>>>>>> record, they add it to the store and then do a >>>>>>>> range >>>>>>>>>> scan >>>>>>>>>>>>>>>>>>>> and possibly forward a _different_ record. Of >>>>>>>> course, >>>>>>>>>> this >>>>>>>>>>>>>>>>>>>> is an operation that is deeply coupled to the >>>>>>>>>> internals, and >>>>>>>>>>>>>>>>>>>> the Suppress processor accordingly actually does >>>>>>>> get >>>>>>>>>> access >>>>>>>>>>>>>>>>>>>> to the internal context so that it can set the >>>>>>>> context >>>>>>>>>>>>>>>>>>>> before forwarding. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> Still, it seems like I've had a handful of >>>>>>>>>> conversations >>>>>>>>>>>>>>>>>>>> with people over the years in which they tell me >>>>>>>> they >>>>>>>>>> are >>>>>>>>>>>>>>>>>>>> using state stores in a way that transcends the >>>>>>>> "get >>>>>>>>>> and put >>>>>>>>>>>>>>>>>>>> the currently processing record" access >>>>>> pattern. I >>>>>>>>>> doubt >>>>>>>>>>>>>>>>>>>> that those folks would even have considered the >>>>>>>>>> possiblity >>>>>>>>>>>>>>>>>>>> that the currently processing record's _context_ >>>>>>>> could >>>>>>>>>>>>>>>>>>>> pollute their state store operations, as I >>>>>> myself >>>>>>>>>> never gave >>>>>>>>>>>>>>>>>>>> it a second thought until the current >>>>>> conversation >>>>>>>>>> began. In >>>>>>>>>>>>>>>>>>>> cases like that, we have actually set a trap for >>>>>>>> these >>>>>>>>>>>>>>>>>>>> people, and it seems better to dismantle the >>>>>> trap. >>>>>>>>>>>>>>>>>>>> As you noted, really the only people who would >>>>>> be >>>>>>>>>> negatively >>>>>>>>>>>>>>>>>>>> impacted are people who implement their own >>>>>> state >>>>>>>>>> stores. >>>>>>>>>>>>>>>>>>>> These folks will get the deprecation warning and >>>>>>>> try to >>>>>>>>>>>>>>>>>>>> adapt their stores to the new interface. If they >>>>>>>> needed >>>>>>>>>>>>>>>>>>>> access to the record context, they would find >>>>>>>> it's now >>>>>>>>>>>>>>>>>>>> missing. They'd ask us about it, and we'd have >>>>>> the >>>>>>>>>> ability >>>>>>>>>>>>>>>>>>>> to explain the lurking bug that they have had in >>>>>>>> their >>>>>>>>>>>>>>>>>>>> stores all along, as well as the new recommended >>>>>>>>>> pattern >>>>>>>>>>>>>>>>>>>> (just pass everything you need in the value). If >>>>>>>> that's >>>>>>>>>>>>>>>>>>>> unsatisfying, _then_ we should consider amending >>>>>>>> the >>>>>>>>>> API. >>>>>>>>>>>>>>>>>>>> Thanks, >>>>>>>>>>>>>>>>>>>> -John >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> On Thu, 2020-09-10 at 15:21 -0700, Sophie >>>>>>>> Blee-Goldman >>>>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>>> Regarding your first sentence, "...the >>>>>>>> processor >>>>>>>>>> would >>>>>>>>>>>>> null >>>>>>>>>>>>>>>>>>>>>> out the record context...", this is not >>>>>>>> possible, >>>>>>>>>> since >>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>> processor doesn't have write access to the >>>>>>>>>> context. We >>>>>>>>>>>>> could >>>>>>>>>>>>>>>>>>>>>> add it, >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> Sorry, this was poorly phrased, I definitely >>>>>>>> did not >>>>>>>>>> mean >>>>>>>>>>>>> to imply that >>>>>>>>>>>>>>>>>>>> we >>>>>>>>>>>>>>>>>>>>> should make the context modifiable by the >>>>>>>> Processors >>>>>>>>>>>>> themselves. I >>>>>>>>>>>>>>>>>>> meant >>>>>>>>>>>>>>>>>>>>> this should be handled by the internal >>>>>>>> processing >>>>>>>>>>>>> framework that deals >>>>>>>>>>>>>>>>>>>> with >>>>>>>>>>>>>>>>>>>>> passing records from one Processor to the >>>>>> next, >>>>>>>>>> setting >>>>>>>>>>>>> the record >>>>>>>>>>>>>>>>>>>> context >>>>>>>>>>>>>>>>>>>>> when a new record is picked up, invoking the >>>>>>>>>> punctuators, >>>>>>>>>>>>> etc. I >>>>>>>>>>>>>>>>>>> believe >>>>>>>>>>>>>>>>>>>>> this >>>>>>>>>>>>>>>>>>>>> all currently happens in the StreamTask? It >>>>>>>> already >>>>>>>>>> can >>>>>>>>>>>>> and does >>>>>>>>>>>>>>>>>>>> overwrite >>>>>>>>>>>>>>>>>>>>> the record context as new records are >>>>>>>> processed, and >>>>>>>>>> is >>>>>>>>>>>>> also >>>>>>>>>>>>>>>>>>> responsible >>>>>>>>>>>>>>>>>>>>> for calling the punctuators, so it doesn't >>>>>> seem >>>>>>>> like >>>>>>>>>> a >>>>>>>>>>>>> huge leap to >>>>>>>>>>>>>>>>>>> just >>>>>>>>>>>>>>>>>>>> say >>>>>>>>>>>>>>>>>>>>> "null out the current record before >>>>>> punctuating" >>>>>>>>>>>>>>>>>>>>> To clarify, I was never advocating or even >>>>>>>>>> considering to >>>>>>>>>>>>> give the >>>>>>>>>>>>>>>>>>>>> Processors >>>>>>>>>>>>>>>>>>>>> write access to the record context. Sorry if >>>>>> my >>>>>>>> last >>>>>>>>>>>>> message (or all of >>>>>>>>>>>>>>>>>>>>> them) >>>>>>>>>>>>>>>>>>>>> was misleading. I just wanted to point out >>>>>> that >>>>>>>> the >>>>>>>>>>>>> punctuator concern >>>>>>>>>>>>>>>>>>> is >>>>>>>>>>>>>>>>>>>>> orthogonal to the question of whether we >>>>>> should >>>>>>>>>> include >>>>>>>>>>>>> the record >>>>>>>>>>>>>>>>>>>> context >>>>>>>>>>>>>>>>>>>>> in the StateStoreContext. It's definitely a >>>>>> real >>>>>>>>>> problem, >>>>>>>>>>>>> but it's a >>>>>>>>>>>>>>>>>>>>> problem >>>>>>>>>>>>>>>>>>>>> that exists at the Processor level and not >>>>>> just >>>>>>>> the >>>>>>>>>>>>> StateStore. >>>>>>>>>>>>>>>>>>>>> So, I don't think there is any reason we >>>>>> *can't* >>>>>>>>>> retain >>>>>>>>>>>>> the record >>>>>>>>>>>>>>>>>>>> context >>>>>>>>>>>>>>>>>>>>> in the >>>>>>>>>>>>>>>>>>>>> StateStoreContext, and if any users came along >>>>>>>> with a >>>>>>>>>>>>> clear use case >>>>>>>>>>>>>>>>>>> I'd >>>>>>>>>>>>>>>>>>>>> find >>>>>>>>>>>>>>>>>>>>> that convincing. In the absence of any >>>>>>>> examples, the >>>>>>>>>>>>> conservative >>>>>>>>>>>>>>>>>>>> approach >>>>>>>>>>>>>>>>>>>>> sounds good to me. >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> If it turns out that someone did need the >>>>>> record >>>>>>>>>> context >>>>>>>>>>>>> in their >>>>>>>>>>>>>>>>>>> custom >>>>>>>>>>>>>>>>>>>>> state >>>>>>>>>>>>>>>>>>>>> store, I'm sure they'll submit a politely >>>>>>>> worded bug >>>>>>>>>>>>> report alerting us >>>>>>>>>>>>>>>>>>>>> that we >>>>>>>>>>>>>>>>>>>>> broke their application. >>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>> On Thu, Sep 10, 2020 at 3:05 PM John Roesler < >>>>>>>>>>>>> vvcep...@apache.org> >>>>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>>> Thanks, Sophie, >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> Yes, now that you point it out, I can see >>>>>>>> that the >>>>>>>>>> record >>>>>>>>>>>>>>>>>>>>>> context itself should be nulled out by >>>>>> Streams >>>>>>>>>> before >>>>>>>>>>>>>>>>>>>>>> invoking punctuators. From that perspective, >>>>>>>> we >>>>>>>>>> don't >>>>>>>>>>>>> need >>>>>>>>>>>>>>>>>>>>>> to think about the second-order problem of >>>>>>>> what's >>>>>>>>>> in the >>>>>>>>>>>>>>>>>>>>>> context for the state store when called >>>>>> from a >>>>>>>>>>>>> punctuator. >>>>>>>>>>>>>>>>>>>>>> Regarding your first sentence, "...the >>>>>>>> processor >>>>>>>>>> would >>>>>>>>>>>>> null >>>>>>>>>>>>>>>>>>>>>> out the record context...", this is not >>>>>>>> possible, >>>>>>>>>> since >>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>> processor doesn't have write access to the >>>>>>>>>> context. We >>>>>>>>>>>>> could >>>>>>>>>>>>>>>>>>>>>> add it, but then all kinds of strange >>>>>> effects >>>>>>>>>> would ensue >>>>>>>>>>>>>>>>>>>>>> when downstream processors execute but the >>>>>>>> context >>>>>>>>>> is >>>>>>>>>>>>> empty, >>>>>>>>>>>>>>>>>>>>>> etc. Better to just let the framework manage >>>>>>>> the >>>>>>>>>> record >>>>>>>>>>>>>>>>>>>>>> context and keep it read-only for >>>>>> Processors. >>>>>>>>>>>>>>>>>>>>>> Reading between the lines of your last >>>>>> reply, >>>>>>>> it >>>>>>>>>> sounds >>>>>>>>>>>>> that >>>>>>>>>>>>>>>>>>>>>> the disconnect may just have been a mutual >>>>>>>>>>>>> misunderstanding >>>>>>>>>>>>>>>>>>>>>> about whether or not Processors currently >>>>>> have >>>>>>>>>> access to >>>>>>>>>>>>> set >>>>>>>>>>>>>>>>>>>>>> the record context. Since they do not, if we >>>>>>>>>> wanted to >>>>>>>>>>>>> add >>>>>>>>>>>>>>>>>>>>>> the record context to StateStoreContext in a >>>>>>>>>> well-defined >>>>>>>>>>>>>>>>>>>>>> way, we'd also have to add the ability for >>>>>>>>>> Processors to >>>>>>>>>>>>>>>>>>>>>> manipulate it. But then, we're just >>>>>> creating a >>>>>>>>>>>>> side-channel >>>>>>>>>>>>>>>>>>>>>> for Processors to pass some information in >>>>>>>>>> arguments to >>>>>>>>>>>>>>>>>>>>>> "put()" and other information implicitly >>>>>>>> through >>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>> context. It seems better just to go for a >>>>>>>> single >>>>>>>>>> channel >>>>>>>>>>>>> for >>>>>>>>>>>>>>>>>>>>>> now. >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> It sounds like you're basically in favor of >>>>>>>> the >>>>>>>>>>>>> conservative >>>>>>>>>>>>>>>>>>>>>> approach, and you just wanted to understand >>>>>>>> the >>>>>>>>>> blockers >>>>>>>>>>>>>>>>>>>>>> that I implied. Does my clarification make >>>>>>>> sense? >>>>>>>>>>>>>>>>>>>>>> Thanks, >>>>>>>>>>>>>>>>>>>>>> -John >>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>> On Thu, 2020-09-10 at 10:54 -0700, Sophie >>>>>>>>>> Blee-Goldman >>>>>>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>>>> I was just thinking that the processor >>>>>> would >>>>>>>>>> null out >>>>>>>>>>>>> the record >>>>>>>>>>>>>>>>>>>> context >>>>>>>>>>>>>>>>>>>>>>> after it >>>>>>>>>>>>>>>>>>>>>>> finished processing the record, so I'm not >>>>>>>> sure I >>>>>>>>>>>>> follow why this >>>>>>>>>>>>>>>>>>>> would >>>>>>>>>>>>>>>>>>>>>> not >>>>>>>>>>>>>>>>>>>>>>> be >>>>>>>>>>>>>>>>>>>>>>> possible? AFAIK we never call a punctuator >>>>>>>> in the >>>>>>>>>>>>> middle of >>>>>>>>>>>>>>>>>>>> processing a >>>>>>>>>>>>>>>>>>>>>>> record through the topology, and even if >>>>>> we >>>>>>>> did, >>>>>>>>>> we >>>>>>>>>>>>> still know when >>>>>>>>>>>>>>>>>>>> it is >>>>>>>>>>>>>>>>>>>>>>> about >>>>>>>>>>>>>>>>>>>>>>> to be called and could set it to null >>>>>>>> beforehand. >>>>>>>>>>>>>>>>>>>>>>> I'm not trying to advocate for it here, >>>>>> I'm >>>>>>>> in >>>>>>>>>>>>> agreement that >>>>>>>>>>>>>>>>>>>> anything >>>>>>>>>>>>>>>>>>>>>> you >>>>>>>>>>>>>>>>>>>>>>> want >>>>>>>>>>>>>>>>>>>>>>> to access within the store can and should >>>>>> be >>>>>>>>>> accessed >>>>>>>>>>>>> within the >>>>>>>>>>>>>>>>>>>> calling >>>>>>>>>>>>>>>>>>>>>>> Processor/Punctuator before reaching the >>>>>>>> store. >>>>>>>>>> The >>>>>>>>>>>>> "we can always >>>>>>>>>>>>>>>>>>>> add it >>>>>>>>>>>>>>>>>>>>>>> later if necessary" argument is also >>>>>> pretty >>>>>>>>>>>>> convincing. Just trying >>>>>>>>>>>>>>>>>>>> to >>>>>>>>>>>>>>>>>>>>>>> understand >>>>>>>>>>>>>>>>>>>>>>> why this wouldn't be possible. >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> FWIW, the question of "what is the current >>>>>>>>>> record in >>>>>>>>>>>>> the context >>>>>>>>>>>>>>>>>>> of a >>>>>>>>>>>>>>>>>>>>>>> Punctuator" >>>>>>>>>>>>>>>>>>>>>>> exists independently of whether we want to >>>>>>>> add >>>>>>>>>> this to >>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>> StateStoreContext >>>>>>>>>>>>>>>>>>>>>>> or not. The full ProcessorContext, >>>>>>>> including the >>>>>>>>>>>>> current record >>>>>>>>>>>>>>>>>>>> context, >>>>>>>>>>>>>>>>>>>>>> is >>>>>>>>>>>>>>>>>>>>>>> already available within a Punctuator, so >>>>>>>>>> removing the >>>>>>>>>>>>> current >>>>>>>>>>>>>>>>>>> record >>>>>>>>>>>>>>>>>>>>>>> context >>>>>>>>>>>>>>>>>>>>>>> from the StateStoreContext does not solve >>>>>>>> the >>>>>>>>>> problem. >>>>>>>>>>>>> Users can -- >>>>>>>>>>>>>>>>>>>> and >>>>>>>>>>>>>>>>>>>>>> have >>>>>>>>>>>>>>>>>>>>>>> (see KAFKA-9584 < >>>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-9584 >>>>>>>>>>>>>>>>>>>> ;;) >>>>>>>>>>>>>>>>>>>> -- >>>>>>>>>>>>>>>>>>>>>> hit >>>>>>>>>>>>>>>>>>>>>>> such subtle bugs without ever invoking a >>>>>>>>>> StateStore >>>>>>>>>>>>>>>>>>>>>>> from their punctuator. >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> Again, I think I do agree that we should >>>>>>>> leave >>>>>>>>>> the >>>>>>>>>>>>> current record >>>>>>>>>>>>>>>>>>>> context >>>>>>>>>>>>>>>>>>>>>>> off of >>>>>>>>>>>>>>>>>>>>>>> the StateStoreContext, but I don't think >>>>>> the >>>>>>>>>>>>> Punctuator argument >>>>>>>>>>>>>>>>>>>> against >>>>>>>>>>>>>>>>>>>>>> it >>>>>>>>>>>>>>>>>>>>>>> is >>>>>>>>>>>>>>>>>>>>>>> very convincing. It sounds to me like we >>>>>>>> need to >>>>>>>>>>>>> disallow access to >>>>>>>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>>> current >>>>>>>>>>>>>>>>>>>>>>> record context from within the Punctuator, >>>>>>>>>> independent >>>>>>>>>>>>> of anything >>>>>>>>>>>>>>>>>>>> to do >>>>>>>>>>>>>>>>>>>>>>> with >>>>>>>>>>>>>>>>>>>>>>> state stores >>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>> On Thu, Sep 10, 2020 at 7:12 AM John >>>>>>>> Roesler < >>>>>>>>>>>>> vvcep...@apache.org> >>>>>>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>>>>> Thanks for the thoughts, Sophie. >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> I agree that the extra information could >>>>>>>> be >>>>>>>>>> useful. >>>>>>>>>>>>> My only >>>>>>>>>>>>>>>>>>>> concern is >>>>>>>>>>>>>>>>>>>>>>>> that it doesn’t seem like we can >>>>>> actually >>>>>>>>>> supply >>>>>>>>>>>>> that extra >>>>>>>>>>>>>>>>>>>> information >>>>>>>>>>>>>>>>>>>>>>>> correctly. So, then we have a situation >>>>>>>> where >>>>>>>>>> the >>>>>>>>>>>>> system offers >>>>>>>>>>>>>>>>>>>> useful >>>>>>>>>>>>>>>>>>>>>> API >>>>>>>>>>>>>>>>>>>>>>>> calls that are only correct in a narrow >>>>>>>> range >>>>>>>>>> of use >>>>>>>>>>>>> cases. >>>>>>>>>>>>>>>>>>>> Outside of >>>>>>>>>>>>>>>>>>>>>>>> those use cases, you get incorrect >>>>>>>> behavior. >>>>>>>>>>>>>>>>>>>>>>>> If it were possible to null out the >>>>>>>> context >>>>>>>>>> before >>>>>>>>>>>>> you put a >>>>>>>>>>>>>>>>>>>> document >>>>>>>>>>>>>>>>>>>>>> to >>>>>>>>>>>>>>>>>>>>>>>> which the context doesn’t apply, then >>>>>> the >>>>>>>>>> concern >>>>>>>>>>>>> would be >>>>>>>>>>>>>>>>>>>> mitigated. >>>>>>>>>>>>>>>>>>>>>> But >>>>>>>>>>>>>>>>>>>>>>>> it would still be pretty weird from the >>>>>>>>>> perspective >>>>>>>>>>>>> of the store >>>>>>>>>>>>>>>>>>>> that >>>>>>>>>>>>>>>>>>>>>>>> sometimes the context is populated and >>>>>>>> other >>>>>>>>>> times, >>>>>>>>>>>>> it’s null. >>>>>>>>>>>>>>>>>>>>>>>> But that seems moot, since it doesn’t >>>>>> seem >>>>>>>>>> possible >>>>>>>>>>>>> to null out >>>>>>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>>>> context. Only the Processor could know >>>>>>>> whether >>>>>>>>>> it’s >>>>>>>>>>>>> about to put >>>>>>>>>>>>>>>>>>> a >>>>>>>>>>>>>>>>>>>>>> document >>>>>>>>>>>>>>>>>>>>>>>> different from the context or not. And >>>>>> it >>>>>>>>>> would be >>>>>>>>>>>>> inappropriate >>>>>>>>>>>>>>>>>>> to >>>>>>>>>>>>>>>>>>>>>> offer a >>>>>>>>>>>>>>>>>>>>>>>> public ProcessorContext api to manage >>>>>> the >>>>>>>>>> record >>>>>>>>>>>>> context. >>>>>>>>>>>>>>>>>>>>>>>> Ultimately, it still seems like if you >>>>>>>> want to >>>>>>>>>> store >>>>>>>>>>>>> headers, you >>>>>>>>>>>>>>>>>>>> can >>>>>>>>>>>>>>>>>>>>>>>> store them explicitly, right? That >>>>>>>> doesn’t seem >>>>>>>>>>>>> onerous to me, >>>>>>>>>>>>>>>>>>> and >>>>>>>>>>>>>>>>>>>> it >>>>>>>>>>>>>>>>>>>>>> kind >>>>>>>>>>>>>>>>>>>>>>>> of seems better than relying on >>>>>> undefined >>>>>>>> or >>>>>>>>>>>>> asymmetrical >>>>>>>>>>>>>>>>>>> behavior >>>>>>>>>>>>>>>>>>>> in >>>>>>>>>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>>>> store itself. >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> Anyway, I’m not saying that we couldn’t >>>>>>>> solve >>>>>>>>>> these >>>>>>>>>>>>> problems. >>>>>>>>>>>>>>>>>>> Just >>>>>>>>>>>>>>>>>>>>>> that it >>>>>>>>>>>>>>>>>>>>>>>> seems a little that we can be >>>>>>>> conservative and >>>>>>>>>> avoid >>>>>>>>>>>>> them for >>>>>>>>>>>>>>>>>>> now. >>>>>>>>>>>>>>>>>>>> If >>>>>>>>>>>>>>>>>>>>>> it >>>>>>>>>>>>>>>>>>>>>>>> turns out we really need to solve them, >>>>>>>> we can >>>>>>>>>>>>> always do it >>>>>>>>>>>>>>>>>>> later. >>>>>>>>>>>>>>>>>>>>>>>> Thanks, >>>>>>>>>>>>>>>>>>>>>>>> John >>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>> On Wed, Sep 9, 2020, at 22:46, Sophie >>>>>>>>>> Blee-Goldman >>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>>>>>>> If you were to call "put" from a >>>>>>>>>> punctuator, or >>>>>>>>>>>>> do a >>>>>>>>>>>>>>>>>>>>>>>>>> `range()` query and then update one >>>>>> of >>>>>>>>>> those >>>>>>>>>>>>> records with >>>>>>>>>>>>>>>>>>>>>>>>>> `put()`, you'd have a very subtle >>>>>> bug >>>>>>>> on >>>>>>>>>> your >>>>>>>>>>>>> hands. >>>>>>>>>>>>>>>>>>>>>>>>> Can you elaborate on this a bit? I >>>>>> agree >>>>>>>>>> that the >>>>>>>>>>>>> punctuator >>>>>>>>>>>>>>>>>>>> case is >>>>>>>>>>>>>>>>>>>>>> an >>>>>>>>>>>>>>>>>>>>>>>>> obvious exemption to the assumption >>>>>> that >>>>>>>>>> store >>>>>>>>>>>>> invocations >>>>>>>>>>>>>>>>>>> always >>>>>>>>>>>>>>>>>>>>>>>>> have a corresponding "current record", >>>>>>>> but I >>>>>>>>>> don't >>>>>>>>>>>>> understand >>>>>>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>>>>> second example. Are you envisioning a >>>>>>>>>> scenario >>>>>>>>>>>>> where the >>>>>>>>>>>>>>>>>>> #process >>>>>>>>>>>>>>>>>>>>>>>>> method performs a range query and then >>>>>>>>>> updates >>>>>>>>>>>>> records? Or were >>>>>>>>>>>>>>>>>>>>>>>>> you just giving another example of the >>>>>>>>>> punctuator >>>>>>>>>>>>> case? >>>>>>>>>>>>>>>>>>>>>>>>> I only bring it up because I agree >>>>>> that >>>>>>>> the >>>>>>>>>>>>> current record >>>>>>>>>>>>>>>>>>>>>> information >>>>>>>>>>>>>>>>>>>>>>>> could >>>>>>>>>>>>>>>>>>>>>>>>> still be useful within the context of >>>>>>>> the >>>>>>>>>> store. >>>>>>>>>>>>> As a non-user >>>>>>>>>>>>>>>>>>> my >>>>>>>>>>>>>>>>>>>>>> input >>>>>>>>>>>>>>>>>>>>>>>> on >>>>>>>>>>>>>>>>>>>>>>>>> this >>>>>>>>>>>>>>>>>>>>>>>>> definitely has limited value, but it >>>>>>>> just >>>>>>>>>> isn't >>>>>>>>>>>>> striking me as >>>>>>>>>>>>>>>>>>>>>> obvious >>>>>>>>>>>>>>>>>>>>>>>> that >>>>>>>>>>>>>>>>>>>>>>>>> we >>>>>>>>>>>>>>>>>>>>>>>>> should remove access to the current >>>>>>>> record >>>>>>>>>> context >>>>>>>>>>>>> from the >>>>>>>>>>>>>>>>>>> state >>>>>>>>>>>>>>>>>>>>>> stores. >>>>>>>>>>>>>>>>>>>>>>>>> If there is no current record, as in >>>>>> the >>>>>>>>>>>>> punctuator case, we >>>>>>>>>>>>>>>>>>>> should >>>>>>>>>>>>>>>>>>>>>> just >>>>>>>>>>>>>>>>>>>>>>>>> set >>>>>>>>>>>>>>>>>>>>>>>>> the record context to null (or >>>>>>>>>> Optional.empty, >>>>>>>>>>>>> etc). >>>>>>>>>>>>>>>>>>>>>>>>> That said, the put() always has to >>>>>> come >>>>>>>> from >>>>>>>>>>>>> somewhere, and >>>>>>>>>>>>>>>>>>> that >>>>>>>>>>>>>>>>>>>>>>>>> somewhere is always going to be >>>>>> either a >>>>>>>>>> Processor >>>>>>>>>>>>> or a >>>>>>>>>>>>>>>>>>>> Punctuator, >>>>>>>>>>>>>>>>>>>>>> both >>>>>>>>>>>>>>>>>>>>>>>>> of which will still have access to the >>>>>>>> full >>>>>>>>>>>>> context. So >>>>>>>>>>>>>>>>>>>> additional >>>>>>>>>>>>>>>>>>>>>> info >>>>>>>>>>>>>>>>>>>>>>>>> such as >>>>>>>>>>>>>>>>>>>>>>>>> the timestamp can and should probably >>>>>> be >>>>>>>>>> supplied >>>>>>>>>>>>> to the store >>>>>>>>>>>>>>>>>>>> before >>>>>>>>>>>>>>>>>>>>>>>>> calling put(), rather than looked up >>>>>> by >>>>>>>> the >>>>>>>>>> store. >>>>>>>>>>>>> But I can >>>>>>>>>>>>>>>>>>> see >>>>>>>>>>>>>>>>>>>> some >>>>>>>>>>>>>>>>>>>>>>>> other >>>>>>>>>>>>>>>>>>>>>>>>> things being useful, for example the >>>>>>>> current >>>>>>>>>>>>> record's headers. >>>>>>>>>>>>>>>>>>>> Maybe >>>>>>>>>>>>>>>>>>>>>>>> if/when >>>>>>>>>>>>>>>>>>>>>>>>> we add better (or any) support for >>>>>>>> headers in >>>>>>>>>>>>> state stores this >>>>>>>>>>>>>>>>>>>> will >>>>>>>>>>>>>>>>>>>>>> be >>>>>>>>>>>>>>>>>>>>>>>>> less true. >>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>> Of course as John has made clear, it's >>>>>>>>>> pretty hard >>>>>>>>>>>>> to judge >>>>>>>>>>>>>>>>>>>> without >>>>>>>>>>>>>>>>>>>>>>>>> examples >>>>>>>>>>>>>>>>>>>>>>>>> and more insight as to what actually >>>>>>>> goes on >>>>>>>>>>>>> within a custom >>>>>>>>>>>>>>>>>>>> state >>>>>>>>>>>>>>>>>>>>>> store >>>>>>>>>>>>>>>>>>>>>>>>> On Wed, Sep 9, 2020 at 8:07 PM John >>>>>>>> Roesler < >>>>>>>>>>>>>>>>>>> vvcep...@apache.org >>>>>>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>>>>>>> Hi Paul, >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> It's good to hear from you! >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> I'm glad you're in favor of the >>>>>>>> direction. >>>>>>>>>>>>> Especially when >>>>>>>>>>>>>>>>>>>>>>>>>> it comes to public API and usability >>>>>>>>>> concens, I >>>>>>>>>>>>> tend to >>>>>>>>>>>>>>>>>>>>>>>>>> think that "the folks who matter" >>>>>> are >>>>>>>>>> actually >>>>>>>>>>>>> the folks who >>>>>>>>>>>>>>>>>>>>>>>>>> have to use the APIs to accomplish >>>>>>>> real >>>>>>>>>> tasks. >>>>>>>>>>>>> It can be >>>>>>>>>>>>>>>>>>>>>>>>>> hard for me to be sure I'm thinking >>>>>>>>>> clearly from >>>>>>>>>>>>> that >>>>>>>>>>>>>>>>>>>>>>>>>> perspective. >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> Funny story, I also started down >>>>>> this >>>>>>>> road >>>>>>>>>> a >>>>>>>>>>>>> couple of times >>>>>>>>>>>>>>>>>>>>>>>>>> already and backed them out before >>>>>>>> the KIP >>>>>>>>>>>>> because I was >>>>>>>>>>>>>>>>>>>>>>>>>> afraid of the scope of the proposal. >>>>>>>>>>>>> Unfortunately, needing >>>>>>>>>>>>>>>>>>>>>>>>>> to make a new ProcessorContext kind >>>>>> of >>>>>>>>>> forced my >>>>>>>>>>>>> hand. >>>>>>>>>>>>>>>>>>>>>>>>>> I see you've called me out about the >>>>>>>>>>>>> ChangeLogging stores :) >>>>>>>>>>>>>>>>>>>>>>>>>> In fact, I think these are the >>>>>>>> main/only >>>>>>>>>> reason >>>>>>>>>>>>> that stores >>>>>>>>>>>>>>>>>>>>>>>>>> might really need to invoke >>>>>>>> "forward()". My >>>>>>>>>>>>> secret plan was >>>>>>>>>>>>>>>>>>>>>>>>>> to cheat and either accomplish >>>>>>>>>> change-logging by >>>>>>>>>>>>> a different >>>>>>>>>>>>>>>>>>>>>>>>>> mechanism than implementing the >>>>>> store >>>>>>>>>> interface, >>>>>>>>>>>>> or by just >>>>>>>>>>>>>>>>>>>>>>>>>> breaking encapsulation to sneak the >>>>>>>> "real" >>>>>>>>>>>>> ProcessorContext >>>>>>>>>>>>>>>>>>>>>>>>>> into the ChangeLogging stores. But >>>>>>>> those >>>>>>>>>> are all >>>>>>>>>>>>>>>>>>>>>>>>>> implementation details. I think the >>>>>>>> key >>>>>>>>>> question >>>>>>>>>>>>> is whether >>>>>>>>>>>>>>>>>>>>>>>>>> anyone else has a store >>>>>>>> implementation that >>>>>>>>>>>>> needs to call >>>>>>>>>>>>>>>>>>>>>>>>>> "forward()". It's not what you >>>>>>>> mentioned, >>>>>>>>>> but >>>>>>>>>>>>> since you >>>>>>>>>>>>>>>>>>>>>>>>>> spoke up, I'll just ask: if you have >>>>>>>> a use >>>>>>>>>> case >>>>>>>>>>>>> for calling >>>>>>>>>>>>>>>>>>>>>>>>>> "forward()" in a store, please share >>>>>>>> it. >>>>>>>>>>>>>>>>>>>>>>>>>> Regarding the other record-specific >>>>>>>> context >>>>>>>>>>>>> methods, I think >>>>>>>>>>>>>>>>>>>>>>>>>> you have a good point, but I also >>>>>>>> can't >>>>>>>>>> quite >>>>>>>>>>>>> wrap my head >>>>>>>>>>>>>>>>>>>>>>>>>> around how we can actually guarantee >>>>>>>> it to >>>>>>>>>> work >>>>>>>>>>>>> in general. >>>>>>>>>>>>>>>>>>>>>>>>>> For example, the case you cited, >>>>>>>> where the >>>>>>>>>>>>> implementation of >>>>>>>>>>>>>>>>>>>>>>>>>> `KeyValueStore#put(key, value)` uses >>>>>>>> the >>>>>>>>>> context >>>>>>>>>>>>> to augment >>>>>>>>>>>>>>>>>>>>>>>>>> the record with timestamp >>>>>>>> information. This >>>>>>>>>>>>> relies on the >>>>>>>>>>>>>>>>>>>>>>>>>> assumption that you would only call >>>>>>>>>> "put()" from >>>>>>>>>>>>> inside a >>>>>>>>>>>>>>>>>>>>>>>>>> `Processor#process(key, value)` call >>>>>>>> in >>>>>>>>>> which >>>>>>>>>>>>> the record >>>>>>>>>>>>>>>>>>>>>>>>>> being processed is the same record >>>>>>>> that >>>>>>>>>> you're >>>>>>>>>>>>> trying to put >>>>>>>>>>>>>>>>>>>>>>>>>> into the store. >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> If you were to call "put" from a >>>>>>>>>> punctuator, or >>>>>>>>>>>>> do a >>>>>>>>>>>>>>>>>>>>>>>>>> `range()` query and then update one >>>>>> of >>>>>>>>>> those >>>>>>>>>>>>> records with >>>>>>>>>>>>>>>>>>>>>>>>>> `put()`, you'd have a very subtle >>>>>> bug >>>>>>>> on >>>>>>>>>> your >>>>>>>>>>>>> hands. Right >>>>>>>>>>>>>>>>>>>>>>>>>> now, the Streams component that >>>>>>>> actually >>>>>>>>>> calls >>>>>>>>>>>>> the Processor >>>>>>>>>>>>>>>>>>>>>>>>>> takes care to set the right record >>>>>>>> context >>>>>>>>>>>>> before invoking >>>>>>>>>>>>>>>>>>>>>>>>>> the method, and in the case of >>>>>>>> caching, >>>>>>>>>> etc., it >>>>>>>>>>>>> also takes >>>>>>>>>>>>>>>>>>>>>>>>>> care to swap out the old context and >>>>>>>> keep >>>>>>>>>> it >>>>>>>>>>>>> somewhere safe. >>>>>>>>>>>>>>>>>>>>>>>>>> But when it comes to public API >>>>>>>> Processors >>>>>>>>>>>>> calling methods >>>>>>>>>>>>>>>>>>>>>>>>>> on StateStores, there's no >>>>>>>> opportunity for >>>>>>>>>> any >>>>>>>>>>>>> component to >>>>>>>>>>>>>>>>>>>>>>>>>> make sure the context is always >>>>>>>> correct. >>>>>>>>>>>>>>>>>>>>>>>>>> In the face of that situation, it >>>>>>>> seemed >>>>>>>>>> better >>>>>>>>>>>>> to just move >>>>>>>>>>>>>>>>>>>>>>>>>> in the direction of a "normal" data >>>>>>>> store. >>>>>>>>>> I.e., >>>>>>>>>>>>> when you >>>>>>>>>>>>>>>>>>>>>>>>>> use a HashMap or RocksDB or other >>>>>>>> "state >>>>>>>>>>>>> stores", you don't >>>>>>>>>>>>>>>>>>>>>>>>>> expect them to automatically know >>>>>>>> extra >>>>>>>>>> stuff >>>>>>>>>>>>> about the >>>>>>>>>>>>>>>>>>>>>>>>>> record you're storing. If you need >>>>>>>> them to >>>>>>>>>> know >>>>>>>>>>>>> something, >>>>>>>>>>>>>>>>>>>>>>>>>> you just put it in the value. >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> All of that said, I'm just reasoning >>>>>>>> from >>>>>>>>>> first >>>>>>>>>>>>> principles >>>>>>>>>>>>>>>>>>>>>>>>>> here. To really know if this is a >>>>>>>> mistake >>>>>>>>>> or >>>>>>>>>>>>> not, I need to >>>>>>>>>>>>>>>>>>>>>>>>>> be in your place. So please push >>>>>> back >>>>>>>> if >>>>>>>>>> you >>>>>>>>>>>>> think what I >>>>>>>>>>>>>>>>>>>>>>>>>> said is nonsense. My personal plan >>>>>>>> was to >>>>>>>>>> keep >>>>>>>>>>>>> an eye out >>>>>>>>>>>>>>>>>>>>>>>>>> during the period where the old API >>>>>>>> was >>>>>>>>>> still >>>>>>>>>>>>> present, but >>>>>>>>>>>>>>>>>>>>>>>>>> deprecated, to see if people were >>>>>>>>>> struggling to >>>>>>>>>>>>> use the new >>>>>>>>>>>>>>>>>>>>>>>>>> API. If so, then we'd have a chance >>>>>> to >>>>>>>>>> address >>>>>>>>>>>>> it before >>>>>>>>>>>>>>>>>>>>>>>>>> dropping the old API. But it's even >>>>>>>> better >>>>>>>>>> if >>>>>>>>>>>>> you can help >>>>>>>>>>>>>>>>>>>>>>>>>> think it through now. >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> It did also cross my mind to _not_ >>>>>>>> add the >>>>>>>>>>>>>>>>>>>>>>>>>> StateStoreContext, but just to >>>>>>>> continue to >>>>>>>>>> punt >>>>>>>>>>>>> on the >>>>>>>>>>>>>>>>>>>>>>>>>> question by just dropping in the new >>>>>>>>>>>>> ProcessorContext to the >>>>>>>>>>>>>>>>>>>>>>>>>> new init method. If >>>>>> StateStoreContext >>>>>>>>>> seems too >>>>>>>>>>>>> bold, we can >>>>>>>>>>>>>>>>>>>>>>>>>> go that direction. But if we >>>>>> actually >>>>>>>> add >>>>>>>>>> some >>>>>>>>>>>>> methods to >>>>>>>>>>>>>>>>>>>>>>>>>> StateStoreContext, I'd like to be >>>>>>>> able to >>>>>>>>>> ensure >>>>>>>>>>>>> they would >>>>>>>>>>>>>>>>>>>>>>>>>> be well defined. I think the current >>>>>>>>>> situation >>>>>>>>>>>>> was more of >>>>>>>>>>>>>>>>>>>>>>>>>> an oversight than a choice. >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> Thanks again for your reply, >>>>>>>>>>>>>>>>>>>>>>>>>> -John >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>> On Wed, 2020-09-09 at 21:23 -0500, >>>>>>>> Paul >>>>>>>>>> Whalen >>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>>>>>>>> John, >>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>> It's exciting to see this KIP head >>>>>>>> in >>>>>>>>>> this >>>>>>>>>>>>> direction! In >>>>>>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>> last >>>>>>>>>>>>>>>>>>>>>>>> year >>>>>>>>>>>>>>>>>>>>>>>>>> or >>>>>>>>>>>>>>>>>>>>>>>>>>> so I've tried to sketch out some >>>>>>>>>> usability >>>>>>>>>>>>> improvements for >>>>>>>>>>>>>>>>>>>>>> custom >>>>>>>>>>>>>>>>>>>>>>>> state >>>>>>>>>>>>>>>>>>>>>>>>>>> stores, and I also ended up >>>>>>>> splitting >>>>>>>>>> out the >>>>>>>>>>>>>>>>>>>> StateStoreContext >>>>>>>>>>>>>>>>>>>>>> from >>>>>>>>>>>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>>>>>>> ProcessorContext in an attempt to >>>>>>>>>> facilitate >>>>>>>>>>>>> what I was >>>>>>>>>>>>>>>>>>>> doing. I >>>>>>>>>>>>>>>>>>>>>>>> sort of >>>>>>>>>>>>>>>>>>>>>>>>>>> abandoned it when I realized how >>>>>>>> large >>>>>>>>>> the >>>>>>>>>>>>> ideal change >>>>>>>>>>>>>>>>>>> might >>>>>>>>>>>>>>>>>>>>>> have >>>>>>>>>>>>>>>>>>>>>>>> to be, >>>>>>>>>>>>>>>>>>>>>>>>>>> but it's great to see that there >>>>>> is >>>>>>>> other >>>>>>>>>>>>> interest in >>>>>>>>>>>>>>>>>>> moving >>>>>>>>>>>>>>>>>>>> in >>>>>>>>>>>>>>>>>>>>>> this >>>>>>>>>>>>>>>>>>>>>>>>>>> direction (from the folks that >>>>>>>> matter :) >>>>>>>>>> ). >>>>>>>>>>>>>>>>>>>>>>>>>>> Having taken a stab at it myself, >>>>>> I >>>>>>>> have >>>>>>>>>> a >>>>>>>>>>>>> comment/question >>>>>>>>>>>>>>>>>>>> on >>>>>>>>>>>>>>>>>>>>>> this >>>>>>>>>>>>>>>>>>>>>>>>>> bullet >>>>>>>>>>>>>>>>>>>>>>>>>>> about StateStoreContext: >>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>> It does *not* include anything >>>>>>>>>> processor- or >>>>>>>>>>>>> record- >>>>>>>>>>>>>>>>>>>> specific, >>>>>>>>>>>>>>>>>>>>>> like >>>>>>>>>>>>>>>>>>>>>>>>>>>> `forward()` or any information >>>>>>>> about >>>>>>>>>> the >>>>>>>>>>>>> "current" >>>>>>>>>>>>>>>>>>> record, >>>>>>>>>>>>>>>>>>>>>> which is >>>>>>>>>>>>>>>>>>>>>>>>>> only a >>>>>>>>>>>>>>>>>>>>>>>>>>>> well-defined in the context of >>>>>> the >>>>>>>>>>>>> Processor. Processors >>>>>>>>>>>>>>>>>>>>>> process >>>>>>>>>>>>>>>>>>>>>>>> one >>>>>>>>>>>>>>>>>>>>>>>>>> record >>>>>>>>>>>>>>>>>>>>>>>>>>>> at a time, but state stores may >>>>>> be >>>>>>>>>> used to >>>>>>>>>>>>> store and >>>>>>>>>>>>>>>>>>> fetch >>>>>>>>>>>>>>>>>>>> many >>>>>>>>>>>>>>>>>>>>>>>>>> records, so >>>>>>>>>>>>>>>>>>>>>>>>>>>> there is no "current record". >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>> I totally agree that >>>>>>>> record-specific or >>>>>>>>>>>>> processor-specific >>>>>>>>>>>>>>>>>>>>>> context >>>>>>>>>>>>>>>>>>>>>>>> in a >>>>>>>>>>>>>>>>>>>>>>>>>>> state store is often not >>>>>>>> well-defined >>>>>>>>>> and it >>>>>>>>>>>>> would be good >>>>>>>>>>>>>>>>>>> to >>>>>>>>>>>>>>>>>>>>>>>> separate >>>>>>>>>>>>>>>>>>>>>>>>>> that >>>>>>>>>>>>>>>>>>>>>>>>>>> out, but sometimes it (at least >>>>>>>>>>>>> record-specific context) is >>>>>>>>>>>>>>>>>>>>>> actually >>>>>>>>>>>>>>>>>>>>>>>>>>> useful, for example, passing the >>>>>>>> record's >>>>>>>>>>>>> timestamp through >>>>>>>>>>>>>>>>>>>> to >>>>>>>>>>>>>>>>>>>>>> the >>>>>>>>>>>>>>>>>>>>>>>>>>> underlying storage (or changelog >>>>>>>> topic): >>>>>> https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStore.java#L121 >>>>>>>>>>>>>>>>>>>>>>>>>>> You could have the writer client >>>>>> of >>>>>>>> the >>>>>>>>>> state >>>>>>>>>>>>> store pass >>>>>>>>>>>>>>>>>>> this >>>>>>>>>>>>>>>>>>>>>>>> through, >>>>>>>>>>>>>>>>>>>>>>>>>> but >>>>>>>>>>>>>>>>>>>>>>>>>>> it would be nice to be able to >>>>>> write >>>>>>>>>> state >>>>>>>>>>>>> stores where the >>>>>>>>>>>>>>>>>>>>>> client >>>>>>>>>>>>>>>>>>>>>>>> did >>>>>>>>>>>>>>>>>>>>>>>>>> not >>>>>>>>>>>>>>>>>>>>>>>>>>> have this responsibility. I'm not >>>>>>>> sure >>>>>>>>>> if the >>>>>>>>>>>>> solution is >>>>>>>>>>>>>>>>>>>> to add >>>>>>>>>>>>>>>>>>>>>>>> some >>>>>>>>>>>>>>>>>>>>>>>>>>> things back to StateStoreContext, >>>>>> or >>>>>>>>>> make yet >>>>>>>>>>>>> another >>>>>>>>>>>>>>>>>>> context >>>>>>>>>>>>>>>>>>>>>> that >>>>>>>>>>>>>>>>>>>>>>>>>>> represents record-specific context >>>>>>>> while >>>>>>>>>>>>> inside a state >>>>>>>>>>>>>>>>>>>> store. >>>>>>>>>>>>>>>>>>>>>>>>>>> Best, >>>>>>>>>>>>>>>>>>>>>>>>>>> Paul >>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>> On Wed, Sep 9, 2020 at 5:43 PM >>>>>> John >>>>>>>>>> Roesler < >>>>>>>>>>>>>>>>>>>> j...@vvcephei.org> >>>>>>>>>>>>>>>>>>>>>>>> wrote: >>>>>>>>>>>>>>>>>>>>>>>>>>>> Hello all, >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> I've been slowly pushing KIP-478 >>>>>>>>>> forward >>>>>>>>>>>>> over the last >>>>>>>>>>>>>>>>>>>> year, >>>>>>>>>>>>>>>>>>>>>>>>>>>> and I'm happy to say that we're >>>>>>>> making >>>>>>>>>> good >>>>>>>>>>>>> progress now. >>>>>>>>>>>>>>>>>>>>>>>>>>>> However, several issues with the >>>>>>>>>> original >>>>>>>>>>>>> design have >>>>>>>>>>>>>>>>>>> come >>>>>>>>>>>>>>>>>>>>>>>>>>>> to light. >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> The major changes: >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> We discovered that the original >>>>>>>> plan >>>>>>>>>> of just >>>>>>>>>>>>> adding >>>>>>>>>>>>>>>>>>> generic >>>>>>>>>>>>>>>>>>>>>>>>>>>> parameters to ProcessorContext >>>>>>>> was too >>>>>>>>>>>>> disruptive, so we >>>>>>>>>>>>>>>>>>>> are >>>>>>>>>>>>>>>>>>>>>>>>>>>> now adding a new >>>>>>>> api.ProcessorContext. >>>>>>>>>>>>>>>>>>>>>>>>>>>> That choice forces us to add a >>>>>> new >>>>>>>>>>>>> StateStore.init method >>>>>>>>>>>>>>>>>>>>>>>>>>>> for the new context, but >>>>>>>>>> ProcessorContext >>>>>>>>>>>>> really isn't >>>>>>>>>>>>>>>>>>>> ideal >>>>>>>>>>>>>>>>>>>>>>>>>>>> for state stores to begin with, >>>>>>>> so I'm >>>>>>>>>>>>> proposing a new >>>>>>>>>>>>>>>>>>>>>>>>>>>> StateStoreContext for this >>>>>>>> purpose. In >>>>>>>>>> a >>>>>>>>>>>>> nutshell, there >>>>>>>>>>>>>>>>>>>> are >>>>>>>>>>>>>>>>>>>>>>>>>>>> quite a few methods in >>>>>>>>>> ProcessorContext that >>>>>>>>>>>>> actually >>>>>>>>>>>>>>>>>>>> should >>>>>>>>>>>>>>>>>>>>>>>>>>>> never be called from inside a >>>>>>>>>> StateStore. >>>>>>>>>>>>>>>>>>>>>>>>>>>> Also, since there is a new >>>>>>>>>> ProcessorContext >>>>>>>>>>>>> interface, we >>>>>>>>>>>>>>>>>>>>>>>>>>>> need a new MockProcessorContext >>>>>>>>>>>>> implementation in the >>>>>>>>>>>>>>>>>>> test- >>>>>>>>>>>>>>>>>>>>>>>>>>>> utils module. >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> The changeset for the KIP >>>>>>>> document is >>>>>>>>>> here: >>>>>> https://cwiki.apache.org/confluence/pages/diffpagesbyversion.action?pageId=118172121&selectedPageVersions=14&selectedPageVersions=10 >>>>>>>>>>>>>>>>>>>>>>>>>>>> And the KIP itself is here: >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-478+-+Strongly+typed+Processor+API >>>>>>>>>>>>>>>>>>>>>>>>>>>> If you have any concerns, please >>>>>>>> let >>>>>>>>>> me know! >>>>>>>>>>>>>>>>>>>>>>>>>>>> Thanks, >>>>>>>>>>>>>>>>>>>>>>>>>>>> -John >>>>>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>>>>> >
