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 >>>>>>>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>>>>>>> >
