10/20: I think I'm aligned with John's replies as well.
Guozhang
On Fri, Nov 15, 2019 at 1:45 AM Vinoth Chandar <vchan...@confluent.io>
wrote:
> >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.
> > > > >>>
>
--
-- Guozhang
