I looked into how to use a NetworkClient to replace StreamsKafkaClient to do this one-time checking, and the complexity is actually pretty high: since it is a barebone NetworkClient, we have to handle the connection / readiness / find a broker to send to / etc logic plus introducing all these dependencies into KafkaStreams class. So I have decided to not do this in this KIP. If people feel strongly about this let's discuss more.
I'll start the voting process on the mailing list now. Guozhang On Fri, Nov 10, 2017 at 11:47 AM, Bill Bejeck <bbej...@gmail.com> wrote: > I'm leaning towards option 3, although option 2 is a reasonable tradeoff > between the two. > > Overall I'm leaning towards option 3 because: > > 1. As Guozhang has said we are failing "fast enough" with an Exception > from the first rebalance. > 2. Less complexity/maintenance cost by not having a transient network > client > 3. Ideally, this check should be on the AdminClient itself but adding > such a check creates "scope creep" for this KIP. > > IMHO the combination of these reasons makes option 3 my preferred approach. > > Thanks, > Bill > > On Tue, Nov 7, 2017 at 12:48 PM, Guozhang Wang <wangg...@gmail.com> wrote: > > > Here is what I'm thinking about this trade-off: we want to fail fast when > > brokers do not yet support the requested API version, with the cost that > we > > need to do this one-time thing with an expensed NetworkClient plus a bit > > longer starting up latency. Here are a few different options: > > > > 1) we ask all the brokers: this is one extreme of the trade-off, we still > > need to handle UnsupportedApiVersionsException since brokers can > > downgrade. > > 2) we ask a random broker: this is what we did, and a bit weaker than 1) > > but saves on latency. > > 3) we do not ask anyone: this is the other extreme of the trade-off. > > > > To me I think 1) is an overkill, so I did not include that in my > proposals. > > Between 2) and 3) I'm slightly preferring 3) since even under this case > we > > are sort of failing fast because we will throw the exception at the first > > rebalance, but I can still see the value of option 2). Ideally we can > have > > some APIs from AdminClient to check API versions but this does not exist > > today and I do not want to drag too long with growing scope on this KIP, > so > > the current proposal's implementation uses expendable Network which is a > > bit fuzzy. > > > > > > Guozhang > > > > > > On Tue, Nov 7, 2017 at 2:07 AM, Matthias J. Sax <matth...@confluent.io> > > wrote: > > > > > I would prefer to keep the current check. We could even improve it, and > > > do the check to more than one brokers (even all provided > > > bootstrap.servers) or some random servers after we got all meta data > > > about the cluster. > > > > > > > > > -Matthias > > > > > > On 11/7/17 1:01 AM, Guozhang Wang wrote: > > > > Hello folks, > > > > > > > > One update I'd like to propose regarding "compatibility checking": > > > > currently we create a single StreamsKafkaClient at the beginning to > > issue > > > > an ApiVersionsRequest to a random broker and then check on its > > versions, > > > > and fail if it does not satisfy the version (0.10.1+ without EOS, > > 0.11.0 > > > > with EOS); after this check we throw this client away. My original > plan > > > is > > > > to replace this logic with the NetworkClient's own apiVersions, but > > after > > > > some digging while working on the PR I found that exposing this > > > apiVersions > > > > variable from NetworkClient through AdminClient is not very straight > > > > forward, plus it would need an API change on the AdminClient itself > as > > > well > > > > to expose the versions information. > > > > > > > > On the other hand, this one-time compatibility checking's benefit may > > be > > > > not significant: 1) it asks a random broker upon starting up, and > hence > > > > does not guarantee all broker's support the corresponding API > versions > > at > > > > that time; 2) brokers can still be upgraded / downgraded after the > > > streams > > > > app is up and running, and hence we still need to handle > > > > UnsupportedVersionExceptions thrown from the producer / consumer / > > admin > > > > client during the runtime anyways. > > > > > > > > So I'd like to propose two options in this KIP: > > > > > > > > 1) we remove this one-time compatibility check on Streams starting up > > in > > > > this KIP, and solely rely on handling producer / consumer / admin > > > client's > > > > API UnsupportedVersionException throwable exceptions. Please share > your > > > > thoughts about this. > > > > > > > > 2) we create a one-time NetworkClient upon starting up, send the > > > > ApiVersionsRequest and get the response and do the checking; after > that > > > > throw this client away. > > > > > > > > Please let me know what do you think. Thanks! > > > > > > > > > > > > Guozhang > > > > > > > > > > > > On Mon, Nov 6, 2017 at 7:56 AM, Matthias J. Sax < > matth...@confluent.io > > > > > > > wrote: > > > > > > > >> Thanks for the update and clarification. > > > >> > > > >> Sounds good to me :) > > > >> > > > >> > > > >> -Matthias > > > >> > > > >> > > > >> > > > >> On 11/6/17 12:16 AM, Guozhang Wang wrote: > > > >>> Thanks Matthias, > > > >>> > > > >>> 1) Updated the KIP page to include KAFKA-6126. > > > >>> 2) For passing configs, I agree, will make a pass over the existing > > > >> configs > > > >>> passed to StreamsKafkaClient and update the wiki page accordingly, > to > > > >>> capture all changes that would happen for the replacement in this > > > single > > > >>> KIP. > > > >>> 3) For internal topic purging, I'm not sure if we need to include > > this > > > >> as a > > > >>> public change since internal topics are meant for abstracted away > > from > > > >> the > > > >>> Streams users; they should not leverage such internal topics > > elsewhere > > > >>> themselves. The only thing I can think of is for Kafka operators > this > > > >> would > > > >>> mean that such internal topics would be largely reduced in their > > > >> footprint, > > > >>> but that would not be needed in the KIP as well. > > > >>> > > > >>> > > > >>> Guozhang > > > >>> > > > >>> > > > >>> On Sat, Nov 4, 2017 at 9:00 AM, Matthias J. Sax < > > matth...@confluent.io > > > > > > > >>> wrote: > > > >>> > > > >>>> I like this KIP. Can you also link to > > > >>>> https://issues.apache.org/jira/browse/KAFKA-6126 in the KIP? > > > >>>> > > > >>>> What I am wondering though: if we start to partially (ie, step by > > > step) > > > >>>> replace the existing StreamsKafkaClient with Java AdminClient, > don't > > > we > > > >>>> need more KIPs? For example, if we use purge-api for internal > > topics, > > > it > > > >>>> seems like a change that requires a KIP. Similar for passing > configs > > > -- > > > >>>> the old client might have different config than the old client? > Can > > we > > > >>>> double check this? > > > >>>> > > > >>>> Thus, it might make sense to replace the old client with the new > one > > > in > > > >>>> one shot. > > > >>>> > > > >>>> > > > >>>> -Matthias > > > >>>> > > > >>>> On 11/4/17 4:01 AM, Ted Yu wrote: > > > >>>>> Looks good overall. > > > >>>>> > > > >>>>> bq. the creation within StreamsPartitionAssignor > > > >>>>> > > > >>>>> Typo above: should be StreamPartitionAssignor > > > >>>>> > > > >>>>> On Fri, Nov 3, 2017 at 4:49 PM, Guozhang Wang < > wangg...@gmail.com> > > > >>>> wrote: > > > >>>>> > > > >>>>>> Hello folks, > > > >>>>>> > > > >>>>>> I have filed a new KIP on adding AdminClient into Streams for > > > internal > > > >>>>>> topic management. > > > >>>>>> > > > >>>>>> Looking for feedback on > > > >>>>>> > > > >>>>>> *https://cwiki.apache.org/confluence/display/KAFKA/KIP- > > > >>>>>> 220%3A+Add+AdminClient+into+Kafka+Streams%27+ClientSupplier > > > >>>>>> <https://cwiki.apache.org/confluence/display/KAFKA/KIP- > > > >>>>>> 220%3A+Add+AdminClient+into+Kafka+Streams%27+ClientSupplier>* > > > >>>>>> > > > >>>>>> -- > > > >>>>>> -- Guozhang > > > >>>>>> > > > >>>>> > > > >>>> > > > >>>> > > > >>> > > > >>> > > > >> > > > >> > > > > > > > > > > > > > > > > > > > > -- > > -- Guozhang > > > -- -- Guozhang
