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