Ping :) Any thoughts? On Mon, Nov 4, 2019 at 5:13 PM Vinoth Chandar <vchan...@confluent.io> wrote:
> >> I'm having some trouble wrapping my head around what race conditions > might occur, other than the fundamentally broken state in which different > instances are running totally different topologies. > 3. @both Without the topic partitions that the tasks can map back to, we > have to rely on topology/cluster metadata in each Streams instance to map > the task back. If the source topics are wild carded for e,g then each > instance could have different source topics in topology, until the next > rebalance happens. You can also read my comments from here > https://issues.apache.org/jira/browse/KAFKA-7149?focusedCommentId=16904106&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-16904106 > > > >> seems hard to imagine how encoding arbitrarily long topic names plus an > integer for the partition number could be as efficient as task ids, which > are just two integers. > 3. if you still have concerns about the efficacy of dictionary encoding, > happy to engage. The link above also has some benchmark code I used. > Theoretically, we would send each topic name atleast once, so yes if you > compare a 10-20 character topic name + an integer to two integers, it will > be more bytes. But its constant overhead proportional to size of topic name > and with 4,8,12, partitions the size difference between baseline (version 4 > where we just repeated topic names for each topic partition) and the two > approaches becomes narrow. > > >>Plus, Navinder is going to implement a bunch of protocol code that we > might just want to change when the discussion actually does take place, if > ever. > >>it'll just be a mental burden for everyone to remember that we want to > have this follow-up discussion. > 3. Is n't people changing same parts of code and tracking follow ups a > common thing, we need to deal with anyway? For this KIP, is n't it enough > to reason about whether the additional map on top of the topic dictionary > would incur more overhead than the sending task_ids? I don't think it's > case, both of them send two integers. As I see it, we can do a separate > follow up to (re)pursue the task_id conversion and get it working for both > maps within the next release? > > >>Can you elaborate on "breaking up the API"? It looks like there are > already separate API calls in the proposal, one for time-lag, and another > for offset-lag, so are they not already broken up? > The current APIs (e.g lagInfoForStore) for lags return StoreLagInfo > objects which has both time and offset lags. If we had separate APIs, say > (e.g offsetLagForStore(), timeLagForStore()), we can implement offset > version using the offset lag that the streams instance already tracks i.e > no need for external calls. The time based lag API would incur the kafka > read for the timestamp. makes sense? > > Based on the discussions so far, I only see these two pending issues to be > aligned on. Is there any other open item people want to bring up? > > On Mon, Nov 4, 2019 at 11:24 AM Sophie Blee-Goldman <sop...@confluent.io> > wrote: > >> Regarding 3) I'm wondering, does your concern still apply even now >> that the pluggable PartitionGrouper interface has been deprecated? >> Now that we can be sure that the DefaultPartitionGrouper is used to >> generate >> the taskId -> partitions mapping, we should be able to convert any taskId >> to any >> partitions. >> >> On Mon, Nov 4, 2019 at 11:17 AM John Roesler <j...@confluent.io> wrote: >> >> > Hey Vinoth, thanks for the reply! >> > >> > 3. >> > I get that it's not the main focus of this KIP, but if it's ok, it >> > would be nice to hash out this point right now. It only came up >> > because this KIP-535 is substantially extending the pattern in >> > question. If we push it off until later, then the reviewers are going >> > to have to suspend their concerns not just while voting for the KIP, >> > but also while reviewing the code. Plus, Navinder is going to >> > implement a bunch of protocol code that we might just want to change >> > when the discussion actually does take place, if ever. Finally, it'll >> > just be a mental burden for everyone to remember that we want to have >> > this follow-up discussion. >> > >> > It makes sense what you say... the specific assignment is already >> > encoded in the "main" portion of the assignment, not in the "userdata" >> > part. It also makes sense that it's simpler to reason about races if >> > you simply get all the information about the topics and partitions >> > directly from the assignor, rather than get the partition number from >> > the assignor and the topic name from your own a priori knowledge of >> > the topology. On the other hand, I'm having some trouble wrapping my >> > head around what race conditions might occur, other than the >> > fundamentally broken state in which different instances are running >> > totally different topologies. Sorry, but can you remind us of the >> > specific condition? >> > >> > To the efficiency counterargument, it seems hard to imagine how >> > encoding arbitrarily long topic names plus an integer for the >> > partition number could be as efficient as task ids, which are just two >> > integers. It seems like this would only be true if topic names were 4 >> > characters or less. >> > >> > 4. >> > Yeah, clearly, it would not be a good idea to query the metadata >> > before every single IQ query. I think there are plenty of established >> > patterns for distributed database clients to follow. Can you elaborate >> > on "breaking up the API"? It looks like there are already separate API >> > calls in the proposal, one for time-lag, and another for offset-lag, >> > so are they not already broken up? FWIW, yes, I agree, the offset lag >> > is already locally known, so we don't need to build in an extra >> > synchronous broker API call, just one for the time-lag. >> > >> > Thanks again for the discussion, >> > -John >> > >> > On Mon, Nov 4, 2019 at 11:17 AM Vinoth Chandar <vchan...@confluent.io> >> > wrote: >> > > >> > > 3. Right now, we still get the topic partitions assigned as a part of >> the >> > > top level Assignment object (the one that wraps AssignmentInfo) and >> use >> > > that to convert taskIds back. This list of only contains assignments >> for >> > > that particular instance. Attempting to also reverse map for "all" the >> > > tasksIds in the streams cluster i.e all the topic partitions in these >> > > global assignment maps was what was problematic. By explicitly sending >> > the >> > > global assignment maps as actual topic partitions, group coordinator >> > (i.e >> > > the leader that computes the assignment's ) is able to consistently >> > enforce >> > > its view of the topic metadata. Still don't think doing such a change >> > that >> > > forces you to reconsider semantics, is not needed to save bits on >> wire. >> > May >> > > be we can discuss this separately from this KIP? >> > > >> > > 4. There needs to be some caching/interval somewhere though since we >> > don't >> > > want to make 1 kafka read per 1 IQ potentially. But I think its a >> valid >> > > suggestion, to make this call just synchronous and leave the caching >> or >> > how >> > > often you want to call to the application. Would it be good to then >> break >> > > up the APIs for time and offset based lag? We can obtain offset based >> > lag >> > > for free? Only incur the overhead of reading kafka if we want time >> > > based lags? >> > > >> > > On Fri, Nov 1, 2019 at 2:49 PM Sophie Blee-Goldman < >> sop...@confluent.io> >> > > wrote: >> > > >> > > > Adding on to John's response to 3), can you clarify when and why >> > exactly we >> > > > cannot >> > > > convert between taskIds and partitions? If that's really the case I >> > don't >> > > > feel confident >> > > > that the StreamsPartitionAssignor is not full of bugs... >> > > > >> > > > It seems like it currently just encodes a list of all partitions >> (the >> > > > assignment) and also >> > > > a list of the corresponding task ids, duplicated to ensure each >> > partition >> > > > has the corresponding >> > > > taskId at the same offset into the list. Why is that problematic? >> > > > >> > > > >> > > > On Fri, Nov 1, 2019 at 12:39 PM John Roesler <j...@confluent.io> >> > wrote: >> > > > >> > > > > Thanks, all, for considering the points! >> > > > > >> > > > > 3. Interesting. I have a vague recollection of that... Still, >> though, >> > > > > it seems a little fishy. After all, we return the assignments >> > > > > themselves as task ids, and the members have to map these to topic >> > > > > partitions in order to configure themselves properly. If it's too >> > > > > complicated to get this right, then how do we know that Streams is >> > > > > computing the correct partitions at all? >> > > > > >> > > > > 4. How about just checking the log-end timestamp when you call the >> > > > > method? Then, when you get an answer, it's as fresh as it could >> > > > > possibly be. And as a user you have just one, obvious, "knob" to >> > > > > configure how much overhead you want to devote to checking... If >> you >> > > > > want to call the broker API less frequently, you just call the >> > Streams >> > > > > API less frequently. And you don't have to worry about the >> > > > > relationship between your invocations of that method and the >> config >> > > > > setting (e.g., you'll never get a negative number, which you >> could if >> > > > > you check the log-end timestamp less frequently than you check the >> > > > > lag). >> > > > > >> > > > > Thanks, >> > > > > -John >> > > > > >> > > > > On Thu, Oct 31, 2019 at 11:52 PM Navinder Brar >> > > > > <navinder_b...@yahoo.com.invalid> wrote: >> > > > > > >> > > > > > Thanks John for going through this. >> > > > > > >> > > > > > - +1, makes sense >> > > > > > - +1, no issues there >> > > > > > - Yeah the initial patch I had submitted for K-7149( >> > > > > https://github.com/apache/kafka/pull/6935) to reduce >> assignmentInfo >> > > > > object had taskIds but the merged PR had similar size according to >> > Vinoth >> > > > > and it was simpler so if the end result is of same size, it would >> not >> > > > make >> > > > > sense to pivot from dictionary and again move to taskIDs. >> > > > > > - Not sure about what a good default would be if we don't >> have a >> > > > > configurable setting. This gives the users the flexibility to the >> > users >> > > > to >> > > > > serve their requirements as at the end of the day it would take >> CPU >> > > > cycles. >> > > > > I am ok with starting it with a default and see how it goes based >> > upon >> > > > > feedback. >> > > > > > >> > > > > > Thanks, >> > > > > > Navinder >> > > > > > On Friday, 1 November, 2019, 03:46:42 am IST, Vinoth >> Chandar < >> > > > > vchan...@confluent.io> wrote: >> > > > > > >> > > > > > 1. Was trying to spell them out separately. but makes sense for >> > > > > > readability. done >> > > > > > >> > > > > > 2. No I immediately agree :) .. makes sense. @navinder? >> > > > > > >> > > > > > 3. I actually attempted only sending taskIds while working on >> > > > KAFKA-7149. >> > > > > > Its non-trivial to handle edges cases resulting from newly added >> > topic >> > > > > > partitions and wildcarded topic entries. I ended up simplifying >> it >> > to >> > > > > just >> > > > > > dictionary encoding the topic names to reduce size. We can apply >> > the >> > > > same >> > > > > > technique here for this map. Additionally, we could also >> dictionary >> > > > > encode >> > > > > > HostInfo, given its now repeated twice. I think this would save >> > more >> > > > > space >> > > > > > than having a flag per topic partition entry. Lmk if you are >> okay >> > with >> > > > > > this. >> > > > > > >> > > > > > 4. This opens up a good discussion. Given we support time lag >> > estimates >> > > > > > also, we need to read the tail record of the changelog >> periodically >> > > > > (unlike >> > > > > > offset lag, which we can potentially piggyback on metadata in >> > > > > > ConsumerRecord IIUC). we thought we should have a config that >> > control >> > > > how >> > > > > > often this read happens? Let me know if there is a simple way to >> > get >> > > > > > timestamp value of the tail record that we are missing. >> > > > > > >> > > > > > On Thu, Oct 31, 2019 at 12:58 PM John Roesler < >> j...@confluent.io> >> > > > wrote: >> > > > > > >> > > > > > > Hey Navinder, >> > > > > > > >> > > > > > > Thanks for updating the KIP, it's a lot easier to see the >> current >> > > > > > > state of the proposal now. >> > > > > > > >> > > > > > > A few remarks: >> > > > > > > 1. I'm sure it was just an artifact of revisions, but you have >> > two >> > > > > > > separate sections where you list additions to the KafkaStreams >> > > > > > > interface. Can you consolidate those so we can see all the >> > additions >> > > > > > > at once? >> > > > > > > >> > > > > > > 2. For messageLagEstimate, can I suggest "offsetLagEstimate" >> > instead, >> > > > > > > to be clearer that we're specifically measuring a number of >> > offsets? >> > > > > > > If you don't immediately agree, then I'd at least point out >> that >> > we >> > > > > > > usually refer to elements of Kafka topics as "records", not >> > > > > > > "messages", so "recordLagEstimate" might be more appropriate. >> > > > > > > >> > > > > > > 3. The proposal mentions adding a map of the standby >> > _partitions_ for >> > > > > > > each host to AssignmentInfo. I assume this is designed to >> mirror >> > the >> > > > > > > existing "partitionsByHost" map. To keep the size of these >> > metadata >> > > > > > > messages down, maybe we can consider making two changes: >> > > > > > > (a) for both actives and standbys, encode the _task ids_ >> instead >> > of >> > > > > > > _partitions_. Every member of the cluster has a copy of the >> > topology, >> > > > > > > so they can convert task ids into specific partitions on their >> > own, >> > > > > > > and task ids are only (usually) three characters. >> > > > > > > (b) instead of encoding two maps (hostinfo -> actives AND >> > hostinfo -> >> > > > > > > standbys), which requires serializing all the hostinfos twice, >> > maybe >> > > > > > > we can pack them together in one map with a structured value >> > > > (hostinfo >> > > > > > > -> [actives,standbys]). >> > > > > > > Both of these ideas still require bumping the protocol version >> > to 6, >> > > > > > > and they basically mean we drop the existing >> `PartitionsByHost` >> > field >> > > > > > > and add a new `TasksByHost` field with the structured value I >> > > > > > > mentioned. >> > > > > > > >> > > > > > > 4. Can we avoid adding the new "lag refresh" config? The lags >> > would >> > > > > > > necessarily be approximate anyway, so adding the config seems >> to >> > > > > > > increase the operational complexity of the system for little >> > actual >> > > > > > > benefit. >> > > > > > > >> > > > > > > Thanks for the pseudocode, by the way, it really helps >> visualize >> > how >> > > > > > > these new interfaces would play together. And thanks again for >> > the >> > > > > > > update! >> > > > > > > -John >> > > > > > > >> > > > > > > On Thu, Oct 31, 2019 at 2:41 PM John Roesler < >> j...@confluent.io> >> > > > > wrote: >> > > > > > > > >> > > > > > > > Hey Vinoth, >> > > > > > > > >> > > > > > > > I started going over the KIP again yesterday. There are a >> lot >> > of >> > > > > > > > updates, and I didn't finish my feedback in one day. I'm >> > working on >> > > > > it >> > > > > > > > now. >> > > > > > > > >> > > > > > > > Thanks, >> > > > > > > > John >> > > > > > > > >> > > > > > > > On Thu, Oct 31, 2019 at 11:42 AM Vinoth Chandar < >> > > > > vchan...@confluent.io> >> > > > > > > wrote: >> > > > > > > > > >> > > > > > > > > Wondering if anyone has thoughts on these changes? I liked >> > that >> > > > > the new >> > > > > > > > > metadata fetch APIs provide all the information at once >> with >> > > > > consistent >> > > > > > > > > naming.. >> > > > > > > > > >> > > > > > > > > Any guidance on what you would like to be discussed or >> > fleshed >> > > > out >> > > > > more >> > > > > > > > > before we call a VOTE? >> > > > > > > > > >> > > > > > > > > On Wed, Oct 30, 2019 at 10:31 AM Navinder Brar >> > > > > > > > > <navinder_b...@yahoo.com.invalid> wrote: >> > > > > > > > > >> > > > > > > > > > Hi, >> > > > > > > > > > We have made some edits in the KIP( >> > > > > > > > > > >> > > > > > > >> > > > > >> > > > >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-535%3A+Allow+state+stores+to+serve+stale+reads+during+rebalance >> > > > > > > ) >> > > > > > > > > > after due deliberation on the agreed design to support >> the >> > new >> > > > > query >> > > > > > > > > > design. This includes the new public API to query >> > offset/time >> > > > lag >> > > > > > > > > > information and other details related to querying >> standby >> > tasks >> > > > > > > which have >> > > > > > > > > > come up after thinking of thorough details. >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > - Addition of new config, “lag.fetch.interval.ms” to >> > > > > configure >> > > > > > > the >> > > > > > > > > > interval of time/offset lag >> > > > > > > > > > - Addition of new class StoreLagInfo to store the >> > > > periodically >> > > > > > > obtained >> > > > > > > > > > time/offset lag >> > > > > > > > > > - Addition of two new functions in KafkaStreams, >> > > > > > > List<StoreLagInfo> >> > > > > > > > > > allLagInfo() and List<StoreLagInfo> >> lagInfoForStore(String >> > > > > > > storeName) to >> > > > > > > > > > return the lag information for an instance and a store >> > > > > respectively >> > > > > > > > > > - Addition of new class KeyQueryMetadata. We need >> > > > > topicPartition >> > > > > > > for >> > > > > > > > > > each key to be matched with the lag API for the topic >> > > > partition. >> > > > > One >> > > > > > > way is >> > > > > > > > > > to add new functions and fetch topicPartition from >> > > > > > > StreamsMetadataState but >> > > > > > > > > > we thought having one call and fetching StreamsMetadata >> and >> > > > > > > topicPartition >> > > > > > > > > > is more cleaner. >> > > > > > > > > > - >> > > > > > > > > > Renaming partitionsForHost to activePartitionsForHost in >> > > > > > > StreamsMetadataState >> > > > > > > > > > and partitionsByHostState to activePartitionsByHostState >> > > > > > > > > > in StreamsPartitionAssignor >> > > > > > > > > > - We have also added the pseudo code of how all the >> > changes >> > > > > will >> > > > > > > exist >> > > > > > > > > > together and support the new querying APIs >> > > > > > > > > > >> > > > > > > > > > Please let me know if anything is pending now, before a >> > vote >> > > > can >> > > > > be >> > > > > > > > > > started on this. On Saturday, 26 October, 2019, >> 05:41:44 >> > pm >> > > > IST, >> > > > > > > Navinder >> > > > > > > > > > Brar <navinder_b...@yahoo.com.invalid> wrote: >> > > > > > > > > > >> > > > > > > > > > >> Since there are two soft votes for separate >> > active/standby >> > > > > API >> > > > > > > > > > methods, I also change my position on that. Fine with 2 >> > > > separate >> > > > > > > > > > methods. Once we remove the lag information from these >> > APIs, >> > > > > > > returning a >> > > > > > > > > > List is less attractive, since the ordering has no >> special >> > > > > meaning >> > > > > > > now. >> > > > > > > > > > Agreed, now that we are not returning lag, I am also >> sold >> > on >> > > > > having >> > > > > > > two >> > > > > > > > > > separate functions. We already have one which returns >> > > > > > > streamsMetadata for >> > > > > > > > > > active tasks, and now we can add another one for >> standbys. >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > On Saturday, 26 October, 2019, 03:55:16 am IST, >> Vinoth >> > > > > Chandar < >> > > > > > > > > > vchan...@confluent.io> wrote: >> > > > > > > > > > >> > > > > > > > > > +1 to Sophie's suggestion. Having both lag in terms of >> > time >> > > > and >> > > > > > > offsets is >> > > > > > > > > > good and makes for a more complete API. >> > > > > > > > > > >> > > > > > > > > > Since there are two soft votes for separate >> active/standby >> > API >> > > > > > > methods, I >> > > > > > > > > > also change my position on that. Fine with 2 separate >> > methods. >> > > > > > > > > > Once we remove the lag information from these APIs, >> > returning a >> > > > > List >> > > > > > > is >> > > > > > > > > > less attractive, since the ordering has no special >> meaning >> > now. >> > > > > > > > > > >> > > > > > > > > > >> lag in offsets vs time: Having both, as suggested by >> > Sophie >> > > > > would >> > > > > > > of >> > > > > > > > > > course be best. What is a little unclear to me is, how >> in >> > > > details >> > > > > > > are we >> > > > > > > > > > going to compute both? >> > > > > > > > > > @navinder may be next step is to flesh out these details >> > and >> > > > > surface >> > > > > > > any >> > > > > > > > > > larger changes we need to make if need be. >> > > > > > > > > > >> > > > > > > > > > Any other details we need to cover, before a VOTE can be >> > called >> > > > > on >> > > > > > > this? >> > > > > > > > > > >> > > > > > > > > > >> > > > > > > > > > On Fri, Oct 25, 2019 at 1:51 PM Bill Bejeck < >> > bbej...@gmail.com >> > > > > >> > > > > > > wrote: >> > > > > > > > > > >> > > > > > > > > > > I am jumping in a little late here. >> > > > > > > > > > > >> > > > > > > > > > > Overall I agree with the proposal to push decision >> > making on >> > > > > > > what/how to >> > > > > > > > > > > query in the query layer. >> > > > > > > > > > > >> > > > > > > > > > > For point 5 from above, I'm slightly in favor of >> having >> > a new >> > > > > > > method, >> > > > > > > > > > > "standbyMetadataForKey()" or something similar. >> > > > > > > > > > > Because even if we return all tasks in one list, the >> user >> > > > will >> > > > > > > still have >> > > > > > > > > > > to perform some filtering to separate the different >> > tasks, >> > > > so I >> > > > > > > don't >> > > > > > > > > > feel >> > > > > > > > > > > making two calls is a burden, and IMHO makes things >> more >> > > > > > > transparent for >> > > > > > > > > > > the user. >> > > > > > > > > > > If the final vote is for using an "isActive" field, >> I'm >> > good >> > > > > with >> > > > > > > that as >> > > > > > > > > > > well. >> > > > > > > > > > > >> > > > > > > > > > > Just my 2 cents. >> > > > > > > > > > > >> > > > > > > > > > > On Fri, Oct 25, 2019 at 5:09 AM Navinder Brar >> > > > > > > > > > > <navinder_b...@yahoo.com.invalid> wrote: >> > > > > > > > > > > >> > > > > > > > > > > > I think now we are aligned on almost all the design >> > parts. >> > > > > > > Summarising >> > > > > > > > > > > > below what has been discussed above and we have a >> > general >> > > > > > > consensus on. >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > - Rather than broadcasting lag across all nodes >> at >> > > > > > > rebalancing/with >> > > > > > > > > > > the >> > > > > > > > > > > > heartbeat, we will just return a list of all >> available >> > > > > standby’s >> > > > > > > in the >> > > > > > > > > > > > system and the user can make IQ query any of those >> > nodes >> > > > > which >> > > > > > > will >> > > > > > > > > > > return >> > > > > > > > > > > > the response, and the lag and offset time. Based on >> > which >> > > > > user >> > > > > > > can >> > > > > > > > > > decide >> > > > > > > > > > > > if he wants to return the response back or call >> another >> > > > > standby. >> > > > > > > > > > > > - The current metadata query frequency will not >> > change. >> > > > > It >> > > > > > > will be >> > > > > > > > > > > the >> > > > > > > > > > > > same as it does now, i.e. before each query. >> > > > > > > > > > > > >> > > > > > > > > > > > - For fetching list<StreamsMetadata> in >> > > > > > > StreamsMetadataState.java >> > > > > > > > > > and >> > > > > > > > > > > > List<QueryableStoreProvider> in >> > > > > > > StreamThreadStateStoreProvider.java >> > > > > > > > > > > (which >> > > > > > > > > > > > will return all active stores which are >> > running/restoring >> > > > and >> > > > > > > replica >> > > > > > > > > > > > stores which are running), we will add new functions >> > and >> > > > not >> > > > > > > disturb >> > > > > > > > > > the >> > > > > > > > > > > > existing functions >> > > > > > > > > > > > >> > > > > > > > > > > > - There is no need to add new StreamsConfig for >> > > > > implementing >> > > > > > > this >> > > > > > > > > > KIP >> > > > > > > > > > > > >> > > > > > > > > > > > - We will add standbyPartitionsByHost in >> > AssignmentInfo >> > > > > and >> > > > > > > > > > > > StreamsMetadataState which would change the existing >> > > > > > > rebuildMetadata() >> > > > > > > > > > > and >> > > > > > > > > > > > setPartitionsByHostState() >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > If anyone has any more concerns please feel free to >> > add. >> > > > Post >> > > > > > > this I >> > > > > > > > > > will >> > > > > > > > > > > > be initiating a vote. >> > > > > > > > > > > > ~Navinder >> > > > > > > > > > > > >> > > > > > > > > > > > On Friday, 25 October, 2019, 12:05:29 pm IST, >> > Matthias >> > > > J. >> > > > > Sax >> > > > > > > < >> > > > > > > > > > > > matth...@confluent.io> wrote: >> > > > > > > > > > > > >> > > > > > > > > > > > Just to close the loop @Vinoth: >> > > > > > > > > > > > >> > > > > > > > > > > > > 1. IIUC John intends to add (or we can do this in >> > this >> > > > > KIP) lag >> > > > > > > > > > > > information >> > > > > > > > > > > > > to AssignmentInfo, which gets sent to every >> > participant. >> > > > > > > > > > > > >> > > > > > > > > > > > As explained by John, currently KIP-441 plans to >> only >> > > > report >> > > > > the >> > > > > > > > > > > > information to the leader. But I guess, with the new >> > > > > proposal to >> > > > > > > not >> > > > > > > > > > > > broadcast this information anyway, this concern is >> > > > > invalidated >> > > > > > > anyway >> > > > > > > > > > > > >> > > > > > > > > > > > > 2. At-least I was under the assumption that it >> can be >> > > > > called >> > > > > > > per >> > > > > > > > > > query, >> > > > > > > > > > > > > since the API docs don't seem to suggest >> otherwise. >> > Do >> > > > you >> > > > > see >> > > > > > > any >> > > > > > > > > > > > > potential issues if we call this every query? (we >> > should >> > > > > > > benchmark >> > > > > > > > > > this >> > > > > > > > > > > > > nonetheless) >> > > > > > > > > > > > >> > > > > > > > > > > > I did not see a real issue if people refresh the >> > metadata >> > > > > > > frequently, >> > > > > > > > > > > > because it would be a local call. My main point was, >> > that >> > > > > this >> > > > > > > would >> > > > > > > > > > > > change the current usage pattern of the API, and we >> > would >> > > > > > > clearly need >> > > > > > > > > > > > to communicate this change. Similar to (1), this >> > concern in >> > > > > > > invalidated >> > > > > > > > > > > > anyway. >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > @John: I think it's a great idea to get rid of >> > reporting >> > > > > lag, and >> > > > > > > > > > > > pushing the decision making process about "what to >> > query" >> > > > > into >> > > > > > > the >> > > > > > > > > > query >> > > > > > > > > > > > serving layer itself. This simplifies the overall >> > design of >> > > > > this >> > > > > > > KIP >> > > > > > > > > > > > significantly, and actually aligns very well with >> the >> > idea >> > > > > that >> > > > > > > Kafka >> > > > > > > > > > > > Streams (as it is a library) should only provide the >> > basic >> > > > > > > building >> > > > > > > > > > > > block. Many of my raised questions are invalided by >> > this. >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > Some questions are still open though: >> > > > > > > > > > > > >> > > > > > > > > > > > > 10) Do we need to distinguish between >> > active(restoring) >> > > > and >> > > > > > > standby >> > > > > > > > > > > > > tasks? Or could be treat both as the same? >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > @Vinoth: about (5). I see your point about multiple >> > calls >> > > > vs >> > > > > a >> > > > > > > single >> > > > > > > > > > > > call. I still slightly prefer multiple calls, but >> it's >> > > > highly >> > > > > > > > > > subjective >> > > > > > > > > > > > and I would also be fine to add an #isActive() >> method. >> > > > Would >> > > > > be >> > > > > > > good >> > > > > > > > > > the >> > > > > > > > > > > > get feedback from others. >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > For (14), ie, lag in offsets vs time: Having both, >> as >> > > > > suggested >> > > > > > > by >> > > > > > > > > > > > Sophie would of course be best. What is a little >> > unclear to >> > > > > me >> > > > > > > is, how >> > > > > > > > > > > > in details are we going to compute both? >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > -Matthias >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > > On 10/24/19 11:07 PM, Sophie Blee-Goldman wrote: >> > > > > > > > > > > > > Just to chime in on the "report lag vs timestamp >> > > > > difference" >> > > > > > > issue, I >> > > > > > > > > > > > would >> > > > > > > > > > > > > actually advocate for both. As mentioned already, >> > time >> > > > > > > difference is >> > > > > > > > > > > > > probably a lot easier and/or more useful to reason >> > about >> > > > in >> > > > > > > terms of >> > > > > > > > > > > > > "freshness" >> > > > > > > > > > > > > of the state. But in the case when all queried >> > stores are >> > > > > far >> > > > > > > behind, >> > > > > > > > > > > lag >> > > > > > > > > > > > > could >> > > > > > > > > > > > > be used to estimate the recovery velocity. You can >> > then >> > > > > get a >> > > > > > > (pretty >> > > > > > > > > > > > rough) >> > > > > > > > > > > > > idea of when a store might be ready, and wait >> until >> > > > around >> > > > > > > then to >> > > > > > > > > > > query >> > > > > > > > > > > > > again. >> > > > > > > > > > > > > >> > > > > > > > > > > > > On Thu, Oct 24, 2019 at 9:53 PM Guozhang Wang < >> > > > > > > wangg...@gmail.com> >> > > > > > > > > > > > wrote: >> > > > > > > > > > > > > >> > > > > > > > > > > > >> I think I agree with John's recent reasoning as >> > well: >> > > > > instead >> > > > > > > of >> > > > > > > > > > > letting >> > > > > > > > > > > > >> the storeMetadataAPI to return the staleness >> > > > information, >> > > > > > > letting >> > > > > > > > > > the >> > > > > > > > > > > > >> client to query either active or standby and >> letting >> > > > > standby >> > > > > > > query >> > > > > > > > > > > > response >> > > > > > > > > > > > >> to include both the values + timestamp (or lag, >> as >> > in >> > > > > diffs of >> > > > > > > > > > > > timestamps) >> > > > > > > > > > > > >> would actually be more intuitive -- not only the >> > streams >> > > > > > > client is >> > > > > > > > > > > > simpler, >> > > > > > > > > > > > >> from user's perspective they also do not need to >> > > > > periodically >> > > > > > > > > > refresh >> > > > > > > > > > > > their >> > > > > > > > > > > > >> staleness information from the client, but only >> > need to >> > > > > make >> > > > > > > > > > decisions >> > > > > > > > > > > > on >> > > > > > > > > > > > >> the fly whenever they need to query. >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> Again the standby replica then need to know the >> > current >> > > > > active >> > > > > > > > > > task's >> > > > > > > > > > > > >> timestamp, which can be found from the log end >> > record's >> > > > > > > encoded >> > > > > > > > > > > > timestamp; >> > > > > > > > > > > > >> today we standby tasks do not read that specific >> > record, >> > > > > but >> > > > > > > only >> > > > > > > > > > > > refresh >> > > > > > > > > > > > >> its knowledge on the log end offset, but I think >> > > > > refreshing >> > > > > > > the >> > > > > > > > > > latest >> > > > > > > > > > > > >> record timestamp is not a very bad request to add >> > on the >> > > > > > > standby >> > > > > > > > > > > > replicas. >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> Guozhang >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> On Thu, Oct 24, 2019 at 5:43 PM Vinoth Chandar < >> > > > > > > > > > vchan...@confluent.io >> > > > > > > > > > > > >> > > > > > > > > > > > >> wrote: >> > > > > > > > > > > > >> >> > > > > > > > > > > > >>> +1 As someone implementing a query routing >> layer, >> > there >> > > > > is >> > > > > > > already >> > > > > > > > > > a >> > > > > > > > > > > > need >> > > > > > > > > > > > >>> to have mechanisms in place to do >> > healthchecks/failure >> > > > > > > detection to >> > > > > > > > > > > > >> detect >> > > > > > > > > > > > >>> failures for queries, while Streams rebalancing >> > > > > eventually >> > > > > > > kicks in >> > > > > > > > > > > the >> > > > > > > > > > > > >>> background. >> > > > > > > > > > > > >>> So, pushing this complexity to the IQ client app >> > keeps >> > > > > > > Streams >> > > > > > > > > > > simpler >> > > > > > > > > > > > as >> > > > > > > > > > > > >>> well. IQs will be potentially issues at an >> order of >> > > > > > > magnitude more >> > > > > > > > > > > > >>> frequently and it can achieve good freshness for >> > the >> > > > lag >> > > > > > > > > > information. >> > > > > > > > > > > > >>> >> > > > > > > > > > > > >>> I would like to add however, that we would also >> > need to >> > > > > > > introduce >> > > > > > > > > > > apis >> > > > > > > > > > > > in >> > > > > > > > > > > > >>> KafkaStreams class, for obtaining lag >> information >> > for >> > > > all >> > > > > > > stores >> > > > > > > > > > > local >> > > > > > > > > > > > to >> > > > > > > > > > > > >>> that host. This is for the IQs to relay back >> with >> > the >> > > > > > > response/its >> > > > > > > > > > > own >> > > > > > > > > > > > >>> heartbeat mechanism. >> > > > > > > > > > > > >>> >> > > > > > > > > > > > >>> On Thu, Oct 24, 2019 at 3:12 PM John Roesler < >> > > > > > > j...@confluent.io> >> > > > > > > > > > > > wrote: >> > > > > > > > > > > > >>> >> > > > > > > > > > > > >>>> Hi all, >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> I've been mulling about this KIP, and I think I >> > was on >> > > > > the >> > > > > > > wrong >> > > > > > > > > > > track >> > > > > > > > > > > > >>>> earlier with regard to task lags. Tl;dr: I >> don't >> > think >> > > > > we >> > > > > > > should >> > > > > > > > > > add >> > > > > > > > > > > > >>>> lags at all to the metadata API (and also not >> to >> > the >> > > > > > > > > > AssignmentInfo >> > > > > > > > > > > > >>>> protocol message). >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> Like I mentioned early on, reporting lag via >> > > > > > > > > > > > >>>> SubscriptionInfo/AssignmentInfo would only work >> > while >> > > > > > > rebalances >> > > > > > > > > > are >> > > > > > > > > > > > >>>> happening. Once the group stabilizes, no >> members >> > would >> > > > > be >> > > > > > > notified >> > > > > > > > > > > of >> > > > > > > > > > > > >>>> each others' lags anymore. I had been thinking >> > that >> > > > the >> > > > > > > solution >> > > > > > > > > > > would >> > > > > > > > > > > > >>>> be the heartbeat proposal I mentioned earlier, >> but >> > > > that >> > > > > > > proposal >> > > > > > > > > > > would >> > > > > > > > > > > > >>>> have reported the heartbeats of the members >> only >> > to >> > > > the >> > > > > > > leader >> > > > > > > > > > > member >> > > > > > > > > > > > >>>> (the one who makes assignments). To be useful >> in >> > the >> > > > > > > context of >> > > > > > > > > > > _this_ >> > > > > > > > > > > > >>>> KIP, we would also have to report the lags in >> the >> > > > > heartbeat >> > > > > > > > > > > responses >> > > > > > > > > > > > >>>> to of _all_ members. This is a concern to be >> > because >> > > > now >> > > > > > > _all_ the >> > > > > > > > > > > > >>>> lags get reported to _all_ the members on >> _every_ >> > > > > > > heartbeat... a >> > > > > > > > > > lot >> > > > > > > > > > > > >>>> of chatter. >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> Plus, the proposal for KIP-441 is only to >> report >> > the >> > > > > lags >> > > > > > > of each >> > > > > > > > > > > > >>>> _task_. This is the sum of the lags of all the >> > stores >> > > > > in the >> > > > > > > > > > tasks. >> > > > > > > > > > > > >>>> But this would be insufficient for KIP-535. For >> > this >> > > > > kip, >> > > > > > > we would >> > > > > > > > > > > > >>>> want the lag specifically of the store we want >> to >> > > > > query. So >> > > > > > > this >> > > > > > > > > > > > >>>> means, we have to report the lags of all the >> > stores of >> > > > > all >> > > > > > > the >> > > > > > > > > > > members >> > > > > > > > > > > > >>>> to every member... even more chatter! >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> The final nail in the coffin to me is that IQ >> > clients >> > > > > would >> > > > > > > have >> > > > > > > > > > to >> > > > > > > > > > > > >>>> start refreshing their metadata quite >> frequently >> > to >> > > > > stay up >> > > > > > > to >> > > > > > > > > > date >> > > > > > > > > > > on >> > > > > > > > > > > > >>>> the lags, which adds even more overhead to the >> > system. >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> Consider a strawman alternative: we bring >> KIP-535 >> > back >> > > > > to >> > > > > > > > > > extending >> > > > > > > > > > > > >>>> the metadata API to tell the client the active >> and >> > > > > standby >> > > > > > > > > > replicas >> > > > > > > > > > > > >>>> for the key in question (not including and >> > > > > "staleness/lag" >> > > > > > > > > > > > >>>> restriction, just returning all the replicas). >> > Then, >> > > > the >> > > > > > > client >> > > > > > > > > > > picks >> > > > > > > > > > > > >>>> a replica and sends the query. The server >> returns >> > the >> > > > > > > current lag >> > > > > > > > > > > > >>>> along with the response (maybe in an HTML >> header >> > or >> > > > > > > something). >> > > > > > > > > > > Then, >> > > > > > > > > > > > >>>> the client keeps a map of its last observed >> lags >> > for >> > > > > each >> > > > > > > replica, >> > > > > > > > > > > and >> > > > > > > > > > > > >>>> uses this information to prefer fresher >> replicas. >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> OR, if it wants only to query the active >> replica, >> > it >> > > > > would >> > > > > > > throw >> > > > > > > > > > an >> > > > > > > > > > > > >>>> error on any lag response greater than zero, >> > refreshes >> > > > > its >> > > > > > > > > > metadata >> > > > > > > > > > > by >> > > > > > > > > > > > >>>> re-querying the metadata API, and tries again >> > with the >> > > > > > > current >> > > > > > > > > > > active >> > > > > > > > > > > > >>>> replica. >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> This way, the lag information will be super >> fresh >> > for >> > > > > the >> > > > > > > client, >> > > > > > > > > > > and >> > > > > > > > > > > > >>>> we keep the Metadata API / >> > Assignment,Subscription / >> > > > and >> > > > > > > Heartbeat >> > > > > > > > > > > as >> > > > > > > > > > > > >>>> slim as possible. >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> Side note: I do think that some time soon, >> we'll >> > have >> > > > to >> > > > > > > add a >> > > > > > > > > > > library >> > > > > > > > > > > > >>>> for IQ server/clients. I think that this logic >> > will >> > > > > start >> > > > > > > to get >> > > > > > > > > > > > >>>> pretty complex. >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> I hope this thinking is reasonably clear! >> > > > > > > > > > > > >>>> Thanks again, >> > > > > > > > > > > > >>>> -John >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> Does that >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> On Wed, Oct 23, 2019 at 10:16 AM Vinoth >> Chandar < >> > > > > > > > > > > > vchan...@confluent.io >> > > > > > > > > > > > >>> >> > > > > > > > > > > > >>>> wrote: >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>> Responding to the points raised by Matthias >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>> 1. IIUC John intends to add (or we can do >> this in >> > > > this >> > > > > > > KIP) lag >> > > > > > > > > > > > >>>> information >> > > > > > > > > > > > >>>>> to AssignmentInfo, which gets sent to every >> > > > > participant. >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>> 2. At-least I was under the assumption that it >> > can be >> > > > > > > called per >> > > > > > > > > > > > >> query, >> > > > > > > > > > > > >>>>> since the API docs don't seem to suggest >> > otherwise. >> > > > Do >> > > > > you >> > > > > > > see >> > > > > > > > > > any >> > > > > > > > > > > > >>>>> potential issues if we call this every query? >> (we >> > > > > should >> > > > > > > > > > benchmark >> > > > > > > > > > > > >> this >> > > > > > > > > > > > >>>>> nonetheless) >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>> 4. Agree. metadataForKey() implicitly would >> > return >> > > > the >> > > > > > > active >> > > > > > > > > > host >> > > > > > > > > > > > >>>> metadata >> > > > > > > > > > > > >>>>> (as it was before). We should also document >> this >> > in >> > > > > that >> > > > > > > APIs >> > > > > > > > > > > > >> javadoc, >> > > > > > > > > > > > >>>>> given we have another method(s) that returns >> more >> > > > host >> > > > > > > metadata >> > > > > > > > > > > now. >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>> 5. While I see the point, the app/caller has >> to >> > make >> > > > > two >> > > > > > > > > > different >> > > > > > > > > > > > >>> APIs >> > > > > > > > > > > > >>>>> calls to obtain active/standby and potentially >> > do the >> > > > > same >> > > > > > > set of >> > > > > > > > > > > > >>>> operation >> > > > > > > > > > > > >>>>> to query the state. I personally still like a >> > method >> > > > > like >> > > > > > > > > > > isActive() >> > > > > > > > > > > > >>>>> better, but don't have strong opinions. >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>> 9. If we do expose the lag information, could >> we >> > just >> > > > > > > leave it >> > > > > > > > > > upto >> > > > > > > > > > > > >> to >> > > > > > > > > > > > >>>> the >> > > > > > > > > > > > >>>>> caller to decide whether it errors out or not >> > and not >> > > > > make >> > > > > > > the >> > > > > > > > > > > > >> decision >> > > > > > > > > > > > >>>>> within Streams? i.e we don't need a new config >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>> 14. +1 . If it's easier to do right away. We >> > started >> > > > > with >> > > > > > > number >> > > > > > > > > > of >> > > > > > > > > > > > >>>>> records, following the lead from KIP-441 >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>> On Wed, Oct 23, 2019 at 5:44 AM Navinder Brar >> > > > > > > > > > > > >>>>> <navinder_b...@yahoo.com.invalid> wrote: >> > > > > > > > > > > > >>>>> >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>> Thanks, everyone for taking a look. Some very >> > cool >> > > > > ideas >> > > > > > > have >> > > > > > > > > > > flown >> > > > > > > > > > > > >>> in. >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>>>> There was a follow-on idea I POCed to >> > continuously >> > > > > > > share lag >> > > > > > > > > > > > >>>>>> information in the heartbeat protocol+1 that >> > would >> > > > be >> > > > > > > great, I >> > > > > > > > > > > will >> > > > > > > > > > > > >>>> update >> > > > > > > > > > > > >>>>>> the KIP assuming this work will finish soon >> > > > > > > > > > > > >>>>>>>> I think that adding a new method to >> > > > > > > StreamsMetadataState and >> > > > > > > > > > > > >>>>>> deprecating the existing method isthe best >> way >> > to >> > > > go; >> > > > > we >> > > > > > > just >> > > > > > > > > > > can't >> > > > > > > > > > > > >>>> change >> > > > > > > > > > > > >>>>>> the return types of any existing methods.+1 >> on >> > this, >> > > > > we >> > > > > > > will add >> > > > > > > > > > > > >> new >> > > > > > > > > > > > >>>>>> methods for users who would be interested in >> > > > querying >> > > > > > > back a >> > > > > > > > > > list >> > > > > > > > > > > > >> of >> > > > > > > > > > > > >>>>>> possible options to query from and leave the >> > current >> > > > > > > function >> > > > > > > > > > > > >>>>>> getStreamsMetadataForKey() untouched for >> users >> > who >> > > > > want >> > > > > > > absolute >> > > > > > > > > > > > >>>>>> consistency. >> > > > > > > > > > > > >>>>>>>> why not just always return all available >> > metadata >> > > > > > > (including >> > > > > > > > > > > > >>>>>> active/standby or lag) and let the caller >> > decide to >> > > > > which >> > > > > > > node >> > > > > > > > > > > they >> > > > > > > > > > > > >>>> want to >> > > > > > > > > > > > >>>>>> route the query+1. I think this makes sense >> as >> > from >> > > > a >> > > > > user >> > > > > > > > > > > > >> standpoint >> > > > > > > > > > > > >>>> there >> > > > > > > > > > > > >>>>>> is no difference b/w an active and a standby >> if >> > both >> > > > > have >> > > > > > > same >> > > > > > > > > > > lag, >> > > > > > > > > > > > >>>> Infact >> > > > > > > > > > > > >>>>>> users would be able to use this API to reduce >> > query >> > > > > load >> > > > > > > on >> > > > > > > > > > > > >> actives, >> > > > > > > > > > > > >>> so >> > > > > > > > > > > > >>>>>> returning all available options along with >> the >> > > > current >> > > > > > > lag in >> > > > > > > > > > each >> > > > > > > > > > > > >>>> would >> > > > > > > > > > > > >>>>>> make sense and leave it to user how they want >> > to use >> > > > > this >> > > > > > > data. >> > > > > > > > > > > > >> This >> > > > > > > > > > > > >>>> has >> > > > > > > > > > > > >>>>>> another added advantage. If a user queries >> any >> > > > random >> > > > > > > machine >> > > > > > > > > > for >> > > > > > > > > > > a >> > > > > > > > > > > > >>>> key and >> > > > > > > > > > > > >>>>>> that machine has a replica for the >> > partition(where >> > > > key >> > > > > > > belongs) >> > > > > > > > > > > > >> user >> > > > > > > > > > > > >>>> might >> > > > > > > > > > > > >>>>>> choose to serve the data from there >> itself(if it >> > > > > doesn’t >> > > > > > > lag >> > > > > > > > > > much) >> > > > > > > > > > > > >>>> rather >> > > > > > > > > > > > >>>>>> than finding the active and making an IQ to >> > that. >> > > > This >> > > > > > > would >> > > > > > > > > > save >> > > > > > > > > > > > >>> some >> > > > > > > > > > > > >>>>>> critical time in serving for some >> applications. >> > > > > > > > > > > > >>>>>>>> Adding the lag in terms of timestamp diff >> > > > comparing >> > > > > the >> > > > > > > > > > > > >> committed >> > > > > > > > > > > > >>>>>> offset.+1 on this, I think it’s more >> readable. >> > But >> > > > as >> > > > > > > John said >> > > > > > > > > > > the >> > > > > > > > > > > > >>>>>> function allMetadataForKey() is just >> returning >> > the >> > > > > > > possible >> > > > > > > > > > > options >> > > > > > > > > > > > >>>> from >> > > > > > > > > > > > >>>>>> where users can query a key, so we can even >> > drop the >> > > > > > > parameter >> > > > > > > > > > > > >>>>>> enableReplicaServing/tolerableDataStaleness >> and >> > just >> > > > > > > return all >> > > > > > > > > > > the >> > > > > > > > > > > > >>>>>> streamsMetadata containing that key along >> with >> > the >> > > > > offset >> > > > > > > limit. >> > > > > > > > > > > > >>>>>> Answering the questions posted by Matthias in >> > > > > sequence. >> > > > > > > > > > > > >>>>>> 1. @John can you please comment on this >> one.2. >> > Yeah >> > > > > the >> > > > > > > usage >> > > > > > > > > > > > >> pattern >> > > > > > > > > > > > >>>>>> would include querying this prior to every >> > request >> > > > 3. >> > > > > > > Will add >> > > > > > > > > > the >> > > > > > > > > > > > >>>> changes >> > > > > > > > > > > > >>>>>> to StreamsMetadata in the KIP, would include >> > changes >> > > > > in >> > > > > > > > > > > > >>>> rebuildMetadata() >> > > > > > > > > > > > >>>>>> etc.4. Makes sense, already addressed above5. >> > Is it >> > > > > > > important >> > > > > > > > > > from >> > > > > > > > > > > > >> a >> > > > > > > > > > > > >>>> user >> > > > > > > > > > > > >>>>>> perspective if they are querying an >> > > > > active(processing), >> > > > > > > > > > > > >>>> active(restoring), >> > > > > > > > > > > > >>>>>> a standby task if we have away of denoting >> lag >> > in a >> > > > > > > readable >> > > > > > > > > > > manner >> > > > > > > > > > > > >>>> which >> > > > > > > > > > > > >>>>>> kind of signifies the user that this is the >> best >> > > > node >> > > > > to >> > > > > > > query >> > > > > > > > > > the >> > > > > > > > > > > > >>>> fresh >> > > > > > > > > > > > >>>>>> data.6. Yes, I intend to return the actives >> and >> > > > > replicas >> > > > > > > in the >> > > > > > > > > > > > >> same >> > > > > > > > > > > > >>>> return >> > > > > > > > > > > > >>>>>> list in allMetadataForKey()7. tricky8. yes, >> we >> > need >> > > > > new >> > > > > > > > > > functions >> > > > > > > > > > > > >> to >> > > > > > > > > > > > >>>> return >> > > > > > > > > > > > >>>>>> activeRestoring and standbyRunning tasks.9. >> > > > > StreamsConfig >> > > > > > > > > > doesn’t >> > > > > > > > > > > > >>> look >> > > > > > > > > > > > >>>> like >> > > > > > > > > > > > >>>>>> of much use to me since we are giving all >> > possible >> > > > > > > options via >> > > > > > > > > > > this >> > > > > > > > > > > > >>>>>> function, or they can use existing function >> > > > > > > > > > > > >>> getStreamsMetadataForKey() >> > > > > > > > > > > > >>>> and >> > > > > > > > > > > > >>>>>> get just the active10. I think treat them >> both >> > the >> > > > > same >> > > > > > > and let >> > > > > > > > > > > the >> > > > > > > > > > > > >>>> lag do >> > > > > > > > > > > > >>>>>> the talking11. We are just sending them the >> > option >> > > > to >> > > > > > > query from >> > > > > > > > > > > in >> > > > > > > > > > > > >>>>>> allMetadataForKey(), which doesn’t include >> any >> > > > > handle. We >> > > > > > > then >> > > > > > > > > > > > >> query >> > > > > > > > > > > > >>>> that >> > > > > > > > > > > > >>>>>> machine for the key where it calls >> allStores() >> > and >> > > > > tries >> > > > > > > to find >> > > > > > > > > > > > >> the >> > > > > > > > > > > > >>>> task >> > > > > > > > > > > > >>>>>> in >> activeRunning/activeRestoring/standbyRunning >> > and >> > > > > adds >> > > > > > > the >> > > > > > > > > > store >> > > > > > > > > > > > >>>> handle >> > > > > > > > > > > > >>>>>> here. 12. Need to verify, but during the >> exact >> > point >> > > > > when >> > > > > > > store >> > > > > > > > > > is >> > > > > > > > > > > > >>>> closed >> > > > > > > > > > > > >>>>>> to transition it from restoring to running >> the >> > > > queries >> > > > > > > will >> > > > > > > > > > fail. >> > > > > > > > > > > > >> The >> > > > > > > > > > > > >>>>>> caller in such case can have their own >> > configurable >> > > > > > > retries to >> > > > > > > > > > > > >> check >> > > > > > > > > > > > >>>> again >> > > > > > > > > > > > >>>>>> or try the replica if a call fails to >> active13. >> > I >> > > > > think >> > > > > > > KIP-216 >> > > > > > > > > > is >> > > > > > > > > > > > >>>> working >> > > > > > > > > > > > >>>>>> on those lines, we might not need few of >> those >> > > > > exceptions >> > > > > > > since >> > > > > > > > > > > now >> > > > > > > > > > > > >>> the >> > > > > > > > > > > > >>>>>> basic idea of this KIP is to support IQ >> during >> > > > > > > rebalancing.14. >> > > > > > > > > > > > >>>> Addressed >> > > > > > > > > > > > >>>>>> above, agreed it looks more readable. >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>> On Tuesday, 22 October, 2019, 08:39:07 pm >> > IST, >> > > > > > > Matthias J. >> > > > > > > > > > Sax >> > > > > > > > > > > > >> < >> > > > > > > > > > > > >>>>>> matth...@confluent.io> wrote: >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>> One more thought: >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>> 14) Is specifying the allowed lag in number >> of >> > > > > records a >> > > > > > > useful >> > > > > > > > > > > way >> > > > > > > > > > > > >>> for >> > > > > > > > > > > > >>>>>> users to declare how stale an instance is >> > allowed to >> > > > > be? >> > > > > > > Would >> > > > > > > > > > it >> > > > > > > > > > > > >> be >> > > > > > > > > > > > >>>>>> more intuitive for users to specify the >> allowed >> > lag >> > > > in >> > > > > > > time >> > > > > > > > > > units >> > > > > > > > > > > > >>>> (would >> > > > > > > > > > > > >>>>>> event time or processing time be better)? It >> > seems >> > > > > hard >> > > > > > > for >> > > > > > > > > > users >> > > > > > > > > > > > >> to >> > > > > > > > > > > > >>>>>> reason how "fresh" a store really is when >> > number of >> > > > > > > records is >> > > > > > > > > > > > >> used. >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>> -Matthias >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>>>> On 10/21/19 9:02 PM, Matthias J. Sax wrote: >> > > > > > > > > > > > >>>>>>> Some more follow up thoughts: >> > > > > > > > > > > > >>>>>>> >> > > > > > > > > > > > >>>>>>> 11) If we get a store handle of an >> > > > active(restoring) >> > > > > > > task, and >> > > > > > > > > > > > >> the >> > > > > > > > > > > > >>>> task >> > > > > > > > > > > > >>>>>>> transits to running, does the store handle >> > become >> > > > > > > invalid and a >> > > > > > > > > > > > >> new >> > > > > > > > > > > > >>>> one >> > > > > > > > > > > > >>>>>>> must be retrieved? Or can we "switch it out" >> > > > > underneath >> > > > > > > -- for >> > > > > > > > > > > > >> this >> > > > > > > > > > > > >>>>>>> case, how does the user know when they >> start to >> > > > > query the >> > > > > > > > > > > > >>> up-to-date >> > > > > > > > > > > > >>>>>> state? >> > > > > > > > > > > > >>>>>>> >> > > > > > > > > > > > >>>>>>> 12) Standby tasks will have the store open >> in >> > > > regular >> > > > > > > mode, >> > > > > > > > > > while >> > > > > > > > > > > > >>>>>>> active(restoring) tasks open stores in >> "upgrade >> > > > mode" >> > > > > > > for more >> > > > > > > > > > > > >>>> efficient >> > > > > > > > > > > > >>>>>>> bulk loading. When we switch the store into >> > active >> > > > > mode, >> > > > > > > we >> > > > > > > > > > close >> > > > > > > > > > > > >>> it >> > > > > > > > > > > > >>>> and >> > > > > > > > > > > > >>>>>>> reopen it. What is the impact if we query >> the >> > store >> > > > > > > during >> > > > > > > > > > > > >> restore? >> > > > > > > > > > > > >>>> What >> > > > > > > > > > > > >>>>>>> is the impact if we close the store to >> transit >> > to >> > > > > > > running (eg, >> > > > > > > > > > > > >>> there >> > > > > > > > > > > > >>>>>>> might be open iterators)? >> > > > > > > > > > > > >>>>>>> >> > > > > > > > > > > > >>>>>>> 13) Do we need to introduced new exception >> > types? >> > > > > Compare >> > > > > > > > > > KIP-216 >> > > > > > > > > > > > >>>>>>> ( >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > >> > > > > >> > > > >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-216%3A+IQ+should+throw+different+exceptions+for+different+errors >> > > > > > > > > > > > >>>>>> ) >> > > > > > > > > > > > >>>>>>> that aims to improve the user experience >> with >> > > > regard >> > > > > to >> > > > > > > IQ >> > > > > > > > > > > > >>>> exceptions. >> > > > > > > > > > > > >>>>>>> >> > > > > > > > > > > > >>>>>>> >> > > > > > > > > > > > >>>>>>> -Matthias >> > > > > > > > > > > > >>>>>>> >> > > > > > > > > > > > >>>>>>> On 10/21/19 6:39 PM, Matthias J. Sax wrote: >> > > > > > > > > > > > >>>>>>>> Thanks for the KIP. >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> Couple of comments: >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 1) With regard to KIP-441, my current >> > > > understanding >> > > > > is >> > > > > > > that >> > > > > > > > > > the >> > > > > > > > > > > > >>> lag >> > > > > > > > > > > > >>>>>>>> information is only reported to the leader >> > (please >> > > > > > > correct me >> > > > > > > > > > > > >> if I >> > > > > > > > > > > > >>>> am >> > > > > > > > > > > > >>>>>>>> wrong). This seems to be quite a >> limitation to >> > > > > actually >> > > > > > > use >> > > > > > > > > > the >> > > > > > > > > > > > >>> lag >> > > > > > > > > > > > >>>>>>>> information. >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 2) The idea of the metadata API is actually >> > to get >> > > > > > > metadata >> > > > > > > > > > once >> > > > > > > > > > > > >>> and >> > > > > > > > > > > > >>>>>>>> only refresh the metadata if a store was >> > migrated. >> > > > > The >> > > > > > > current >> > > > > > > > > > > > >>>> proposal >> > > > > > > > > > > > >>>>>>>> would require to get the metadata before >> each >> > > > query. >> > > > > > > The KIP >> > > > > > > > > > > > >>> should >> > > > > > > > > > > > >>>>>>>> describe the usage pattern and impact in >> more >> > > > > detail. >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 3) Currently, the KIP does not list the >> > public API >> > > > > > > changes in >> > > > > > > > > > > > >>>> detail. >> > > > > > > > > > > > >>>>>>>> Please list all methods you intend to >> > deprecate >> > > > and >> > > > > > > list all >> > > > > > > > > > > > >>>> methods you >> > > > > > > > > > > > >>>>>>>> intend to add (best, using a code-block >> > markup -- >> > > > > > > compare >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > >> > > > > >> > > > >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-470%3A+TopologyTestDriver+test+input+and+output+usability+improvements >> > > > > > > > > > > > >>>>>>>> as an example) >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 4) Also note (as already pointed out by >> John), >> > > > that >> > > > > we >> > > > > > > cannot >> > > > > > > > > > > > >> have >> > > > > > > > > > > > >>>> any >> > > > > > > > > > > > >>>>>>>> breaking API changes. Thus, the API should >> be >> > > > > designed >> > > > > > > in a >> > > > > > > > > > > > >> fully >> > > > > > > > > > > > >>>>>>>> backward compatible manner. >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 5) Returning a list of metadata object >> makes >> > it >> > > > hard >> > > > > > > for user >> > > > > > > > > > to >> > > > > > > > > > > > >>>> know if >> > > > > > > > > > > > >>>>>>>> the first object refers to the >> > active(processing), >> > > > > > > > > > > > >>>> active(restoring), or >> > > > > > > > > > > > >>>>>>>> a standby task. IMHO, we should be more >> > explicit. >> > > > > For >> > > > > > > > > > example, a >> > > > > > > > > > > > >>>>>>>> metadata object could have a flag that one >> can >> > > > test >> > > > > via >> > > > > > > > > > > > >>>> `#isActive()`. >> > > > > > > > > > > > >>>>>>>> Or maybe even better, we could keep the >> > current >> > > > API >> > > > > > > as-is and >> > > > > > > > > > > > >> add >> > > > > > > > > > > > >>>>>>>> something like `standbyMetadataForKey()` >> (and >> > > > > similar >> > > > > > > methods >> > > > > > > > > > > > >> for >> > > > > > > > > > > > >>>>>>>> other). Having just a flag `isActive()` is >> a >> > > > little >> > > > > > > subtle and >> > > > > > > > > > > > >>>> having >> > > > > > > > > > > > >>>>>>>> new overloads would make the API much >> clearer >> > > > > (passing >> > > > > > > in a >> > > > > > > > > > > > >>> boolean >> > > > > > > > > > > > >>>> flag >> > > > > > > > > > > > >>>>>>>> does not seem to be a nice API). >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 6) Do you intent to return all standby >> > metadata >> > > > > > > information at >> > > > > > > > > > > > >>> once, >> > > > > > > > > > > > >>>>>>>> similar to `allMetadata()` -- seems to be >> > useful. >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 7) Even if the lag information is >> propagated >> > to >> > > > all >> > > > > > > instances, >> > > > > > > > > > > > >> it >> > > > > > > > > > > > >>>> will >> > > > > > > > > > > > >>>>>>>> happen in an async manner. Hence, I am >> > wondering >> > > > if >> > > > > we >> > > > > > > should >> > > > > > > > > > > > >>>> address >> > > > > > > > > > > > >>>>>>>> this race condition (I think we should). >> The >> > idea >> > > > > would >> > > > > > > be to >> > > > > > > > > > > > >>> check >> > > > > > > > > > > > >>>> if a >> > > > > > > > > > > > >>>>>>>> standby/active(restoring) task is actually >> > still >> > > > > within >> > > > > > > the >> > > > > > > > > > lag >> > > > > > > > > > > > >>>> bounds >> > > > > > > > > > > > >>>>>>>> when a query is executed and we would >> throw an >> > > > > > > exception if >> > > > > > > > > > not. >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 8) The current `KafkaStreams#state()` >> method >> > only >> > > > > > > returns a >> > > > > > > > > > > > >> handle >> > > > > > > > > > > > >>>> to >> > > > > > > > > > > > >>>>>>>> stores of active(processing) tasks. How >> can a >> > user >> > > > > > > actually >> > > > > > > > > > get >> > > > > > > > > > > > >> a >> > > > > > > > > > > > >>>> handle >> > > > > > > > > > > > >>>>>>>> to an store of an active(restoring) or >> standby >> > > > task >> > > > > for >> > > > > > > > > > > > >> querying? >> > > > > > > > > > > > >>>> Seems >> > > > > > > > > > > > >>>>>>>> we should add a new method to get standby >> > handles? >> > > > > > > Changing >> > > > > > > > > > the >> > > > > > > > > > > > >>>>>>>> semantics to existing `state()` would be >> > possible, >> > > > > but >> > > > > > > I think >> > > > > > > > > > > > >>>> adding a >> > > > > > > > > > > > >>>>>>>> new method is preferable? >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 9) How does the user actually specify the >> > > > acceptable >> > > > > > > lag? A >> > > > > > > > > > > > >> global >> > > > > > > > > > > > >>>>>>>> config via StreamsConfig (this would be a >> > public >> > > > API >> > > > > > > change >> > > > > > > > > > that >> > > > > > > > > > > > >>>> needs >> > > > > > > > > > > > >>>>>>>> to be covered in the KIP)? Or on a >> per-store >> > or >> > > > even >> > > > > > > per-query >> > > > > > > > > > > > >>>> basis for >> > > > > > > > > > > > >>>>>>>> more flexibility? We could also have a >> global >> > > > > setting >> > > > > > > that is >> > > > > > > > > > > > >> used >> > > > > > > > > > > > >>>> as >> > > > > > > > > > > > >>>>>>>> default and allow to overwrite it on a >> > per-query >> > > > > basis. >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> 10) Do we need to distinguish between >> > > > > active(restoring) >> > > > > > > and >> > > > > > > > > > > > >>> standby >> > > > > > > > > > > > >>>>>>>> tasks? Or could be treat both as the same? >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> -Matthias >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>>> On 10/21/19 5:40 PM, Vinoth Chandar wrote: >> > > > > > > > > > > > >>>>>>>>>>> I'm wondering, rather than putting >> > "acceptable >> > > > > lag" >> > > > > > > into >> > > > > > > > > > the >> > > > > > > > > > > > >>>>>>>>> configuration at all, or even making it a >> > > > > parameter on >> > > > > > > > > > > > >>>>>> `allMetadataForKey`, >> > > > > > > > > > > > >>>>>>>>> why not just _always_ return all available >> > > > metadata >> > > > > > > > > > (including >> > > > > > > > > > > > >>>>>>>>> active/standby or lag) and let the caller >> > decide >> > > > to >> > > > > > > which >> > > > > > > > > > node >> > > > > > > > > > > > >>> they >> > > > > > > > > > > > >>>>>> want to >> > > > > > > > > > > > >>>>>>>>> route the query? >> > > > > > > > > > > > >>>>>>>>> +1 on exposing lag information via the >> APIs. >> > IMO >> > > > > > > without >> > > > > > > > > > having >> > > > > > > > > > > > >>>>>>>>> continuously updated/fresh lag >> information, >> > its >> > > > > true >> > > > > > > value >> > > > > > > > > > as a >> > > > > > > > > > > > >>>> signal >> > > > > > > > > > > > >>>>>> for >> > > > > > > > > > > > >>>>>>>>> query routing decisions is much limited. >> But >> > we >> > > > can >> > > > > > > design >> > > > > > > > > > the >> > > > > > > > > > > > >>> API >> > > > > > > > > > > > >>>>>> around >> > > > > > > > > > > > >>>>>>>>> this model and iterate? Longer term, we >> > should >> > > > have >> > > > > > > > > > > > >> continuously >> > > > > > > > > > > > >>>>>> shared lag >> > > > > > > > > > > > >>>>>>>>> information. >> > > > > > > > > > > > >>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> more general to refactor it to >> > > > > > > "allMetadataForKey(long >> > > > > > > > > > > > >>>>>>>>> tolerableDataStaleness, ...)", and when >> it's >> > set >> > > > > to 0 >> > > > > > > it >> > > > > > > > > > means >> > > > > > > > > > > > >>>> "active >> > > > > > > > > > > > >>>>>> task >> > > > > > > > > > > > >>>>>>>>> only". >> > > > > > > > > > > > >>>>>>>>> +1 IMO if we plan on having >> > > > > `enableReplicaServing`, it >> > > > > > > makes >> > > > > > > > > > > > >>> sense >> > > > > > > > > > > > >>>> to >> > > > > > > > > > > > >>>>>>>>> generalize based on dataStaleness. This >> seems >> > > > > > > complementary >> > > > > > > > > > to >> > > > > > > > > > > > >>>>>> exposing the >> > > > > > > > > > > > >>>>>>>>> lag information itself. >> > > > > > > > > > > > >>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> This is actually not a public api >> change at >> > > > all, >> > > > > and >> > > > > > > I'm >> > > > > > > > > > > > >>>> planning to >> > > > > > > > > > > > >>>>>>>>> implement it asap as a precursor to the >> rest >> > of >> > > > > KIP-441 >> > > > > > > > > > > > >>>>>>>>> +1 again. Do we have a concrete timeline >> for >> > when >> > > > > this >> > > > > > > change >> > > > > > > > > > > > >>> will >> > > > > > > > > > > > >>>>>> land on >> > > > > > > > > > > > >>>>>>>>> master? I would like to get the >> > implementation >> > > > > wrapped >> > > > > > > up (as >> > > > > > > > > > > > >>> much >> > > > > > > > > > > > >>>> as >> > > > > > > > > > > > >>>>>>>>> possible) by end of the month. :). But I >> > agree >> > > > this >> > > > > > > > > > sequencing >> > > > > > > > > > > > >>>> makes >> > > > > > > > > > > > >>>>>>>>> sense.. >> > > > > > > > > > > > >>>>>>>>> >> > > > > > > > > > > > >>>>>>>>> >> > > > > > > > > > > > >>>>>>>>> On Mon, Oct 21, 2019 at 2:56 PM Guozhang >> > Wang < >> > > > > > > > > > > > >>> wangg...@gmail.com> >> > > > > > > > > > > > >>>>>> wrote: >> > > > > > > > > > > > >>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> Hi Navinder, >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> Thanks for the KIP, I have a high level >> > question >> > > > > > > about the >> > > > > > > > > > > > >>>> proposed >> > > > > > > > > > > > >>>>>> API >> > > > > > > > > > > > >>>>>>>>>> regarding: >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> >> > "StreamsMetadataState::allMetadataForKey(boolean >> > > > > > > > > > > > >>>>>> enableReplicaServing...)" >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> I'm wondering if it's more general to >> > refactor >> > > > it >> > > > > to >> > > > > > > > > > > > >>>>>>>>>> "allMetadataForKey(long >> > tolerableDataStaleness, >> > > > > > > ...)", and >> > > > > > > > > > > > >> when >> > > > > > > > > > > > >>>> it's >> > > > > > > > > > > > >>>>>> set to >> > > > > > > > > > > > >>>>>>>>>> 0 it means "active task only". Behind the >> > scene, >> > > > > we >> > > > > > > can have >> > > > > > > > > > > > >> the >> > > > > > > > > > > > >>>>>> committed >> > > > > > > > > > > > >>>>>>>>>> offsets to encode the stream time as >> well, >> > so >> > > > that >> > > > > > > when >> > > > > > > > > > > > >>> processing >> > > > > > > > > > > > >>>>>> standby >> > > > > > > > > > > > >>>>>>>>>> tasks the stream process knows not long >> the >> > lag >> > > > in >> > > > > > > terms of >> > > > > > > > > > > > >>>> offsets >> > > > > > > > > > > > >>>>>>>>>> comparing to the committed offset >> > (internally we >> > > > > call >> > > > > > > it >> > > > > > > > > > > > >> offset >> > > > > > > > > > > > >>>>>> limit), but >> > > > > > > > > > > > >>>>>>>>>> also the lag in terms of timestamp diff >> > > > comparing >> > > > > the >> > > > > > > > > > > > >> committed >> > > > > > > > > > > > >>>>>> offset. >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> Also encoding the timestamp as part of >> > offset >> > > > have >> > > > > > > other >> > > > > > > > > > > > >>> benefits >> > > > > > > > > > > > >>>> for >> > > > > > > > > > > > >>>>>>>>>> improving Kafka Streams time semantics as >> > well, >> > > > > but >> > > > > > > for >> > > > > > > > > > > > >> KIP-535 >> > > > > > > > > > > > >>>>>> itself I >> > > > > > > > > > > > >>>>>>>>>> think it can help giving users a more >> > intuitive >> > > > > > > interface to >> > > > > > > > > > > > >>>> reason >> > > > > > > > > > > > >>>>>> about. >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> Guozhang >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> On Mon, Oct 21, 2019 at 12:30 PM John >> > Roesler < >> > > > > > > > > > > > >>> j...@confluent.io> >> > > > > > > > > > > > >>>>>> wrote: >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> Hey Navinder, >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> Thanks for the KIP! I've been reading >> over >> > the >> > > > > > > discussion >> > > > > > > > > > > > >> thus >> > > > > > > > > > > > >>>> far, >> > > > > > > > > > > > >>>>>>>>>>> and I have a couple of thoughts to pile >> on >> > as >> > > > > well: >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> It seems confusing to propose the API in >> > terms >> > > > > of the >> > > > > > > > > > current >> > > > > > > > > > > > >>>> system >> > > > > > > > > > > > >>>>>>>>>>> state, but also propose how the API >> would >> > look >> > > > > > > if/when >> > > > > > > > > > > > >> KIP-441 >> > > > > > > > > > > > >>> is >> > > > > > > > > > > > >>>>>>>>>>> implemented. It occurs to me that the >> only >> > part >> > > > > of >> > > > > > > KIP-441 >> > > > > > > > > > > > >> that >> > > > > > > > > > > > >>>> would >> > > > > > > > > > > > >>>>>>>>>>> affect you is the availability of the >> lag >> > > > > > > information in >> > > > > > > > > > the >> > > > > > > > > > > > >>>>>>>>>>> SubscriptionInfo message. This is >> actually >> > not >> > > > a >> > > > > > > public api >> > > > > > > > > > > > >>>> change at >> > > > > > > > > > > > >>>>>>>>>>> all, and I'm planning to implement it >> asap >> > as a >> > > > > > > precursor >> > > > > > > > > > to >> > > > > > > > > > > > >>> the >> > > > > > > > > > > > >>>> rest >> > > > > > > > > > > > >>>>>>>>>>> of KIP-441, so maybe you can just build >> on >> > top >> > > > of >> > > > > > > KIP-441 >> > > > > > > > > > and >> > > > > > > > > > > > >>>> assume >> > > > > > > > > > > > >>>>>>>>>>> the lag information will be available. >> > Then you >> > > > > > > could have >> > > > > > > > > > a >> > > > > > > > > > > > >>> more >> > > > > > > > > > > > >>>>>>>>>>> straightforward proposal (e.g., mention >> > that >> > > > > you'd >> > > > > > > return >> > > > > > > > > > the >> > > > > > > > > > > > >>> lag >> > > > > > > > > > > > >>>>>>>>>>> information in AssignmentInfo as well as >> > in the >> > > > > > > > > > > > >> StreamsMetadata >> > > > > > > > > > > > >>>> in >> > > > > > > > > > > > >>>>>>>>>>> some form, or make use of it in the API >> > > > somehow). >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> I'm partially motivated in that former >> > point >> > > > > because >> > > > > > > it >> > > > > > > > > > seems >> > > > > > > > > > > > >>>> like >> > > > > > > > > > > > >>>>>>>>>>> understanding how callers would bound >> the >> > > > > staleness >> > > > > > > for >> > > > > > > > > > their >> > > > > > > > > > > > >>> use >> > > > > > > > > > > > >>>>>> case >> > > > > > > > > > > > >>>>>>>>>>> is _the_ key point for this KIP. FWIW, I >> > think >> > > > > that >> > > > > > > adding >> > > > > > > > > > a >> > > > > > > > > > > > >>> new >> > > > > > > > > > > > >>>>>>>>>>> method to StreamsMetadataState and >> > deprecating >> > > > > the >> > > > > > > existing >> > > > > > > > > > > > >>>> method is >> > > > > > > > > > > > >>>>>>>>>>> the best way to go; we just can't change >> > the >> > > > > return >> > > > > > > types >> > > > > > > > > > of >> > > > > > > > > > > > >>> any >> > > > > > > > > > > > >>>>>>>>>>> existing methods. >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> I'm wondering, rather than putting >> > "acceptable >> > > > > lag" >> > > > > > > into >> > > > > > > > > > the >> > > > > > > > > > > > >>>>>>>>>>> configuration at all, or even making it >> a >> > > > > parameter >> > > > > > > on >> > > > > > > > > > > > >>>>>>>>>>> `allMetadataForKey`, why not just >> _always_ >> > > > > return all >> > > > > > > > > > > > >> available >> > > > > > > > > > > > >>>>>>>>>>> metadata (including active/standby or >> lag) >> > and >> > > > > let >> > > > > > > the >> > > > > > > > > > caller >> > > > > > > > > > > > >>>> decide >> > > > > > > > > > > > >>>>>>>>>>> to which node they want to route the >> query? >> > > > This >> > > > > > > method >> > > > > > > > > > isn't >> > > > > > > > > > > > >>>> making >> > > > > > > > > > > > >>>>>>>>>>> any queries itself; it's merely telling >> you >> > > > where >> > > > > > > the local >> > > > > > > > > > > > >>>> Streams >> > > > > > > > > > > > >>>>>>>>>>> instance _thinks_ the key in question is >> > > > located. >> > > > > > > Just >> > > > > > > > > > > > >>> returning >> > > > > > > > > > > > >>>> all >> > > > > > > > > > > > >>>>>>>>>>> available information lets the caller >> > implement >> > > > > any >> > > > > > > > > > semantics >> > > > > > > > > > > > >>>> they >> > > > > > > > > > > > >>>>>>>>>>> desire around querying only active >> stores, >> > or >> > > > > > > standbys, or >> > > > > > > > > > > > >>>> recovering >> > > > > > > > > > > > >>>>>>>>>>> stores, or whatever. >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> One fly in the ointment, which you may >> > wish to >> > > > > > > consider if >> > > > > > > > > > > > >>>> proposing >> > > > > > > > > > > > >>>>>>>>>>> to use lag information, is that the >> cluster >> > > > would >> > > > > > > only >> > > > > > > > > > become >> > > > > > > > > > > > >>>> aware >> > > > > > > > > > > > >>>>>> of >> > > > > > > > > > > > >>>>>>>>>>> new lag information during rebalances. >> > Even in >> > > > > the >> > > > > > > full >> > > > > > > > > > > > >>>> expression of >> > > > > > > > > > > > >>>>>>>>>>> KIP-441, this information would stop >> being >> > > > > > > propagated when >> > > > > > > > > > > > >> the >> > > > > > > > > > > > >>>>>> cluster >> > > > > > > > > > > > >>>>>>>>>>> achieves a balanced task distribution. >> > There >> > > > was >> > > > > a >> > > > > > > > > > follow-on >> > > > > > > > > > > > >>>> idea I >> > > > > > > > > > > > >>>>>>>>>>> POCed to continuously share lag >> > information in >> > > > > the >> > > > > > > > > > heartbeat >> > > > > > > > > > > > >>>>>> protocol, >> > > > > > > > > > > > >>>>>>>>>>> which you might be interested in, if you >> > want >> > > > to >> > > > > > > make sure >> > > > > > > > > > > > >> that >> > > > > > > > > > > > >>>> nodes >> > > > > > > > > > > > >>>>>>>>>>> are basically _always_ aware of each >> > others' >> > > > lag >> > > > > on >> > > > > > > > > > different >> > > > > > > > > > > > >>>>>>>>>>> partitions: >> > > > > > > https://github.com/apache/kafka/pull/7096 >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> Thanks again! >> > > > > > > > > > > > >>>>>>>>>>> -John >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> On Sat, Oct 19, 2019 at 6:06 AM Navinder >> > Brar >> > > > > > > > > > > > >>>>>>>>>>> <navinder_b...@yahoo.com.invalid> >> wrote: >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> Thanks, Vinoth. Looks like we are on >> the >> > same >> > > > > page. >> > > > > > > I will >> > > > > > > > > > > > >> add >> > > > > > > > > > > > >>>> some >> > > > > > > > > > > > >>>>>> of >> > > > > > > > > > > > >>>>>>>>>>> these explanations to the KIP as well. >> Have >> > > > > assigned >> > > > > > > the >> > > > > > > > > > > > >>>> KAFKA-6144 >> > > > > > > > > > > > >>>>>> to >> > > > > > > > > > > > >>>>>>>>>>> myself and KAFKA-8994 is closed(by >> you). As >> > > > > > > suggested, we >> > > > > > > > > > > > >> will >> > > > > > > > > > > > >>>>>> replace >> > > > > > > > > > > > >>>>>>>>>>> "replica" with "standby". >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> In the new API, >> > > > > > > > > > > > >>> "StreamsMetadataState::allMetadataForKey(boolean >> > > > > > > > > > > > >>>>>>>>>>> enableReplicaServing, String storeName, >> K >> > key, >> > > > > > > > > > Serializer<K> >> > > > > > > > > > > > >>>>>>>>>>> keySerializer)" Do we really need a per >> key >> > > > > > > configuration? >> > > > > > > > > > > > >> or a >> > > > > > > > > > > > >>>> new >> > > > > > > > > > > > >>>>>>>>>>> StreamsConfig is good enough?>> Coming >> from >> > > > > > > experience, >> > > > > > > > > > when >> > > > > > > > > > > > >>>> teams >> > > > > > > > > > > > >>>>>> are >> > > > > > > > > > > > >>>>>>>>>>> building a platform with Kafka Streams >> and >> > > > these >> > > > > > > API's >> > > > > > > > > > serve >> > > > > > > > > > > > >>>> data to >> > > > > > > > > > > > >>>>>>>>>>> multiple teams, we can't have a >> generalized >> > > > > config >> > > > > > > that >> > > > > > > > > > says >> > > > > > > > > > > > >>> as a >> > > > > > > > > > > > >>>>>>>>>> platform >> > > > > > > > > > > > >>>>>>>>>>> we will support stale reads or not. It >> > should >> > > > be >> > > > > the >> > > > > > > choice >> > > > > > > > > > > > >> of >> > > > > > > > > > > > >>>>>> someone >> > > > > > > > > > > > >>>>>>>>>> who >> > > > > > > > > > > > >>>>>>>>>>> is calling the API's to choose whether >> > they are >> > > > > ok >> > > > > > > with >> > > > > > > > > > stale >> > > > > > > > > > > > >>>> reads >> > > > > > > > > > > > >>>>>> or >> > > > > > > > > > > > >>>>>>>>>> not. >> > > > > > > > > > > > >>>>>>>>>>> Makes sense? >> > > > > > > > > > > > >>>>>>>>>>>> On Thursday, 17 October, 2019, >> > 11:56:02 pm >> > > > > IST, >> > > > > > > Vinoth >> > > > > > > > > > > > >>>> Chandar < >> > > > > > > > > > > > >>>>>>>>>>> vchan...@confluent.io> wrote: >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> Looks like we are covering ground :) >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>> Only if it is within a permissible >> > > > range(say >> > > > > > > 10000) we >> > > > > > > > > > > > >> will >> > > > > > > > > > > > >>>> serve >> > > > > > > > > > > > >>>>>>>>>> from >> > > > > > > > > > > > >>>>>>>>>>>> Restoring state of active. >> > > > > > > > > > > > >>>>>>>>>>>> +1 on having a knob like this.. My >> > reasoning >> > > > is >> > > > > as >> > > > > > > > > > follows. >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> Looking at the Streams state as a >> > read-only >> > > > > > > distributed kv >> > > > > > > > > > > > >>>> store. >> > > > > > > > > > > > >>>>>> With >> > > > > > > > > > > > >>>>>>>>>>>> num_standby = f , we should be able to >> > > > tolerate >> > > > > f >> > > > > > > failures >> > > > > > > > > > > > >> and >> > > > > > > > > > > > >>>> if >> > > > > > > > > > > > >>>>>> there >> > > > > > > > > > > > >>>>>>>>>>> is >> > > > > > > > > > > > >>>>>>>>>>>> a f+1' failure, the system should be >> > > > > unavailable. >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> A) So with num_standby=0, the system >> > should be >> > > > > > > unavailable >> > > > > > > > > > > > >>> even >> > > > > > > > > > > > >>>> if >> > > > > > > > > > > > >>>>>>>>>> there >> > > > > > > > > > > > >>>>>>>>>>> is >> > > > > > > > > > > > >>>>>>>>>>>> 1 failure and thats my argument for not >> > > > allowing >> > > > > > > querying >> > > > > > > > > > in >> > > > > > > > > > > > >>>>>>>>>> restoration >> > > > > > > > > > > > >>>>>>>>>>>> state, esp in this case it will be a >> total >> > > > > rebuild >> > > > > > > of the >> > > > > > > > > > > > >>> state >> > > > > > > > > > > > >>>>>> (which >> > > > > > > > > > > > >>>>>>>>>>> IMO >> > > > > > > > > > > > >>>>>>>>>>>> cannot be considered a normal fault >> free >> > > > > operational >> > > > > > > > > > state). >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> B) Even there are standby's, say >> > > > num_standby=2, >> > > > > if >> > > > > > > the >> > > > > > > > > > user >> > > > > > > > > > > > >>>> decides >> > > > > > > > > > > > >>>>>> to >> > > > > > > > > > > > >>>>>>>>>>> shut >> > > > > > > > > > > > >>>>>>>>>>>> down all 3 instances, then only outcome >> > should >> > > > > be >> > > > > > > > > > > > >>> unavailability >> > > > > > > > > > > > >>>>>> until >> > > > > > > > > > > > >>>>>>>>>>> all >> > > > > > > > > > > > >>>>>>>>>>>> of them come back or state is rebuilt >> on >> > other >> > > > > > > nodes in >> > > > > > > > > > the >> > > > > > > > > > > > >>>>>> cluster. In >> > > > > > > > > > > > >>>>>>>>>>>> normal operations, f <= 2 and when a >> > failure >> > > > > does >> > > > > > > happen >> > > > > > > > > > we >> > > > > > > > > > > > >>> can >> > > > > > > > > > > > >>>> then >> > > > > > > > > > > > >>>>>>>>>>> either >> > > > > > > > > > > > >>>>>>>>>>>> choose to be C over A and fail IQs >> until >> > > > > > > replication is >> > > > > > > > > > > > >> fully >> > > > > > > > > > > > >>>>>> caught up >> > > > > > > > > > > > >>>>>>>>>>> or >> > > > > > > > > > > > >>>>>>>>>>>> choose A over C by serving in restoring >> > state >> > > > as >> > > > > > > long as >> > > > > > > > > > lag >> > > > > > > > > > > > >>> is >> > > > > > > > > > > > >>>>>>>>>> minimal. >> > > > > > > > > > > > >>>>>>>>>>> If >> > > > > > > > > > > > >>>>>>>>>>>> even with f=1 say, all the standbys are >> > > > lagging >> > > > > a >> > > > > > > lot due >> > > > > > > > > > to >> > > > > > > > > > > > >>>> some >> > > > > > > > > > > > >>>>>>>>>> issue, >> > > > > > > > > > > > >>>>>>>>>>>> then that should be considered a >> failure >> > since >> > > > > that >> > > > > > > is >> > > > > > > > > > > > >>> different >> > > > > > > > > > > > >>>>>> from >> > > > > > > > > > > > >>>>>>>>>>>> normal/expected operational mode. >> Serving >> > > > reads >> > > > > with >> > > > > > > > > > > > >> unbounded >> > > > > > > > > > > > >>>>>>>>>>> replication >> > > > > > > > > > > > >>>>>>>>>>>> lag and calling it "available" may not >> be >> > very >> > > > > > > usable or >> > > > > > > > > > > > >> even >> > > > > > > > > > > > >>>>>> desirable >> > > > > > > > > > > > >>>>>>>>>>> :) >> > > > > > > > > > > > >>>>>>>>>>>> IMHO, since it gives the user no way to >> > reason >> > > > > > > about the >> > > > > > > > > > app >> > > > > > > > > > > > >>>> that is >> > > > > > > > > > > > >>>>>>>>>>> going >> > > > > > > > > > > > >>>>>>>>>>>> to query this store. >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> So there is definitely a need to >> > distinguish >> > > > > > > between : >> > > > > > > > > > > > >>>> Replication >> > > > > > > > > > > > >>>>>>>>>>> catchup >> > > > > > > > > > > > >>>>>>>>>>>> while being in fault free state vs >> > Restoration >> > > > > of >> > > > > > > state >> > > > > > > > > > when >> > > > > > > > > > > > >>> we >> > > > > > > > > > > > >>>> lose >> > > > > > > > > > > > >>>>>>>>>> more >> > > > > > > > > > > > >>>>>>>>>>>> than f standbys. This knob is a great >> > starting >> > > > > point >> > > > > > > > > > towards >> > > > > > > > > > > > >>>> this. >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> If you agree with some of the >> explanation >> > > > above, >> > > > > > > please >> > > > > > > > > > feel >> > > > > > > > > > > > >>>> free to >> > > > > > > > > > > > >>>>>>>>>>>> include it in the KIP as well since >> this >> > is >> > > > > sort of >> > > > > > > our >> > > > > > > > > > > > >> design >> > > > > > > > > > > > >>>>>>>>>> principle >> > > > > > > > > > > > >>>>>>>>>>>> here.. >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> Small nits : >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> - let's standardize on "standby" >> instead >> > of >> > > > > > > "replica", KIP >> > > > > > > > > > > > >> or >> > > > > > > > > > > > >>>>>> code, to >> > > > > > > > > > > > >>>>>>>>>>> be >> > > > > > > > > > > > >>>>>>>>>>>> consistent with rest of Streams >> code/docs? >> > > > > > > > > > > > >>>>>>>>>>>> - Can we merge KAFKA-8994 into >> KAFKA-6144 >> > now >> > > > > and >> > > > > > > close >> > > > > > > > > > the >> > > > > > > > > > > > >>>> former? >> > > > > > > > > > > > >>>>>>>>>>>> Eventually need to consolidate >> KAFKA-6555 >> > as >> > > > > well >> > > > > > > > > > > > >>>>>>>>>>>> - In the new API, >> > > > > > > > > > > > >>>> >> "StreamsMetadataState::allMetadataForKey(boolean >> > > > > > > > > > > > >>>>>>>>>>>> enableReplicaServing, String >> storeName, K >> > key, >> > > > > > > > > > Serializer<K> >> > > > > > > > > > > > >>>>>>>>>>> keySerializer)" Do >> > > > > > > > > > > > >>>>>>>>>>>> we really need a per key configuration? >> > or a >> > > > new >> > > > > > > > > > > > >> StreamsConfig >> > > > > > > > > > > > >>>> is >> > > > > > > > > > > > >>>>>> good >> > > > > > > > > > > > >>>>>>>>>>>> enough? >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>> On Wed, Oct 16, 2019 at 8:31 PM >> Navinder >> > Brar >> > > > > > > > > > > > >>>>>>>>>>>> <navinder_b...@yahoo.com.invalid> >> wrote: >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> @Vinoth, I have incorporated a few of >> the >> > > > > > > discussions we >> > > > > > > > > > > > >> have >> > > > > > > > > > > > >>>> had >> > > > > > > > > > > > >>>>>> in >> > > > > > > > > > > > >>>>>>>>>>> the >> > > > > > > > > > > > >>>>>>>>>>>>> KIP. >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> In the current code, t0 and t1 serve >> > queries >> > > > > from >> > > > > > > > > > > > >>>> Active(Running) >> > > > > > > > > > > > >>>>>>>>>>>>> partition. For case t2, we are >> planning >> > to >> > > > > return >> > > > > > > > > > > > >>>>>>>>>> List<StreamsMetadata> >> > > > > > > > > > > > >>>>>>>>>>>>> such that it returns >> <StreamsMetadata(A), >> > > > > > > > > > > > >> StreamsMetadata(B)> >> > > > > > > > > > > > >>>> so >> > > > > > > > > > > > >>>>>> that >> > > > > > > > > > > > >>>>>>>>>>> if IQ >> > > > > > > > > > > > >>>>>>>>>>>>> fails on A, the replica on B can serve >> > the >> > > > > data by >> > > > > > > > > > enabling >> > > > > > > > > > > > >>>> serving >> > > > > > > > > > > > >>>>>>>>>>> from >> > > > > > > > > > > > >>>>>>>>>>>>> replicas. This still does not solve >> case >> > t3 >> > > > > and t4 >> > > > > > > since >> > > > > > > > > > B >> > > > > > > > > > > > >>> has >> > > > > > > > > > > > >>>> been >> > > > > > > > > > > > >>>>>>>>>>>>> promoted to active but it is in >> Restoring >> > > > > state to >> > > > > > > > > > catchup >> > > > > > > > > > > > >>>> till A’s >> > > > > > > > > > > > >>>>>>>>>>> last >> > > > > > > > > > > > >>>>>>>>>>>>> committed position as we don’t serve >> from >> > > > > > > Restoring state >> > > > > > > > > > > > >> in >> > > > > > > > > > > > >>>> Active >> > > > > > > > > > > > >>>>>>>>>>> and new >> > > > > > > > > > > > >>>>>>>>>>>>> Replica on R is building itself from >> > scratch. >> > > > > Both >> > > > > > > these >> > > > > > > > > > > > >>> cases >> > > > > > > > > > > > >>>> can >> > > > > > > > > > > > >>>>>> be >> > > > > > > > > > > > >>>>>>>>>>>>> solved if we start serving from >> Restoring >> > > > > state of >> > > > > > > active >> > > > > > > > > > > > >> as >> > > > > > > > > > > > >>>> well >> > > > > > > > > > > > >>>>>>>>>>> since it >> > > > > > > > > > > > >>>>>>>>>>>>> is almost equivalent to previous >> Active. >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> There could be a case where all >> replicas >> > of a >> > > > > > > partition >> > > > > > > > > > > > >>> become >> > > > > > > > > > > > >>>>>>>>>>> unavailable >> > > > > > > > > > > > >>>>>>>>>>>>> and active and all replicas of that >> > partition >> > > > > are >> > > > > > > > > > building >> > > > > > > > > > > > >>>>>> themselves >> > > > > > > > > > > > >>>>>>>>>>> from >> > > > > > > > > > > > >>>>>>>>>>>>> scratch, in this case, the state in >> > Active is >> > > > > far >> > > > > > > behind >> > > > > > > > > > > > >> even >> > > > > > > > > > > > >>>>>> though >> > > > > > > > > > > > >>>>>>>>>>> it is >> > > > > > > > > > > > >>>>>>>>>>>>> in Restoring state. To cater to such >> > cases >> > > > > that we >> > > > > > > don’t >> > > > > > > > > > > > >>> serve >> > > > > > > > > > > > >>>> from >> > > > > > > > > > > > >>>>>>>>>>> this >> > > > > > > > > > > > >>>>>>>>>>>>> state we can either add another state >> > before >> > > > > > > Restoring or >> > > > > > > > > > > > >>>> check the >> > > > > > > > > > > > >>>>>>>>>>>>> difference between last committed >> offset >> > and >> > > > > > > current >> > > > > > > > > > > > >>> position. >> > > > > > > > > > > > >>>> Only >> > > > > > > > > > > > >>>>>>>>>> if >> > > > > > > > > > > > >>>>>>>>>>> it >> > > > > > > > > > > > >>>>>>>>>>>>> is within a permissible range (say >> > 10000) we >> > > > > will >> > > > > > > serve >> > > > > > > > > > > > >> from >> > > > > > > > > > > > >>>>>>>>>> Restoring >> > > > > > > > > > > > >>>>>>>>>>> the >> > > > > > > > > > > > >>>>>>>>>>>>> state of Active. >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> On Wednesday, 16 October, 2019, >> > 10:01:35 >> > > > pm >> > > > > IST, >> > > > > > > > > > Vinoth >> > > > > > > > > > > > >>>> Chandar >> > > > > > > > > > > > >>>>>> < >> > > > > > > > > > > > >>>>>>>>>>>>> vchan...@confluent.io> wrote: >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> Thanks for the updates on the KIP, >> > Navinder! >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> Few comments >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> - AssignmentInfo is not public API?. >> But >> > we >> > > > > will >> > > > > > > change >> > > > > > > > > > it >> > > > > > > > > > > > >>> and >> > > > > > > > > > > > >>>> thus >> > > > > > > > > > > > >>>>>>>>>>> need to >> > > > > > > > > > > > >>>>>>>>>>>>> increment the version and test for >> > > > > version_probing >> > > > > > > etc. >> > > > > > > > > > > > >> Good >> > > > > > > > > > > > >>> to >> > > > > > > > > > > > >>>>>>>>>>> separate >> > > > > > > > > > > > >>>>>>>>>>>>> that from StreamsMetadata changes >> (which >> > is >> > > > > public >> > > > > > > API) >> > > > > > > > > > > > >>>>>>>>>>>>> - From what I see, there is going to >> be >> > > > choice >> > > > > > > between >> > > > > > > > > > the >> > > > > > > > > > > > >>>>>> following >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> A) introducing a new >> > > > > > > *KafkaStreams::allMetadataForKey() >> > > > > > > > > > > > >> *API >> > > > > > > > > > > > >>>> that >> > > > > > > > > > > > >>>>>>>>>>>>> potentially returns >> List<StreamsMetadata> >> > > > > ordered >> > > > > > > from >> > > > > > > > > > most >> > > > > > > > > > > > >>>> upto >> > > > > > > > > > > > >>>>>> date >> > > > > > > > > > > > >>>>>>>>>>> to >> > > > > > > > > > > > >>>>>>>>>>>>> least upto date replicas. Today we >> cannot >> > > > fully >> > > > > > > implement >> > > > > > > > > > > > >>> this >> > > > > > > > > > > > >>>>>>>>>>> ordering, >> > > > > > > > > > > > >>>>>>>>>>>>> since all we know is which hosts are >> > active >> > > > and >> > > > > > > which are >> > > > > > > > > > > > >>>> standbys. >> > > > > > > > > > > > >>>>>>>>>>>>> However, this aligns well with the >> > future. >> > > > > KIP-441 >> > > > > > > adds >> > > > > > > > > > the >> > > > > > > > > > > > >>> lag >> > > > > > > > > > > > >>>>>>>>>>> information >> > > > > > > > > > > > >>>>>>>>>>>>> to the rebalancing protocol. We could >> > also >> > > > sort >> > > > > > > replicas >> > > > > > > > > > > > >>> based >> > > > > > > > > > > > >>>> on >> > > > > > > > > > > > >>>>>> the >> > > > > > > > > > > > >>>>>>>>>>>>> report lags eventually. This is fully >> > > > backwards >> > > > > > > > > > compatible >> > > > > > > > > > > > >>> with >> > > > > > > > > > > > >>>>>>>>>>> existing >> > > > > > > > > > > > >>>>>>>>>>>>> clients. Only drawback I see is the >> > naming of >> > > > > the >> > > > > > > > > > existing >> > > > > > > > > > > > >>>> method >> > > > > > > > > > > > >>>>>>>>>>>>> KafkaStreams::metadataForKey, not >> > conveying >> > > > the >> > > > > > > > > > distinction >> > > > > > > > > > > > >>>> that it >> > > > > > > > > > > > >>>>>>>>>>> simply >> > > > > > > > > > > > >>>>>>>>>>>>> returns the active replica i.e >> > > > > > > allMetadataForKey.get(0). >> > > > > > > > > > > > >>>>>>>>>>>>> B) Change >> > KafkaStreams::metadataForKey() to >> > > > > > > return a >> > > > > > > > > > List. >> > > > > > > > > > > > >>>> Its a >> > > > > > > > > > > > >>>>>>>>>>> breaking >> > > > > > > > > > > > >>>>>>>>>>>>> change. >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> I prefer A, since none of the >> > > > > semantics/behavior >> > > > > > > changes >> > > > > > > > > > > > >> for >> > > > > > > > > > > > >>>>>> existing >> > > > > > > > > > > > >>>>>>>>>>>>> users. Love to hear more thoughts. >> Can we >> > > > also >> > > > > > > work this >> > > > > > > > > > > > >> into >> > > > > > > > > > > > >>>> the >> > > > > > > > > > > > >>>>>>>>>> KIP? >> > > > > > > > > > > > >>>>>>>>>>>>> I already implemented A to unblock >> > myself for >> > > > > now. >> > > > > > > Seems >> > > > > > > > > > > > >>>> feasible >> > > > > > > > > > > > >>>>>> to >> > > > > > > > > > > > >>>>>>>>>>> do. >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> On Tue, Oct 15, 2019 at 12:21 PM >> Vinoth >> > > > > Chandar < >> > > > > > > > > > > > >>>>>>>>>> vchan...@confluent.io >> > > > > > > > > > > > >>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> wrote: >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> I get your point. But suppose there >> > is a >> > > > > > > replica which >> > > > > > > > > > > > >> has >> > > > > > > > > > > > >>>> just >> > > > > > > > > > > > >>>>>>>>>>> become >> > > > > > > > > > > > >>>>>>>>>>>>>> active, so in that case replica will >> > still >> > > > be >> > > > > > > building >> > > > > > > > > > > > >>> itself >> > > > > > > > > > > > >>>> from >> > > > > > > > > > > > >>>>>>>>>>>>> scratch >> > > > > > > > > > > > >>>>>>>>>>>>>> and this active will go to restoring >> > state >> > > > > till it >> > > > > > > > > > catches >> > > > > > > > > > > > >>> up >> > > > > > > > > > > > >>>> with >> > > > > > > > > > > > >>>>>>>>>>>>> previous >> > > > > > > > > > > > >>>>>>>>>>>>>> active, wouldn't serving from a >> > restoring >> > > > > active >> > > > > > > make >> > > > > > > > > > more >> > > > > > > > > > > > >>>> sense >> > > > > > > > > > > > >>>>>>>>>>> than a >> > > > > > > > > > > > >>>>>>>>>>>>>> replica in such case. >> > > > > > > > > > > > >>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>> KIP-441 will change this behavior >> such >> > that >> > > > > > > promotion to >> > > > > > > > > > > > >>>> active >> > > > > > > > > > > > >>>>>>>>>>> happens >> > > > > > > > > > > > >>>>>>>>>>>>>> based on how caught up a replica is. >> So, >> > > > once >> > > > > we >> > > > > > > have >> > > > > > > > > > that >> > > > > > > > > > > > >>>> (work >> > > > > > > > > > > > >>>>>>>>>>> underway >> > > > > > > > > > > > >>>>>>>>>>>>>> already for 2.5 IIUC) and user sets >> > > > > > > > > > num.standby.replicas > >> > > > > > > > > > > > >>> 0, >> > > > > > > > > > > > >>>> then >> > > > > > > > > > > > >>>>>>>>>>> the >> > > > > > > > > > > > >>>>>>>>>>>>>> staleness window should not be that >> > long as >> > > > > you >> > > > > > > > > > describe. >> > > > > > > > > > > > >>> IMO >> > > > > > > > > > > > >>>> if >> > > > > > > > > > > > >>>>>>>>>> user >> > > > > > > > > > > > >>>>>>>>>>>>> wants >> > > > > > > > > > > > >>>>>>>>>>>>>> availability for state, then should >> > > > configure >> > > > > > > > > > > > >>>> num.standby.replicas >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> 0. >> > > > > > > > > > > > >>>>>>>>>>>>> If >> > > > > > > > > > > > >>>>>>>>>>>>>> not, then on a node loss, few >> partitions >> > > > > would be >> > > > > > > > > > > > >>> unavailable >> > > > > > > > > > > > >>>> for >> > > > > > > > > > > > >>>>>> a >> > > > > > > > > > > > >>>>>>>>>>> while >> > > > > > > > > > > > >>>>>>>>>>>>>> (there are other ways to bring this >> > window >> > > > > down, >> > > > > > > which I >> > > > > > > > > > > > >>> won't >> > > > > > > > > > > > >>>>>>>>>> bring >> > > > > > > > > > > > >>>>>>>>>>> in >> > > > > > > > > > > > >>>>>>>>>>>>>> here). We could argue for querying a >> > > > restoring >> > > > > > > active >> > > > > > > > > > > > >> (say a >> > > > > > > > > > > > >>>> new >> > > > > > > > > > > > >>>>>>>>>> node >> > > > > > > > > > > > >>>>>>>>>>>>> added >> > > > > > > > > > > > >>>>>>>>>>>>>> to replace a faulty old node) based >> on >> > AP vs >> > > > > CP >> > > > > > > > > > > > >> principles. >> > > > > > > > > > > > >>>> But >> > > > > > > > > > > > >>>>>> not >> > > > > > > > > > > > >>>>>>>>>>> sure >> > > > > > > > > > > > >>>>>>>>>>>>>> reading really really old values for >> the >> > > > sake >> > > > > of >> > > > > > > > > > > > >>> availability >> > > > > > > > > > > > >>>> is >> > > > > > > > > > > > >>>>>>>>>>> useful. >> > > > > > > > > > > > >>>>>>>>>>>>> No >> > > > > > > > > > > > >>>>>>>>>>>>>> AP data system would be inconsistent >> for >> > > > such >> > > > > a >> > > > > > > long >> > > > > > > > > > time >> > > > > > > > > > > > >> in >> > > > > > > > > > > > >>>>>>>>>>> practice. >> > > > > > > > > > > > >>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>> So, I still feel just limiting this >> to >> > > > standby >> > > > > > > reads >> > > > > > > > > > > > >>> provides >> > > > > > > > > > > > >>>> best >> > > > > > > > > > > > >>>>>>>>>>>>>> semantics. >> > > > > > > > > > > > >>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>> Just my 2c. Would love to see what >> > others >> > > > > think >> > > > > > > as well. >> > > > > > > > > > > > >>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>> On Tue, Oct 15, 2019 at 5:34 AM >> Navinder >> > > > Brar >> > > > > > > > > > > > >>>>>>>>>>>>>> <navinder_b...@yahoo.com.invalid> >> > wrote: >> > > > > > > > > > > > >>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> Hi Vinoth, >> > > > > > > > > > > > >>>>>>>>>>>>>>> Thanks for the feedback. >> > > > > > > > > > > > >>>>>>>>>>>>>>> Can we link the JIRA, discussion >> > thread >> > > > > also to >> > > > > > > the >> > > > > > > > > > > > >> KIP.>> >> > > > > > > > > > > > >>>>>> Added. >> > > > > > > > > > > > >>>>>>>>>>>>>>> Based on the discussion on >> KAFKA-6144, >> > I >> > > > was >> > > > > > > under the >> > > > > > > > > > > > >>>> impression >> > > > > > > > > > > > >>>>>>>>>>> that >> > > > > > > > > > > > >>>>>>>>>>>>>>> this KIP is also going to cover >> > exposing of >> > > > > the >> > > > > > > standby >> > > > > > > > > > > > >>>>>>>>>> information >> > > > > > > > > > > > >>>>>>>>>>> in >> > > > > > > > > > > > >>>>>>>>>>>>>>> StreamsMetadata and thus subsume >> > > > KAFKA-8994 . >> > > > > > > That >> > > > > > > > > > would >> > > > > > > > > > > > >>>> require >> > > > > > > > > > > > >>>>>> a >> > > > > > > > > > > > >>>>>>>>>>>>> public >> > > > > > > > > > > > >>>>>>>>>>>>>>> API change?>> Sure, I can add >> changes >> > for >> > > > > 8994 >> > > > > > > in this >> > > > > > > > > > > > >> KIP >> > > > > > > > > > > > >>>> and >> > > > > > > > > > > > >>>>>>>>>> link >> > > > > > > > > > > > >>>>>>>>>>>>>>> KAFKA-6144 to KAFKA-8994 as well. >> > > > > > > > > > > > >>>>>>>>>>>>>>> KIP seems to be focussing on >> > restoration >> > > > > when a >> > > > > > > new >> > > > > > > > > > node >> > > > > > > > > > > > >>> is >> > > > > > > > > > > > >>>>>>>>>> added. >> > > > > > > > > > > > >>>>>>>>>>>>>>> KIP-441 is underway and has some >> major >> > > > > changes >> > > > > > > proposed >> > > > > > > > > > > > >> for >> > > > > > > > > > > > >>>> this. >> > > > > > > > > > > > >>>>>>>>>> It >> > > > > > > > > > > > >>>>>>>>>>>>> would >> > > > > > > > > > > > >>>>>>>>>>>>>>> be good to clarify dependencies if >> any. >> > > > > Without >> > > > > > > > > > KIP-441, >> > > > > > > > > > > > >> I >> > > > > > > > > > > > >>>> am not >> > > > > > > > > > > > >>>>>>>>>>> very >> > > > > > > > > > > > >>>>>>>>>>>>> sure >> > > > > > > > > > > > >>>>>>>>>>>>>>> if we should allow reads from nodes >> in >> > > > > RESTORING >> > > > > > > state, >> > > > > > > > > > > > >>> which >> > > > > > > > > > > > >>>>>>>>>> could >> > > > > > > > > > > > >>>>>>>>>>>>> amount >> > > > > > > > > > > > >>>>>>>>>>>>>>> to many minutes/few hours of stale >> > reads? >> > > > > This >> > > > > > > is >> > > > > > > > > > > > >>> different >> > > > > > > > > > > > >>>> from >> > > > > > > > > > > > >>>>>>>>>>>>> allowing >> > > > > > > > > > > > >>>>>>>>>>>>>>> querying standby replicas, which >> could >> > be >> > > > > mostly >> > > > > > > caught >> > > > > > > > > > > > >> up >> > > > > > > > > > > > >>>> and >> > > > > > > > > > > > >>>>>> the >> > > > > > > > > > > > >>>>>>>>>>>>>>> staleness window could be much >> > > > > > > smaller/tolerable. (once >> > > > > > > > > > > > >>>> again the >> > > > > > > > > > > > >>>>>>>>>>> focus >> > > > > > > > > > > > >>>>>>>>>>>>> on >> > > > > > > > > > > > >>>>>>>>>>>>>>> KAFKA-8994).>> I get your point. But >> > > > suppose >> > > > > > > there is a >> > > > > > > > > > > > >>>> replica >> > > > > > > > > > > > >>>>>>>>>>> which >> > > > > > > > > > > > >>>>>>>>>>>>> has >> > > > > > > > > > > > >>>>>>>>>>>>>>> just become active, so in that case >> > replica >> > > > > will >> > > > > > > still >> > > > > > > > > > be >> > > > > > > > > > > > >>>>>> building >> > > > > > > > > > > > >>>>>>>>>>>>> itself >> > > > > > > > > > > > >>>>>>>>>>>>>>> from scratch and this active will >> go to >> > > > > > > restoring state >> > > > > > > > > > > > >>> till >> > > > > > > > > > > > >>>> it >> > > > > > > > > > > > >>>>>>>>>>> catches >> > > > > > > > > > > > >>>>>>>>>>>>> up >> > > > > > > > > > > > >>>>>>>>>>>>>>> with previous active, wouldn't >> serving >> > > > from a >> > > > > > > restoring >> > > > > > > > > > > > >>>> active >> > > > > > > > > > > > >>>>>>>>>> make >> > > > > > > > > > > > >>>>>>>>>>> more >> > > > > > > > > > > > >>>>>>>>>>>>>>> sense than a replica in such case. >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> Finally, we may need to introduce a >> > > > > > > configuration to >> > > > > > > > > > > > >>> control >> > > > > > > > > > > > >>>>>> this. >> > > > > > > > > > > > >>>>>>>>>>> Some >> > > > > > > > > > > > >>>>>>>>>>>>>>> users may prefer errors to stale >> data. >> > Can >> > > > we >> > > > > > > also add >> > > > > > > > > > it >> > > > > > > > > > > > >>> to >> > > > > > > > > > > > >>>> the >> > > > > > > > > > > > >>>>>>>>>>> KIP?>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> Will add this. >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> Regards, >> > > > > > > > > > > > >>>>>>>>>>>>>>> Navinder >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> On2019/10/14 16:56:49, Vinoth >> Chandar < >> > > > > > > > > > v...@confluent.io >> > > > > > > > > > > > >>>>> wrote: >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> Hi Navinder,> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> Thanks for sharing the KIP! Few >> > thoughts> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> - Can we link the JIRA, discussion >> > thread >> > > > > also >> > > > > > > to the >> > > > > > > > > > > > >> KIP> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> - Based on the discussion on >> > KAFKA-6144, I >> > > > > was >> > > > > > > under >> > > > > > > > > > the >> > > > > > > > > > > > >>>>>>>>>> impression >> > > > > > > > > > > > >>>>>>>>>>>>>>> that> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> this KIP is also going to cover >> > exposing >> > > > of >> > > > > the >> > > > > > > > > > standby >> > > > > > > > > > > > >>>>>>>>>>> information in> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> StreamsMetadata and thus subsume >> > > > KAFKA-8994 >> > > > > . >> > > > > > > That >> > > > > > > > > > would >> > > > > > > > > > > > >>>> require >> > > > > > > > > > > > >>>>>>>>>> a >> > > > > > > > > > > > >>>>>>>>>>>>>>> public> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> API change?> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> - KIP seems to be focussing on >> > restoration >> > > > > when >> > > > > > > a new >> > > > > > > > > > > > >> node >> > > > > > > > > > > > >>>> is >> > > > > > > > > > > > >>>>>>>>>>> added.> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> KIP-441 is underway and has some >> major >> > > > > changes >> > > > > > > > > > proposed >> > > > > > > > > > > > >>> for >> > > > > > > > > > > > >>>>>> this. >> > > > > > > > > > > > >>>>>>>>>>> It >> > > > > > > > > > > > >>>>>>>>>>>>>>> would> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> be good to clarify dependencies if >> > any. >> > > > > Without >> > > > > > > > > > > > >> KIP-441, I >> > > > > > > > > > > > >>>> am >> > > > > > > > > > > > >>>>>> not >> > > > > > > > > > > > >>>>>>>>>>> very >> > > > > > > > > > > > >>>>>>>>>>>>>>> sure> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> if we should allow reads from >> nodes in >> > > > > RESTORING >> > > > > > > > > > state, >> > > > > > > > > > > > >>>> which >> > > > > > > > > > > > >>>>>>>>>> could >> > > > > > > > > > > > >>>>>>>>>>>>>>> amount> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> to many minutes/few hours of stale >> > reads? >> > > > > This >> > > > > > > is >> > > > > > > > > > > > >>> different >> > > > > > > > > > > > >>>>>>>>>>>>>>> fromallowing> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> querying standby replicas, which >> > could be >> > > > > mostly >> > > > > > > > > > caught >> > > > > > > > > > > > >> up >> > > > > > > > > > > > >>>> and >> > > > > > > > > > > > >>>>>>>>>> the> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> staleness window could be much >> > > > > > > smaller/tolerable. >> > > > > > > > > > (once >> > > > > > > > > > > > >>>> again >> > > > > > > > > > > > >>>>>> the >> > > > > > > > > > > > >>>>>>>>>>> focus >> > > > > > > > > > > > >>>>>>>>>>>>>>> on> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> KAFKA-8994)> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> - Finally, we may need to >> introduce a >> > > > > > > configuration to >> > > > > > > > > > > > >>>> control >> > > > > > > > > > > > >>>>>>>>>>> this. >> > > > > > > > > > > > >>>>>>>>>>>>>>> Some> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> users may prefer errors to stale >> > data. Can >> > > > > we >> > > > > > > also add >> > > > > > > > > > > > >> it >> > > > > > > > > > > > >>>> to the >> > > > > > > > > > > > >>>>>>>>>>> KIP?> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> Thanks> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> Vinoth> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> On Sun, Oct 13, 2019 at 3:31 PM >> > Navinder >> > > > > Brar> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> <na...@yahoo.com.invalid>wrote:> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>>> Hi,> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>>> Starting a discussion on the KIP >> to >> > Allow >> > > > > state >> > > > > > > > > > stores >> > > > > > > > > > > > >> to >> > > > > > > > > > > > >>>> serve >> > > > > > > > > > > > >>>>>>>>>>>>> stale> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>>> reads during rebalance(> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > > > > >> > > > > > > >> > > > > >> > > > >> > >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-535%3A+Allow+state+stores+to+serve+stale+reads+during+rebalance >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>>> ).> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>>> Thanks & Regards,Navinder> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>>> LinkedIn> >> > > > > > > > > > > > >>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>>> -- >> > > > > > > > > > > > >>>>>>>>>> -- Guozhang >> > > > > > > > > > > > >>>>>>>>>> >> > > > > > > > > > > > >>>>>>>>> >> > > > > > > > > > > > >>>>>>>> >> > > > > > > > > > > > >>>>>>> >> > > > > > > > > > > > >>>>>> >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>>> >> > > > > > > > > > > > >>> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> >> > > > > > > > > > > > >> -- >> > > > > > > > > > > > >> -- Guozhang >> > > > > > > > > > > > >> >> > > > > > > > > > > > > >> > > > > > > > > > > > >> > > > > > > > > > > >> > > > > > > >> > > > > >> > > > >> > >> > >> >
