>during restoring state the active might have some lag
Great catch, yes.. we cannot assume lag = 0 for active. Lets report active
lag as well then. If active is too laggy, the app can then deem the store
partition unavailable (based on what the application is willing to
tolerate).
@matthias do you agree? We can then begin the vote.
On Thu, Nov 14, 2019 at 9:03 AM Navinder Brar
<navinder_b...@yahoo.com.invalid> wrote:
> I agree with Vinoth and John on having "allLocalStoreOffsetLags()", all
> actives don't have 0 lag, as during restoring state the active might have
> some lag and one of the features of this KIP is to provide an option to
> query from active (which might be in restoring state).
> I will update the KIP with rejected alternatives and post this will start
> a vote if everyone agrees on this.
> On Thursday, 14 November, 2019, 09:34:52 pm IST, John Roesler <
> j...@confluent.io> wrote:
>
> Hi all,
>
> Thanks for the "reset", Vinoth. It brings some needed clarity to the
> discussion.
>
> 10. My 2 cents: we might as well include the lags for the active
> copies as well. This is a more context-free API. If we only include
> standbys, this choice won't make sense to users unless they understand
> that the active task cannot lag in the steady state, since it's the
> source of updates. This isn't a bad thing to realize, but it's just
> more mental overhead for the person who wants to list the lags for
> "all local stores".
>
> Another reason is that we could consider also reporting the lag for
> actives during recovery (when they would have non-zero lag). We don't
> have to now, but if we choose to call the method "standby lags", then
> we can't make this choice in the future.
>
> That said, it's just my opinion. I'm fine either way.
>
> 20. Vinoth's reply makes sense to me, fwiw.
>
> Beyond these two points, I'm happy with the current proposal.
>
> Thanks again,
> -John
>
> On Thu, Nov 14, 2019 at 4:48 AM Vinoth Chandar <vchan...@confluent.io>
> wrote:
> >
> > 10. I considered that. Had to pick one or the other. Can just return
> > standby too and rename method to may be “allLocalStandbyOffsetLags()” to
> > have it explicit. (Standby should implicitly convey that we are talking
> > about stores)
> >
> > 20. What I meant was, we are returning HostInfo instead of
> StreamsMetadata
> > since thats sufficient to route query; same for “int partition “ vs topic
> > partition before. Previously KeyQueryMetadata had similar structure but
> > used StreamsMetadata and TopicPartition objects to convey same
> information
> >
> > @navinder KIP is already upto date with the email I sent, except for the
> > reasonings I was laying out. +1 on revisiting rejected alternatives.
> > Please make the follow up changes
> >
> > On Wed, Nov 13, 2019 at 9:12 PM Matthias J. Sax <matth...@confluent.io>
> > wrote:
> >
> > > Thanks for the summary Vinoth!
> > >
> > > I buy the overall argument. Couple of clarification questions:
> > >
> > >
> > > 10. Why do we need to include the active stores in
> > > `allLocalStoreOffsetLags()`? Would it not be simpler to just return lag
> > > for standbys?
> > >
> > >
> > > 20: What does
> > >
> > > > Thin the KeyQueryMetadata object to just contain the minimum
> information
> > > > needed.
> > >
> > > exaclty mean? What is the "minimum information needed" ?
> > >
> > >
> > > @Navinder: if you agree, can you update the KIP accoringly? With all
> the
> > > proposals, it's hard to keep track and it would be great to have the
> > > current proposal summarized in the wiki page.
> > >
> > > Please also update the "Rejected alternative" sections to avoid that we
> > > cycle back to old proposal (including the reason _why_ they got
> rejected).
> > >
> > >
> > > Thanks a lot!
> > >
> > >
> > > -Matthias
> > >
> > >
> > >
> > > On 11/13/19 7:10 PM, Vinoth Chandar wrote:
> > > > Given we have had a healthy discussion on this topic for a month now
> and
> > > > still with many loose ends and open ended conversations, I thought It
> > > would
> > > > be worthwhile to take a step back and re-evaluate everything in the
> > > context
> > > > of the very real use-case and its specific scenarios.
> > > >
> > > > First, let's remind ourselves of the query routing flow of the
> streams
> > > > application ("app" here on)
> > > >
> > > > 1. queries get routed to any random streams instance in the
> cluster
> > > > ("router" here on)
> > > > 2. router then uses Streams metadata to pick active/standby
> instances
> > > > for that key's store/partition
> > > > 3. router instance also maintains global lag information for all
> > > stores
> > > > and all their partitions, by a gossip/broadcast/heartbeat protocol
> > > (done
> > > > outside of Streams framework), but using
> KafkaStreams#allMetadata()
> > > for
> > > > streams instance discovery.
> > > > 4. router then uses information in 2 & 3 to determine which
> instance
> > > to
> > > > send the query to : always picks active instance if alive or the
> most
> > > > in-sync live standby otherwise.
> > > >
> > > > Few things to note :
> > > >
> > > > A) We choose to decouple how the lag information is obtained (control
> > > > plane) from query path (data plane), since that provides more
> flexibility
> > > > in designing the control plane. i.e pick any or combination of
> gossip,
> > > > N-way broadcast, control the rate of propagation, piggybacking on
> request
> > > > responses
> > > > B) Since the app needs to do its own control plane, talking to other
> > > > instances directly for failure detection & exchanging other
> metadata, we
> > > > can leave the lag APIs added to KafkaStreams class itself local and
> > > simply
> > > > return lag for all store/partitions on that instance.
> > > > C) Streams preserves its existing behavior of instances only talking
> to
> > > > each other through the Kafka brokers.
> > > > D) Since the router treats active/standby differently, it would be
> good
> > > for
> > > > the KafkaStreams APIs to hand them back explicitly, with no
> additional
> > > > logic needed for computing them. Specifically, the router only knows
> two
> > > > things - key and store and if we just return a
> > > Collection<StreamsMetadata>
> > > > back, it cannot easily tease apart active and standby. Say, a streams
> > > > instance hosts the same store as both active and standby for
> different
> > > > partitions, matching by just storename the app will find it in both
> > > active
> > > > and standby lists.
> > > > E) From above, we assume the global lag estimate (lag per store topic
> > > > partition) are continuously reported amongst application instances
> and
> > > > already available on the router during step 2 above. Hence,
> attaching lag
> > > > APIs to StreamsMetadata is unnecessary and does not solve the needs
> > > anyway.
> > > > F) Currently returned StreamsMetadata object is really information
> about
> > > a
> > > > streams instance, that is not very specific to the key being queried.
> > > > Specifically, router has no knowledge of the topic partition a given
> key
> > > > belongs, this is needed for matching to the global lag information in
> > > step
> > > > 4 above (and as also the example code in the KIP showed before). The
> > > > StreamsMetadata, since it's about the instance itself, would contain
> all
> > > > topic partitions and stores on that instance, not specific to the
> given
> > > key.
> > > > G) A cleaner API would thin the amount of information returned to
> > > > specifically the given key and store - which is partition the key
> belongs
> > > > to, active host info, list of standby host info. KeyQueryMetadata was
> > > > proposed in this spirit, but still hung onto StreamsMetadata as the
> > > member
> > > > field(s) which leaks more than what's required to route.
> > > > H) For this use case, its sufficient to just support offsetLag
> initially,
> > > > without needing time based lag information right away. So we
> probably
> > > > don't need a new top level StoreLagInfo class anymore.
> > > > I) On whether the local lag information is cached within Streams or
> App
> > > > code, its again preferable and be necessary for the App to be able to
> > > > invoke the lag api from an app thread and keep a cached state in
> memory,
> > > to
> > > > be piggybacked with request responses. If this information is also
> cached
> > > > inside Streams, it does no harm anyway.
> > > >
> > > > Based on this reasoning, I have made the following updates to the KIP
> > > >
> > > > - Drop the proposed KafkaStreams#allStandbyMetadata() and
> > > > KafkaStreams#allStandbyMetadataForStore() methods, with intent of
> > > > introducing minimal APIs to support the usage described above
> > > > - Drop the KafkaStreams#lagInfoForStore() method and just stick to
> one
> > > > method that returns all the local store partition's lags.
> > > > - Rename allLagInfo() to allLocalStoreOffsetLags() to be explicit
> that
> > > the
> > > > lags are local and only on offsets.
> > > > - Drop StoreLagInfo class
> > > > - allLocalStoreOffsetLags() returns Map<String, Map<Integer, Long>>,
> > > which
> > > > maps a store name to a map containing partition to offset lag info.
> It
> > > will
> > > > include both active and standby store partitions. active always has
> 0 lag
> > > > since its caught up with changelog topic.
> > > > - Thin the KeyQueryMetadata object to just contain the minimum
> > > information
> > > > needed. Rename allMetadataForKey() method variants to
> > > queryMetadataForKey()
> > > > variants that return KeyQueryMetadata
> > > > - Propose deprecating two current methods that return metadata based
> on
> > > key
> > > > : KafkaStreams#metadataForKey(..), to get more users to use the new
> > > > queryMetadataForKey APIs
> > > > - We will still have to enhance StreamsMetadata with fields for
> > > > standbyTopicPartitions and standbyStateStoreNames, since that is a
> core
> > > > object that gets updated upon rebalance.
> > > >
> > > >
> > > > Please let us know if this is agreeable. We will also work some of
> this
> > > > discussion into the background/proposed changes sections, upon
> feedback.
> > > >
> > > >
> > > > On Tue, Nov 12, 2019 at 9:17 AM Vinoth Chandar <
> vchan...@confluent.io>
> > > > wrote:
> > > >
> > > >> In all, is everyone OK with
> > > >>
> > > >> - Dropping KeyQueryMetadata, and the allMetadataForKey() apis
> > > >> - Dropping allLagInfo() from KafkaStreams class, Drop StoreLagInfo
> > > class
> > > >> - Add offsetLag(store, key, serializer) -> Optional<Long> &
> > > >> offsetLag(store, key, partitioner) -> Optional<Long> to
> StreamsMetadata
> > > >> - Duplicate the current methods for standbyMetadata in
> KafkaStreams :
> > > >> allStandbyMetadata(), allStandbyMetadataForStore(), two variants of
> > > >> standbyMetadataForKey(),
> > > >>
> > > >>
> > > >> Responses to Guozhang:
> > > >>
> > > >> 1.1 Like I mentioned before, the allStandbyMetadata() and
> > > >> allStandbyMetadataForStore() complement existing allMetadata() and
> > > >> allMetadataForStore(), since we don't want to change behavior of
> > > existing
> > > >> APIs. Based on discussions so far, if we decide to drop
> > > KeyQueryMetadata,
> > > >> then we will need to introduce 4 equivalents for standby metadata as
> > > >> Matthias mentioned.
> > > >> 1.2 I am okay with pushing lag information to a method on
> > > StreamsMetadata
> > > >> (Personally, I won't design it like that, but happy to live with it)
> > > like
> > > >> what Matthias suggested. But assuming topic name <=> store name
> > > equivalency
> > > >> for mapping this seems like a broken API to me. If all of Streams
> code
> > > were
> > > >> written like this, I can understand. But I don't think its the
> case? I
> > > >> would not be comfortable making such assumptions outside of public
> APIs.
> > > >>>> look into each one's standby partition / stores to tell which one
> > > >> StreamsMetadata is corresponding to the instance who holds a
> specific
> > > key
> > > >> as standby, yes, but I feel this one extra iteration is worth to
> avoid
> > > >> introducing a new class.
> > > >> This sort of thing would lead to non-standardized/potentially buggy
> > > client
> > > >> implementations, for something I expect the system would hand me
> > > directly.
> > > >> I don't personally feel introducing a new class is so bad, to
> warrant
> > > the
> > > >> user to do all this matching. Given the current APIs are not
> explicitly
> > > >> named to denote active metadata, it gives us a chance to build
> something
> > > >> more direct and clear IMO. If we do allMetadataForKey apis, then we
> > > should
> > > >> clearly separate active and standby ourselves. Alternate is separate
> > > active
> > > >> and standby APIs as Matthias suggests, which I can make peace with.
> > > >>
> > > >> 1.3 Similar as above. In Streams code, we treat topic partitions and
> > > store
> > > >> names separately?.
> > > >> 2.1 I think most databases build replication using logical offsets,
> not
> > > >> time. Time lag can be a nice to have feature, but offset lag is
> fully
> > > >> sufficient for a lot of use-cases.
> > > >> 2.2.1 We could support a lagInfoForStores() batch api. makes sense.
> > > >>
> > > >>
> > > >> Responses to Matthias :
> > > >> (100) Streams can still keep the upto date version in memory and
> > > >> implementation could be for now just reading this already refreshed
> > > value.
> > > >> Designing the API, with intent of pushing this to the user keeps
> doors
> > > >> open for supporting time based lag in the future.
> > > >> (101) I am not sure what the parameters of evaluating approaches
> here
> > > is.
> > > >> Generally, when I am handed a Metadata object, I don't expect to
> further
> > > >> query it for more information semantically. I would not also force
> user
> > > to
> > > >> make separate calls for active and standby metadata.
> > > >> Well, that may be just me. So sure, we can push this into
> > > StreamsMetadata
> > > >> if everyone agrees!
> > > >> +1 on duplicating all 4 methods for standbys in this case.
> > > >>
> > > >>
> > > >>
> > > >>
> > > >>
> > > >> On Tue, Nov 12, 2019 at 4:12 AM Navinder Brar
> > > >> <navinder_b...@yahoo.com.invalid> wrote:
> > > >>
> > > >>>
> > > >>> - Looking back, I agree that 2 calls to StreamsMetadata to fetch
> > > >>> StreamsMetadata and then using something like ‘long
> > > >>> StreamsMetadata#offsetLag(store, key)’ which Matthias suggested is
> > > better
> > > >>> than introducing a new public API i.e. ‘KeyQueryMetadata’. I will
> > > change
> > > >>> the KIP accordingly.
> > > >>> - >> I am actually not even sure, why we added
> > > >>> `StreamsMetadata#topicPartitions()` originally
> > > >>> I think it is helpful in showing which host holds which source
> topic
> > > >>> partitions for /instances endpoint and when you query a key, you
> need
> > > to
> > > >>> match the partition on which the key belongs to with the hosts
> holding
> > > >>> source topic partitions for that partition. Is there any other way
> to
> > > get
> > > >>> this info?
> > > >>>
> > > >>> On Tuesday, 12 November, 2019, 03:55:16 am IST, Guozhang Wang <
> > > >>> wangg...@gmail.com> wrote:
> > > >>>
> > > >>> Regarding 1.2: StreamsMetadata is 1-1 mapped to the streams
> > > instances, so
> > > >>> 1) allMetadata would still return the same number of
> StreamsMetadata in
> > > >>> collection, just that within the StreamsMetadata now you have new
> APIs
> > > to
> > > >>> access standby partitions / stores. So I think it would not be a
> > > breaking
> > > >>> change to the public API to not include `allStandbyMetadata` and `
> > > >>> allStandbyMetadataForStore` but still rely on
> `allMetadata(ForStore)`?
> > > >>>
> > > >>> Regarding 1.1: Good point about the partition number. But I'm still
> > > >>> wondering if it is definitely necessary to introduce a new
> > > >>> `KeyQueryMetadata`
> > > >>> interface class. E.g. suppose our function signature is
> > > >>>
> > > >>> Collection<StreamsMetadata> allMetadataForKey
> > > >>>
> > > >>> When you get the collection of StreamsMetadata you need to iterate
> over
> > > >>> the
> > > >>> collection and look into each one's standby partition / stores to
> tell
> > > >>> which one StreamsMetadata is corresponding to the instance who
> holds a
> > > >>> specific key as standby, yes, but I feel this one extra iteration
> is
> > > worth
> > > >>> to avoid introducing a new class.
> > > >>>
> > > >>>
> > > >>> Guozhang
> > > >>>
> > > >>> On Sat, Nov 9, 2019 at 10:04 PM Matthias J. Sax <
> matth...@confluent.io
> > > >
> > > >>> wrote:
> > > >>>
> > > >>>> I agree, that we might want to drop the time-base lag for the
> initial
> > > >>>> implementation. There is no good way to get this information
> without a
> > > >>>> broker side change.
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> (100) For the offset lag information, I don't see a reason why
> the app
> > > >>>> should drive when this information is updated though, because KS
> will
> > > >>>> update this information anyway (once per `commit.interval.ms` --
> and
> > > >>>> updating it more frequently does not make sense, as it most likely
> > > won't
> > > >>>> change more frequently anyway).
> > > >>>>
> > > >>>> If you all insist that the app should drive it, I can live with
> it,
> > > but
> > > >>>> I think it makes the API unnecessarily complex without a benefit.
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> (101) I still don't understand why we need to have
> `KeyQueryMetadata`
> > > >>>> though. Note, that an instance can only report lag for it's local
> > > >>>> stores, but not remote stores as it does not know to what offset a
> > > >>>> remote standby has caught up to.
> > > >>>>
> > > >>>>> Because we needed to return the topicPartition the key belongs
> to, in
> > > >>>>> order to correlate with the lag information from the other set of
> > > >>> APIs.
> > > >>>>
> > > >>>> My suggestion is to get the lag information from
> `StreamsMetadata` --
> > > >>>> which partition the store belongs to can be completely
> encapsulated
> > > >>>> within KS as all information is local, and I don't think we need
> to
> > > >>>> expose it to the user at all.
> > > >>>>
> > > >>>> We can just add `long StreamsMetadata#offsetLag(store, key)`. If
> the
> > > >>>> store is local we return its lag, if it's remote we return `-1`
> (ie,
> > > >>>> UNKNOWN). As an alternative, we can change the return type to
> > > >>>> `Optional<Long>`. This works for active and standby task alike.
> > > >>>>
> > > >>>> Note, that a user must verify if `StreamsMeatadata` is for itself
> > > >>>> (local) or remote anyway. We only need to provide a way that
> allows
> > > >>>> users to distinguish between active an standby. (More below.)
> > > >>>>
> > > >>>>
> > > >>>> I am actually not even sure, why we added
> > > >>>> `StreamsMetadata#topicPartitions()` originally -- seems pretty
> > > useless.
> > > >>>> Can we deprecate it as side cleanup in this KIP? Or do I miss
> > > something?
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> (102)
> > > >>>>
> > > >>>>> There are basically 2 reasons. One is that instead of having two
> > > >>>> functions, one to get StreamsMetadata for active and one for
> replicas.
> > > >>> We
> > > >>>> are fetching both in a single call and we have a way to get only
> > > active
> > > >>> or
> > > >>>> only replicas from the KeyQueryMetadata object(just like
> isStandby()
> > > and
> > > >>>> isActive() discussion we had earlier)
> > > >>>>
> > > >>>> I understand, that we need two methods. However, I think we can
> > > simplify
> > > >>>> the API and not introduce `KeyQueryMetadata`, but just "duplicate"
> > > all 4
> > > >>>> existing methods for standby tasks:
> > > >>>>
> > > >>>> // note that `standbyMetadataForKey` return a Collection in
> contrast
> > > to
> > > >>>> existing `metadataForKey`
> > > >>>>
> > > >>>>> Collection<StreamsMetadata> allStandbyMetadata()
> > > >>>>> Collection<StreamsMetadata> allStandbyMetadataForStore(String
> > > >>>> storeName)
> > > >>>>> Collection<StreamsMetadata> metadataForKey(String storeName, K
> key,
> > > >>>> Serializer<K> keySerializer)
> > > >>>>> Collection<StreamsMetadata> metadataForKey(String storeName, K
> key,
> > > >>>> StreamPartitioner<? super K, ?> partitioner)
> > > >>>>
> > > >>>> Because the existing methods return all active metadata, there is
> no
> > > >>>> reason to return `KeyQueryMetadata` as it's more complicated to
> get
> > > the
> > > >>>> standby metadata. With `KeyQueryMetadata` the user needs to make
> more
> > > >>>> calls to get the metadata:
> > > >>>>
> > > >>>> KafkaStreams#allMetadataForKey()
> > > >>>> #getActive()
> > > >>>>
> > > >>>> KafkaStreams#allMetadataForKey()
> > > >>>> #getStandby()
> > > >>>>
> > > >>>> vs:
> > > >>>>
> > > >>>> KafkaStreams#metadataForKey()
> > > >>>>
> > > >>>> KafkaStreams#standbyMetadataForKey()
> > > >>>>
> > > >>>> The wrapping of both within `KeyQueryMetadata` does not seem to
> > > provide
> > > >>>> any benefit but increase our public API surface.
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> @Guozhang:
> > > >>>>
> > > >>>> (1.1. + 1.2.) From my understanding `allMetadata()` (and other
> > > existing
> > > >>>> methods) will only return the metadata of _active_ tasks for
> backward
> > > >>>> compatibility reasons. If we would return standby metadata,
> existing
> > > >>>> code would potentially "break" because the code might pick a
> standby
> > > to
> > > >>>> query a key without noticing.
> > > >>>>
> > > >>>>
> > > >>>>
> > > >>>> -Matthias
> > > >>>>
> > > >>>>
> > > >>>> On 11/8/19 6:07 AM, Navinder Brar wrote:
> > > >>>>> Thanks, Guozhang for going through it again.
> > > >>>>>
> > > >>>>> - 1.1 & 1.2: The main point of adding topicPartition in
> > > >>>> KeyQueryMetadata is not topicName, but the partition number. I
> agree
> > > >>>> changelog topicNames and store names will have 1-1 mapping but we
> also
> > > >>> need
> > > >>>> the partition number of the changelog for which are calculating
> the
> > > lag.
> > > >>>> Now we can add partition number in StreamsMetadata but it will be
> > > >>>> orthogonal to the definition of StreamsMetadata i.e.- “Represents
> the
> > > >>> state
> > > >>>> of an instance (process) in a {@link KafkaStreams} application.”
> If
> > > we
> > > >>> add
> > > >>>> partition number in this, it doesn’t stay metadata for an
> instance,
> > > >>> because
> > > >>>> now it is storing the partition information for a key being
> queried.
> > > So,
> > > >>>> having “KeyQueryMetadata” simplifies this as now it contains all
> the
> > > >>>> metadata and also changelog and partition information for which we
> > > need
> > > >>> to
> > > >>>> calculate the lag.
> > > >>>>>
> > > >>>>> Another way is having another function in parallel to
> metadataForKey,
> > > >>>> which returns the partition number for the key being queried. But
> then
> > > >>> we
> > > >>>> would need 2 calls to StreamsMetadataState, once to fetch
> metadata and
> > > >>>> another to fetch partition number. Let me know if any of these two
> > > ways
> > > >>>> seem more intuitive than KeyQueryMetadata then we can try to
> converge
> > > on
> > > >>>> one.
> > > >>>>> - 1.3: Again, it is required for the partition number. We can
> > > drop
> > > >>>> store name though.
> > > >>>>> - 2.1: I think this was done in accordance with the opinion
> from
> > > >>> John
> > > >>>> as time lag would be better implemented with a broker level
> change and
> > > >>>> offset change is readily implementable. @vinoth?
> > > >>>>> - 2.2.1: Good point. +1
> > > >>>>> - 2.2.2: I am not well aware of it, @vinoth any comments?
> > > >>>>> - 3.1: I think we have already agreed on dropping this, we
> need to
> > > >>>> KIP. Also, is there any opinion on lagInfoForStore(String
> storeName)
> > > vs
> > > >>>> lagInfoForStore(String storeName, int partition)
> > > >>>>> - 3.2: But in functions such as onAssignment(),
> > > >>>> onPartitionsAssigned(), for standbyTasks also the topicPartitions
> we
> > > use
> > > >>>> are input topic partitions and not changelog partitions. Would
> this be
> > > >>>> breaking from that semantics?
> > > >>>>>
> > > >>>>>
> > > >>>>> On Thursday, 7 November, 2019, 11:33:19 pm IST, Guozhang Wang
> <
> > > >>>> wangg...@gmail.com> wrote:
> > > >>>>>
> > > >>>>> Hi Navinder, Vinoth, thanks for the updated KIP!
> > > >>>>>
> > > >>>>> Read through the discussions so far and made another pass on the
> wiki
> > > >>>> page,
> > > >>>>> and here are some more comments:
> > > >>>>>
> > > >>>>> 1. About the public APIs:
> > > >>>>>
> > > >>>>> 1.1. It is not clear to me how allStandbyMetadataForStore
> > > >>>>> and allStandbyMetadata would be differentiated from the original
> APIs
> > > >>>> given
> > > >>>>> that we will augment StreamsMetadata to include both active and
> > > >>> standby
> > > >>>>> topic-partitions and store names, so I think we can still use
> > > >>> allMetadata
> > > >>>>> and allMetadataForStore to get the collection of instance
> metadata
> > > >>> that
> > > >>>>> host the store both as active and standbys. Are there any
> specific
> > > use
> > > >>>>> cases where we ONLY want to get the standby's metadata? And even
> if
> > > >>> there
> > > >>>>> are, we can easily filter it out from the allMetadata /
> > > >>>> allMetadataForStore
> > > >>>>> right?
> > > >>>>>
> > > >>>>> 1.2. Similarly I'm wondering for allMetadataForKey, can we
> return the
> > > >>>> same
> > > >>>>> type: "Collection<StreamsMetadata>" which includes 1 for active,
> and
> > > >>> N-1
> > > >>>>> for standbys, and callers can easily identify them by looking
> inside
> > > >>> the
> > > >>>>> StreamsMetadata objects? In addition I feel the "topicPartition"
> > > field
> > > >>>>> inside "KeyQueryMetadata" is not very important since the
> changelog
> > > >>>>> topic-name is always 1-1 mapping to the store name, so as long
> as the
> > > >>>> store
> > > >>>>> name matches, the changelog topic name should always match (i.e.
> in
> > > >>>>> the pseudo code, just checking store names should be
> sufficient). If
> > > >>> all
> > > >>>> of
> > > >>>>> the above assumption is true, I think we can save us from
> introducing
> > > >>> one
> > > >>>>> more public class here.
> > > >>>>>
> > > >>>>> 1.3. Similarly in StoreLagInfo, seems not necessary to include
> the
> > > >>> topic
> > > >>>>> partition name in addition to the store name.
> > > >>>>>
> > > >>>>> 2. About querying store lags: we've discussed about separating
> the
> > > >>>> querying
> > > >>>>> of the lag information and the querying of the host information
> so I
> > > >>>> still
> > > >>>>> support having separate APIs here. More thoughts:
> > > >>>>>
> > > >>>>> 2.1. Compared with offsets, I'm wondering would time-difference
> be
> > > >>> more
> > > >>>>> intuitive for users to define the acceptable "staleness"? More
> > > >>> strictly,
> > > >>>>> are there any scenarios where we would actually prefer offsets
> over
> > > >>>>> timestamps except that the timestamps are not available?
> > > >>>>>
> > > >>>>> 2.2. I'm also a bit leaning towards not putting the burden of
> > > >>>> periodically
> > > >>>>> refreshing our lag and caching it (and introducing another
> config) on
> > > >>> the
> > > >>>>> streams side but document clearly its cost and let users to
> consider
> > > >>> its
> > > >>>>> call frequency; of course in terms of implementation there are
> some
> > > >>>>> optimizations we can consider:
> > > >>>>>
> > > >>>>> 1) for restoring active tasks, the log-end-offset is read once
> since
> > > >>> it
> > > >>>> is
> > > >>>>> not expected to change, and that offset / timestamp can be
> remembered
> > > >>> for
> > > >>>>> lag calculation and we do not need to refresh again;
> > > >>>>> 2) for standby tasks, there's a "Map<TopicPartition, Long>
> > > >>>>> endOffsets(Collection<TopicPartition> partitions)" in
> KafkaConsumer
> > > to
> > > >>>>> batch a list of topic-partitions in one round-trip, and we can
> use
> > > >>> that
> > > >>>> to
> > > >>>>> let our APIs be sth. like "lagInfoForStores(Collection<String>
> > > >>>> storeNames)"
> > > >>>>> to enable the batching effects.
> > > >>>>>
> > > >>>>> 3. Misc.:
> > > >>>>>
> > > >>>>> 3.1 There's a typo on the pseudo code "globalLagInforation".
> Also it
> > > >>>> seems
> > > >>>>> not describing how that information is collected (personally I
> also
> > > >>> feel
> > > >>>>> one "lagInfoForStores" is sufficient).
> > > >>>>> 3.2 Note there's a slight semantical difference between active
> and
> > > >>>>> standby's "partitions" inside StreamsMetadata, for active tasks
> the
> > > >>>>> partitions are actually input topic partitions for the task:
> some of
> > > >>> them
> > > >>>>> may also act as changelog topics but these are exceptional
> cases; for
> > > >>>>> standby tasks the "standbyTopicPartitions" are actually the
> changelog
> > > >>>>> topics of the task. So maybe renaming it to
> > > >>> "standbyChangelogPartitions"
> > > >>>> to
> > > >>>>> differentiate it?
> > > >>>>>
> > > >>>>>
> > > >>>>> Overall I think this would be a really good KIP to add to
> Streams,
> > > >>> thank
> > > >>>>> you so much!
> > > >>>>>
> > > >>>>>
> > > >>>>> Guozhang
> > > >>>>>
> > > >>>>> On Wed, Nov 6, 2019 at 8:47 PM Navinder Brar
> > > >>>>> <navinder_b...@yahoo.com.invalid> wrote:
> > > >>>>>
> > > >>>>>> +1 on implementing offset based lag for now and push time-based
> lag
> > > >>> to a
> > > >>>>>> later point in time when broker changes are done. Although
> > > time-based
> > > >>>> lag
> > > >>>>>> enhances the readability, it would not be a make or break
> change for
> > > >>>>>> implementing this KIP.
> > > >>>>>>
> > > >>>>>> Vinoth has explained the role of KeyQueryMetadata, let me in
> add in
> > > >>> my 2
> > > >>>>>> cents as well.
> > > >>>>>> - There are basically 2 reasons. One is that instead of
> having
> > > two
> > > >>>>>> functions, one to get StreamsMetadata for active and one for
> > > >>> replicas.
> > > >>>> We
> > > >>>>>> are fetching both in a single call and we have a way to get only
> > > >>> active
> > > >>>> or
> > > >>>>>> only replicas from the KeyQueryMetadata object(just like
> isStandby()
> > > >>> and
> > > >>>>>> isActive() discussion we had earlier)
> > > >>>>>> - Since even after fetching the metadata now we have a
> > > requirement
> > > >>>> of
> > > >>>>>> fetching the topicPartition for which the query came:- to fetch
> lag
> > > >>> for
> > > >>>>>> that specific topicPartition. Instead of having another call to
> > > fetch
> > > >>>> the
> > > >>>>>> partition from StreamsMetadataState we thought using one single
> call
> > > >>> and
> > > >>>>>> fetching partition and all metadata would be better.
> > > >>>>>> - Another option was to change StreamsMetadata object and add
> > > >>>>>> topicPartition in that for which the query came but it doesn’t
> make
> > > >>>> sense
> > > >>>>>> in terms of semantics as it StreamsMetadata. Also,
> KeyQueryMetadata
> > > >>>>>> represents all the metadata for the Key being queried, i.e. the
> > > >>>> partition
> > > >>>>>> it belongs to and the list of StreamsMetadata(hosts) active or
> > > >>> replica
> > > >>>>>> where the key could be found.
> > > >>>>>>
> > > >>>>>>
> > > >>>>>>
> > > >>>>>>
> > > >>>>>>
> > > >>>>>> On Thursday, 7 November, 2019, 01:53:36 am IST, Vinoth
> Chandar <
> > > >>>>>> vchan...@confluent.io> wrote:
> > > >>>>>>
> > > >>>>>> +1 to John, suggestion on Duration/Instant and dropping the
> API to
> > > >>>> fetch
> > > >>>>>> all store's lags. However, I do think we need to return lags per
> > > >>> topic
> > > >>>>>> partition. So not sure if single return value would work? We
> need
> > > >>> some
> > > >>>> new
> > > >>>>>> class that holds a TopicPartition and Duration/Instant variables
> > > >>>> together?
> > > >>>>>>
> > > >>>>>> 10) Because we needed to return the topicPartition the key
> belongs
> > > >>> to,
> > > >>>> in
> > > >>>>>> order to correlate with the lag information from the other set
> of
> > > >>> APIs.
> > > >>>>>> Otherwise, we don't know which topic partition's lag estimate to
> > > >>> use. We
> > > >>>>>> tried to illustrate this on the example code. StreamsMetadata is
> > > >>> simply
> > > >>>>>> capturing state of a streams host/instance, where as
> TopicPartition
> > > >>>> depends
> > > >>>>>> on the key passed in. This is a side effect of our decision to
> > > >>> decouple
> > > >>>> lag
> > > >>>>>> based filtering on the metadata apis.
> > > >>>>>>
> > > >>>>>> 20) Goes back to the previous point. We needed to return
> information
> > > >>>> that
> > > >>>>>> is key specific, at which point it seemed natural for the
> > > >>>> KeyQueryMetadata
> > > >>>>>> to contain active, standby, topic partition for that key. If we
> > > >>> merely
> > > >>>>>> returned a standbyMetadataForKey() ->
> Collection<StreamsMetadata>
> > > >>>> standby,
> > > >>>>>> an active metadataForKey() -> StreamsMetadata, and new
> > > >>>>>> getTopicPartition(key) -> topicPartition object back to the
> caller,
> > > >>> then
> > > >>>>>> arguably you could do the same kind of correlation. IMO having
> a the
> > > >>>>>> KeyQueryMetadata class to encapsulate all this is a friendlier
> API.
> > > >>>>>> allStandbyMetadata() and allStandbyMetadataForStore() are just
> > > >>> counter
> > > >>>>>> parts for metadataForStore() and allMetadata() that we introduce
> > > >>> mostly
> > > >>>> for
> > > >>>>>> consistent API semantics. (their presence implicitly could help
> > > >>> denote
> > > >>>>>> metadataForStore() is for active instances. Happy to drop them
> if
> > > >>> their
> > > >>>>>> utility is not clear)
> > > >>>>>>
> > > >>>>>> 30) This would assume we refresh all the standby lag information
> > > >>> every
> > > >>>>>> time we query for that StreamsMetadata for a specific store? For
> > > time
> > > >>>> based
> > > >>>>>> lag, this will involve fetching the tail kafka record at once
> from
> > > >>>> multiple
> > > >>>>>> kafka topic partitions? I would prefer not to couple them like
> this
> > > >>> and
> > > >>>>>> have the ability to make granular store (or even topic partition
> > > >>> level)
> > > >>>>>> fetches for lag information.
> > > >>>>>>
> > > >>>>>> 32) I actually prefer John's suggestion to let the application
> drive
> > > >>> the
> > > >>>>>> lag fetches/updation and not have flags as the KIP current
> points
> > > to.
> > > >>>> Are
> > > >>>>>> you reexamining that position?
> > > >>>>>>
> > > >>>>>> On fetching lag information, +1 we could do this much more
> > > >>> efficiently
> > > >>>> with
> > > >>>>>> a broker changes. Given I don't yet have a burning need for the
> time
> > > >>>> based
> > > >>>>>> lag, I think we can sequence the APIs such that the offset based
> > > ones
> > > >>>> are
> > > >>>>>> implemented first, while we have a broker side change?
> > > >>>>>> Given we decoupled the offset and time based lag API, I am
> willing
> > > to
> > > >>>> drop
> > > >>>>>> the time based lag functionality (since its not needed right
> away
> > > >>> for my
> > > >>>>>> use-case). @navinder . thoughts?
> > > >>>>>>
> > > >>>>>>
> > > >>>>>> On Tue, Nov 5, 2019 at 11:10 PM Matthias J. Sax <
> > > >>> matth...@confluent.io>
> > > >>>>>> wrote:
> > > >>>>>>
> > > >>>>>>> Navinder,
> > > >>>>>>>
> > > >>>>>>> thanks for updating the KIP. Couple of follow up questions:
> > > >>>>>>>
> > > >>>>>>>
> > > >>>>>>> (10) Why do we need to introduce the class `KeyQueryMetadata`?
> > > >>>>>>>
> > > >>>>>>> (20) Why do we introduce the two methods `allMetadataForKey()`?
> > > >>> Would
> > > >>>> it
> > > >>>>>>> not be simpler to add `Collection<StreamMetatdata>
> > > >>>>>>> standbyMetadataForKey(...)`. This would align with new methods
> > > >>>>>>> `#allStandbyMetadata()` and `#allStandbyMetadataForStore()`?
> > > >>>>>>>
> > > >>>>>>> (30) Why do we need the class `StoreLagInfo` -- it seems
> simpler to
> > > >>>> just
> > > >>>>>>> extend `StreamMetadata` with the corresponding attributes and
> > > >>> methods
> > > >>>>>>> (of active task, the lag would always be reported as zero)
> > > >>>>>>>
> > > >>>>>>> (32) Via (30) we can avoid the two new methods `#allLagInfo()`
> and
> > > >>>>>>> `#lagInfoForStore()`, too, reducing public API and making it
> > > >>> simpler to
> > > >>>>>>> use the feature.
> > > >>>>>>>
> > > >>>>>>> Btw: If we make `StreamMetadata` thread safe, the lag
> information
> > > >>> can
> > > >>>> be
> > > >>>>>>> updated in the background without the need that the application
> > > >>>>>>> refreshes its metadata. Hence, the user can get active and/or
> > > >>> standby
> > > >>>>>>> metadata once, and only needs to refresh it, if a rebalance
> > > >>> happened.
> > > >>>>>>>
> > > >>>>>>>
> > > >>>>>>> About point (4) of the previous thread: I was also thinking
> about
> > > >>>>>>> when/how to update the time-lag information, and I agree that
> we
> > > >>> should
> > > >>>>>>> not update it for each query.
> > > >>>>>>>
> > > >>>>>>> "How": That we need to fetch the last record is a little bit
> > > >>>>>>> unfortunate, but I don't see any other way without a broker
> change.
> > > >>> One
> > > >>>>>>> issue I still see is with "exactly-once" -- if transaction
> markers
> > > >>> are
> > > >>>>>>> in the topic, the last message is not at offset "endOffset -
> 1" and
> > > >>> as
> > > >>>>>>> multiple transaction markers might be after each other, it's
> > > unclear
> > > >>>> how
> > > >>>>>>> to identify the offset of the last record... Thoughts?
> > > >>>>>>>
> > > >>>>>>> Hence, it might be worth to look into a broker change as a
> > > potential
> > > >>>>>>> future improvement. It might be possible that the broker
> caches the
> > > >>>>>>> latest timestamp per partition to serve this data efficiently,
> > > >>> similar
> > > >>>>>>> to `#endOffset()`.
> > > >>>>>>>
> > > >>>>>>> "When": We refresh the end-offset information based on the
> > > >>>>>>> `commit.interval.ms` -- doing it more often is not really
> useful,
> > > >>> as
> > > >>>>>>> state store caches will most likely buffer up all writes to
> > > >>> changelogs
> > > >>>>>>> anyway and are only flushed on commit (including a flush of the
> > > >>>>>>> producer). Hence, I would suggest to update the time-lag
> > > information
> > > >>>>>>> based on the same strategy in the background. This way there
> is no
> > > >>>>>>> additional config or methods and the user does not need to
> worry
> > > >>> about
> > > >>>>>>> it at all.
> > > >>>>>>>
> > > >>>>>>> To avoid refresh overhead if we don't need it (a user might
> not use
> > > >>> IQ
> > > >>>>>>> to begin with), it might be worth to maintain an internal flag
> > > >>>>>>> `updateTimeLagEnabled` that is set to `false` initially and
> only
> > > >>> set to
> > > >>>>>>> `true` on the first call of a user to get standby-metadata.
> > > >>>>>>>
> > > >>>>>>>
> > > >>>>>>> -Matthias
> > > >>>>>>>
> > > >>>>>>>
> > > >>>>>>>
> > > >>>>>>> On 11/4/19 5:13 PM, Vinoth Chandar 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.
> > > >>>
