Hi Omnia, Thanks for putting this together into the KIP! I'll have a look when available.
Thanks. Luke On Thu, Jun 1, 2023 at 1:50 AM Justine Olshan <jols...@confluent.io> wrote: > Hey Omnia, > > I was doing a bit of snooping (I actually just get updates for the KIP > page) and I saw this draft was in progress. I shared it with some of my > colleagues as well who I previously discussed the issue with. > > The initial look was pretty promising to me. I appreciate the detailing of > the rejected options since we had quite a few we worked through :) > > One question I have is how will we handle a scenario where potentially > each new client has a new Kafka Principal? Is that simply not covered by > throttling? > > Thanks, > Justine > > On Wed, May 31, 2023 at 10:08 AM Omnia Ibrahim <o.g.h.ibra...@gmail.com> > wrote: > >> Hi Justine and Luke, >> >> I started a KIP draft here >> https://cwiki.apache.org/confluence/display/KAFKA/KIP-936%3A+Throttle+number+of+active+PIDs >> for a proposal would appreciate it if you could provide any initial >> feedback before opening a broader discussion. >> >> Thanks >> >> On Wed, Feb 22, 2023 at 4:35 PM Omnia Ibrahim <o.g.h.ibra...@gmail.com> >> wrote: >> >>> >>> *Hi Justine, * >>> >>> *My initial thought of throttling the initProducerId was to get ripped >>> off the problem at the source (which creates too many PIDs per client) and >>> fail faster but if having this on the produce request level is easier this >>> should be fine. I am guessing it will be the same direction as we may >>> ClientQuotaManage for Produce throttling with a different quota window than >>> `quota.window.size.seconds `. * >>> >>> *If this is good as an initial solution I can put start a KIP and see >>> what the wider community feels about this. * >>> >>> *Also, I noticed that at some point one of us hit "Replay" instead of >>> "Replay to All" :) So here are the previous conversations* >>> >>> *On Wed, Feb 15, 2023 at 12:20 AM Justine Olshan <jols...@confluent.io >>> <jols...@confluent.io>> wrote:* >>> >>>> Hey Omnia, >>>> >>>> Thanks for the response. I think I understand your explanations here >>>> with respect to principal and clientId usage. >>>> >>>> For the throttling -- handleInitProducerIdRequest will allocate the ID >>>> to the producer, but we don't actually store it on the broker or increment >>>> our metric until the first produce request for that producer is sent (or >>>> sent again after previously expiring). Would you consider throttling the >>>> produce request instead? It may be hard to get any metrics from the >>>> transaction coordinator where the initProducerId request is handled. >>>> >>>> Justine >>> >>> >>> *On Tue, Feb 14, 2023 at 9:29 AM Omnia Ibrahim <o.g.h.ibra...@gmail.com >>> <o.g.h.ibra...@gmail.com>> wrote:* >>> >>>> Hey Justine, >>>> > If I understand your message correctly, there are issues with >>>> identifying the source of the rogue clients? So you propose to add a new >>>> metric for that? >>>> > And also proposing to throttle based on clientId as a potential >>>> follow up? >>>> I want to identify rogue clients by KafkaPrincipal (and/or clientId) >>>> similarly to how we identify clients in Fetch/Produce/Request >>>> QuotaManagers. Using KafkaPrincipal should give cluster admin the ability >>>> to throttle later based on principal which is most likely to be a smaller >>>> set than clientIds. My initial thought was to add a metrics that represent >>>> how many InitProducerIDRequest are sent by KafkaPrincipal (and/or clientId) >>>> similar to Fetch/Produce QuotaManagers. >>>> Then as a follow-up, we can throttle based on either KafkaPrinciple or >>>> clientId (maybe default as well to align this with other QuotaManagers in >>>> Kafka). >>>> >>>> >1. Does we rely on the client using the same ID? What if there are >>>> many clients that all use different client IDs? >>>> This is why I want to use the combination of KafkaPrincipal or clientId >>>> similar to some other quotas we have in Kafka already. This will be a >>>> similar risk to Fetch/Produce quota in Kafka which also relay on the client >>>> to use the same clientId and KafkaPrincipal. >>>> >>>> >2. Are there places where high cardinality of this metric is a >>>> concern? I can imagine many client IDs in the system. Would we treat this >>>> as a rate metric (ie, when we get an init producer ID and return a new >>>> producer ID we emit a count for that client id?) Or something else? >>>> My initial thought here was to follow the steps of ClientQuotaManager >>>> and ClientRequestQuotaManager and use a rate metric. However, I think we >>>> can emit it either >>>> >>>> 1. when we return the new PID. However, I have concerns that we may >>>> circle back to the previous concerns with OMM due to keeping track of >>>> ACTIVE PIDs per KafkaPrincipal(and/or) clientId in the future. Also this >>>> would be the first time Kafka throttle IDs for any client. >>>> 2. or once we recieve initProducerIDRequest and throttle before >>>> even hitting `handleInitProducerIdRequest`. Going this direction we may >>>> need to throttle it within a different quota window than ` >>>> quota.window.size.seconds ` as throttling INIT_PRODUCER_ID request >>>> per second wouldn't be efficient for most cases. I personally think this >>>> direction is easier as it seems more consistent with the existing quota >>>> implementation. Specially that Kafka has already the concept of >>>> throttling >>>> subset of requests (in ControllerMutationQuotaManager) but never had any >>>> concept of throttling active IDs for any client. >>>> >>>> >>>> What do you think? >>>> >>>> Thanks >>>> Omnia >>>> >>> >>> *On Thu, Feb 2, 2023 at 5:39 PM Justine Olshan <jols...@confluent.io >>> <jols...@confluent.io>> wrote:* >>> >>>> Hey Omnia, >>>> Sorry for losing track of this. >>>> >>>> If I understand your message correctly, there are issues with >>>> identifying the source of the rogue clients? So you propose to add a new >>>> metric for that? >>>> And also proposing to throttle based on clientId as a potential follow >>>> up? >>>> >>>> I think both of these make sense. The only things I can think of for >>>> the metric are: >>>> 1. Does we rely on the client using the same ID? What if there are many >>>> clients that all use different client IDs? >>>> 2. Are there places where high cardinality of this metric is a concern? >>>> I can imagine many client IDs in the system. Would we treat this as a rate >>>> metric (ie, when we get an init producer ID and return a new producer ID we >>>> emit a count for that client id?) Or something else? >>>> >>>> Thanks, >>>> Justine >>>> >>> >>> Thanks >>> Omnia >>> >>> On Thu, Feb 2, 2023 at 4:44 PM Omnia Ibrahim <o.g.h.ibra...@gmail.com> >>> wrote: >>> >>>> Hi Luke and Justine, >>>> Are there any thoughts or updates on this? I would love to help with >>>> this as we are hitting this more frequently now. >>>> >>>> best, >>>> >>>> On Mon, Oct 31, 2022 at 6:15 PM Omnia Ibrahim <o.g.h.ibra...@gmail.com> >>>> wrote: >>>> >>>>> Hi Luke and Justine, >>>>> >>>>>> For (3), you said: >>>>>> > - I have some concerns about the impact of this option on the >>>>>> transactional >>>>>> producers, for example, what will happen to an ongoing transaction >>>>>> associated with an expired PID? Would this leave the transactions in a >>>>>> "hanging" state? >>>>>> >>>>>> - How will we notify the client that the transaction can't continue >>>>>> due to >>>>>> an expired PID? >>>>>> >>>>>> - If PID got marked as `expired` this will mean that >>>>>> `admin.DescribeProducers` will not list them which will make >>>>>> *`kafka-transactions.sh >>>>>> --list`* a bit tricky as we can't identify if there are transactions >>>>>> linked >>>>>> to this expired PID or not. The same concern applies to >>>>>> *`kafka-transactions.sh >>>>>> --find-hanging`*. >>>>>> >>>>>> --> Yes, you're right. Those are also concerns for this solution. >>>>>> Currently, there's no way to notify clients about the expiration. >>>>>> Also, the ongoing transactions will be hanging. For the admin cli, >>>>>> we've >>>>>> never thought about it. Good point. >>>>>> In summary, to adopt this solution, there are many issues needed to >>>>>> get >>>>>> fixed. >>>>>> >>>>> >>>>> Justin already clarified that if PID is attached to a transaction it >>>>> will not expire so identifying the transactions shouldn't be a concern >>>>> anymore. >>>>> The only concern here will be that this solution will not solve the >>>>> problem if the rouge producer is a transactional producer with hanging >>>>> transactions. >>>>> If anyone faced this situation they will need to abort the hanging >>>>> transactions manually and then the solution to expire a PID can then work. >>>>> >>>>> --> Yes, I mean KafkaPrinciple (sorry, I didn't make it clear) >>>>>> Yes, We were thinking about throttling by KafkaPrinciple. Client Id >>>>>> is also >>>>>> workable. >>>>>> It's just these 2 attributes are not required. >>>>>> That is, it's possible we take all clients as the same one: {default >>>>>> KafkaPrinciple + default clientID}, and apply throttling on it. >>>>>> Do you have any thoughts about it? >>>>>> Maybe skip throttling for {default KafkaPrinciple + default clientID} >>>>>> >>>>> >>>>> Throttling for default KafkaPrinciple and default ClientID is useful >>>>> when we need to have a hard limit on the whole cluster and whoever is >>>>> running the cluster doesn't knowclientsntIDs or if a KafkaPrinciple is >>>>> reused between different producer applications. >>>>> I usually find it helpful to have a way to apply throttling only on >>>>> the rough clients only once I identify them without punishing everyone on >>>>> the cluster. However, there are two problems with this >>>>> - There's no easy way at the moment to link PIDs to clientId or >>>>> KafkaPrinciple. This need to be addressed first. >>>>> - Is Justin's comment on the throttling, and the fact that will mean >>>>> we either block all requests or have to store the request in memory which >>>>> in both cases has side downs on the producer experince. >>>>> >>>>> I recently had another discussion with my team and it does seem like >>>>>> there >>>>>> should be a way to make it more clear to the clients what is going >>>>>> on. A >>>>>> lot of this protocol is implicit. I'm wondering if maybe there is a >>>>>> way to >>>>>> improve the story for newer clients. (Ie if we choose to expire based >>>>>> on a >>>>>> size limit, we should include a response indicating the ID has >>>>>> expired.) We >>>>>> also discussed ways to redefine the guarantees so that users who have >>>>>> stronger idempotency requirements can ensure them (over >>>>>> availability/memory >>>>>> concerns). Let me know if you have any ideas here. >>>>>> >>>>> >>>>> It may be easier to improve the experience for new clients. However, >>>>> if we improved only the new clients we may need a way to help teams who >>>>> run >>>>> Kafka with rough clients on old versions by at least giving them an easy >>>>> way to identify the clientId/ or KafkaPrinciple that generated these PIDs. >>>>> >>>>> For context, it's very tricky to even identify which clientId is >>>>> creating all these PIDs that caused OOM, which is a contributing part of >>>>> the issue at the moment. So maybe one option here could be adding a new >>>>> metric that tracks the number of generated PIDs per clientId. This will >>>>> help the team who runs the Kafka cluster to >>>>> - contact these rough clients and ask them to fix their clients or >>>>> upgrade to a new client if the new client version has a better experience. >>>>> - or if ended with a throttling solution this may help identify which >>>>> clientId needs to be throttled. >>>>> >>>>> Maybe we can start with a solution for identifying the rough clients >>>>> first and keep looking for a solution to limit them, what do you think? >>>>> >>>>> Thanks >>>>> >>>>> On Tue, Oct 18, 2022 at 5:24 PM Justine Olshan >>>>> <jols...@confluent.io.invalid> wrote: >>>>> >>>>>> Oops. I realized I just replied to Omnia 🤦♀️ >>>>>> >>>>>> Here was my response for the mailing thread: >>>>>> >>>>>> Hey Omnia, >>>>>> Sorry to hear this is a problem for you as well. :( >>>>>> > * I have some concerns about the impact of this option on the >>>>>> transactional producers, for example, what will happen to an ongoing >>>>>> transaction associated with an expired PID? Would this leave the >>>>>> transactions in a "hanging" state?* >>>>>> We currently check if a transaction is ongoing and do not expire the >>>>>> producer ID if it has an ongoing transaction. I suspect we will >>>>>> continue to >>>>>> do this with any solution we pick. >>>>>> >>>>>> My team members and I looked a bit into the throttling case and it >>>>>> can get >>>>>> a bit tricky since it means we need to throttle the produce request >>>>>> before >>>>>> it is processed. This means we either block all requests or have to >>>>>> store >>>>>> the request in memory (which is not great if we are trying to save >>>>>> memory). >>>>>> >>>>>> I recently had another discussion with my team and it does seem like >>>>>> there >>>>>> should be a way to make it more clear to the clients what is going >>>>>> on. A >>>>>> lot of this protocol is implicit. I'm wondering if maybe there is a >>>>>> way to >>>>>> improve the story for newer clients. (Ie if we choose to expire based >>>>>> on a >>>>>> size limit, we should include a response indicating the ID has >>>>>> expired.) We >>>>>> also discussed ways to redefine the guarantees so that users who have >>>>>> stronger idempotency requirements can ensure them (over >>>>>> availability/memory >>>>>> concerns). Let me know if you have any ideas here. >>>>>> >>>>>> Thanks again for commenting here, hopefully we can come to a good >>>>>> solution. >>>>>> >>>>>> On Tue, Oct 18, 2022 at 1:11 AM Luke Chen <show...@gmail.com> wrote: >>>>>> >>>>>> > Hi Omnia, >>>>>> > >>>>>> > Thanks for your reply. >>>>>> > >>>>>> > For (3), you said: >>>>>> > > - I have some concerns about the impact of this option on the >>>>>> > transactional >>>>>> > producers, for example, what will happen to an ongoing transaction >>>>>> > associated with an expired PID? Would this leave the transactions >>>>>> in a >>>>>> > "hanging" state? >>>>>> > >>>>>> > - How will we notify the client that the transaction can't continue >>>>>> due to >>>>>> > an expired PID? >>>>>> > >>>>>> > - If PID got marked as `expired` this will mean that >>>>>> > `admin.DescribeProducers` will not list them which will make >>>>>> > *`kafka-transactions.sh >>>>>> > --list`* a bit tricky as we can't identify if there are >>>>>> transactions linked >>>>>> > to this expired PID or not. The same concern applies to >>>>>> > *`kafka-transactions.sh >>>>>> > --find-hanging`*. >>>>>> > >>>>>> > --> Yes, you're right. Those are also concerns for this solution. >>>>>> > Currently, there's no way to notify clients about the expiration. >>>>>> > Also, the ongoing transactions will be hanging. For the admin cli, >>>>>> we've >>>>>> > never thought about it. Good point. >>>>>> > In summary, to adopt this solution, there are many issues needed to >>>>>> get >>>>>> > fixed. >>>>>> > >>>>>> > >>>>>> > For (5), you said: >>>>>> > > I am assuming you mean KafkaPrincipal here! If so is your concern >>>>>> here >>>>>> > that >>>>>> > those good clients that use the same principal as a rogue one will >>>>>> get >>>>>> > throttled? >>>>>> > >>>>>> > If this is the case, then I believe it should be okay as other >>>>>> throttling >>>>>> > in Kafka on *`/config/users/<user>`* has the same behaviour. >>>>>> > >>>>>> > >>>>>> > What about applying limit/throttling to >>>>>> > *`/config/users/<user>/clients/<client-id>` >>>>>> > *similar to what we have with client quota? This should reduce the >>>>>> concern >>>>>> > about throttling good clients, right? >>>>>> > >>>>>> > --> Yes, I mean KafkaPrinciple (sorry, I didn't make it clear) >>>>>> > Yes, We were thinking about throttling by KafkaPrinciple. Client Id >>>>>> is >>>>>> > also workable. >>>>>> > It's just these 2 attributes are not required. >>>>>> > That is, it's possible we take all clients as the same one: {default >>>>>> > KafkaPrinciple + default clientID}, and apply throttling on it. >>>>>> > Do you have any thoughts about it? >>>>>> > Maybe skip throttling for {default KafkaPrinciple + default >>>>>> clientID} ? >>>>>> > >>>>>> > Luke >>>>>> > >>>>>> > >>>>>> > >>>>>> > On Sat, Oct 15, 2022 at 2:33 AM Omnia Ibrahim < >>>>>> o.g.h.ibra...@gmail.com> >>>>>> > wrote: >>>>>> > >>>>>> >> Hi Luke & Justine, >>>>>> >> Thanks for looking into this issue, we have been experiencing this >>>>>> because >>>>>> >> of rouge clients as well. >>>>>> >> >>>>>> >> > 3. Having a limit to the number of active producer IDs (sort of >>>>>> like an >>>>>> >> LRU >>>>>> >> >cache) >>>>>> >> >-> The idea here is that if we hit a misconfigured client, we >>>>>> will expire >>>>>> >> >the older entries. The concern here is we have risks to lose >>>>>> idempotency >>>>>> >> >guarantees, and currently, we don't have a way to notify clients >>>>>> about >>>>>> >> >losing idempotency guarantees. Besides, the least recently used >>>>>> entries >>>>>> >> >got removed are not always from the "bad" clients. >>>>>> >> >>>>>> >> - I have some concerns about the impact of this option on the >>>>>> >> transactional >>>>>> >> producers, for example, what will happen to an ongoing transaction >>>>>> >> associated with an expired PID? Would this leave the transactions >>>>>> in a >>>>>> >> "hanging" state? >>>>>> >> >>>>>> >> - How will we notify the client that the transaction can't >>>>>> continue due to >>>>>> >> an expired PID? >>>>>> >> >>>>>> >> - If PID got marked as `expired` this will mean that >>>>>> >> `admin.DescribeProducers` will not list them which will make >>>>>> >> *`kafka-transactions.sh >>>>>> >> --list`* a bit tricky as we can't identify if there are >>>>>> transactions >>>>>> >> linked >>>>>> >> to this expired PID or not. The same concern applies to >>>>>> >> *`kafka-transactions.sh >>>>>> >> --find-hanging`*. >>>>>> >> >>>>>> >> >>>>>> >> >5. limit/throttling the producer id based on the principle >>>>>> >> >-> Although we can limit the impact to a certain principle with >>>>>> this >>>>>> >> idea, >>>>>> >> >same concern still exists as solution #1 #2. >>>>>> >> >>>>>> >> I am assuming you mean KafkaPrincipal here! If so is your concern >>>>>> here >>>>>> >> that >>>>>> >> those good clients that use the same principal as a rogue one will >>>>>> get >>>>>> >> throttled? >>>>>> >> >>>>>> >> If this is the case, then I believe it should be okay as other >>>>>> throttling >>>>>> >> in Kafka on *`/config/users/<user>`* has the same behaviour. >>>>>> >> >>>>>> >> >>>>>> >> What about applying limit/throttling to >>>>>> >> *`/config/users/<user>/clients/<client-id>` >>>>>> >> *similar to what we have with client quota? This should reduce the >>>>>> concern >>>>>> >> about throttling good clients, right? >>>>>> >> >>>>>> >> best, >>>>>> >> >>>>>> >> Omnia >>>>>> >> >>>>>> >> On Tue, Oct 11, 2022 at 4:18 AM Luke Chen <show...@gmail.com> >>>>>> wrote: >>>>>> >> >>>>>> >> > Bump this thread to see if there are any comments/thoughts. >>>>>> >> > Thanks. >>>>>> >> > >>>>>> >> > Luke >>>>>> >> > >>>>>> >> > On Mon, Sep 26, 2022 at 11:06 AM Luke Chen <show...@gmail.com> >>>>>> wrote: >>>>>> >> > >>>>>> >> > > Hi devs, >>>>>> >> > > >>>>>> >> > > As stated in the motivation section in KIP-854 >>>>>> >> > > < >>>>>> >> > >>>>>> >> >>>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-854+Separate+configuration+for+producer+ID+expiry >>>>>> >> > >: >>>>>> >> > > >>>>>> >> > > With idempotent producers becoming the default in Kafka, this >>>>>> means >>>>>> >> that >>>>>> >> > > unless otherwise specified, all new producers will be given >>>>>> producer >>>>>> >> IDs. >>>>>> >> > > Some (inefficient) applications may now create many >>>>>> non-transactional >>>>>> >> > > idempotent producers. Each of these producers will be assigned >>>>>> a >>>>>> >> producer >>>>>> >> > > ID and these IDs and their metadata are stored in the broker >>>>>> memory, >>>>>> >> > which >>>>>> >> > > might cause brokers out of memory. >>>>>> >> > > >>>>>> >> > > Justine (in cc.) and I and some other team members are working >>>>>> on the >>>>>> >> > > solutions for this issue. But none of them solves it completely >>>>>> >> without >>>>>> >> > > side effects. Among them, "availability" VS "idempotency >>>>>> guarantees" >>>>>> >> is >>>>>> >> > > what we can't decide which to sacrifice. Some of these >>>>>> solutions >>>>>> >> > sacrifice >>>>>> >> > > availability of produce (1,2,5) and others sacrifice >>>>>> idempotency >>>>>> >> > guarantees >>>>>> >> > > (3). It could be useful to know if people generally have a >>>>>> preference >>>>>> >> one >>>>>> >> > > way or the other. Or what other better solutions there might >>>>>> be. >>>>>> >> > > >>>>>> >> > > Here are the proposals we came up with: >>>>>> >> > > >>>>>> >> > > 1. Limit the total active producer ID allocation number. >>>>>> >> > > -> This is the simplest solution. But since the OOM issue is >>>>>> usually >>>>>> >> > > caused by a rogue or misconfigured client, and this solution >>>>>> might >>>>>> >> > "punish" >>>>>> >> > > the good client from sending messages. >>>>>> >> > > >>>>>> >> > > 2. Throttling the producer ID allocation rate >>>>>> >> > > -> Same concern as the solution #1. >>>>>> >> > > >>>>>> >> > > 3. Having a limit to the number of active producer IDs (sort >>>>>> of like >>>>>> >> an >>>>>> >> > > LRU cache) >>>>>> >> > > -> The idea here is that if we hit a misconfigured client, we >>>>>> will >>>>>> >> expire >>>>>> >> > > the older entries. The concern here is we have risks to lose >>>>>> >> idempotency >>>>>> >> > > guarantees, and currently, we don't have a way to notify >>>>>> clients about >>>>>> >> > > losing idempotency guarantees. Besides, the least recently >>>>>> used >>>>>> >> entries >>>>>> >> > > got removed are not always from the "bad" clients. >>>>>> >> > > >>>>>> >> > > 4. allow clients to "close" the producer ID usage >>>>>> >> > > -> We can provide a way for producer to "close" producerID >>>>>> usage. >>>>>> >> > > Currently, we only have a way to INIT_PRODUCER_ID requested to >>>>>> >> allocate >>>>>> >> > > one. After that, we'll keep the producer ID metadata in broker >>>>>> even if >>>>>> >> > the >>>>>> >> > > producer is "closed". Having a closed API (ex: >>>>>> END_PRODUCER_ID), we >>>>>> >> can >>>>>> >> > > remove the entry from broker side. In client side, we can send >>>>>> it when >>>>>> >> > > producer closing. The concern is, the old clients (including >>>>>> non-java >>>>>> >> > > clients) will still suffer from the OOM issue. >>>>>> >> > > >>>>>> >> > > 5. limit/throttling the producer id based on the principle >>>>>> >> > > -> Although we can limit the impact to a certain principle >>>>>> with this >>>>>> >> > idea, >>>>>> >> > > same concern still exists as solution #1 #2. >>>>>> >> > > >>>>>> >> > > Any thoughts/feedback are welcomed. >>>>>> >> > > >>>>>> >> > > Thank you. >>>>>> >> > > Luke >>>>>> >> > > >>>>>> >> > >>>>>> >> >>>>>> > >>>>>> >>>>> >>> On Wed, Feb 15, 2023 at 12:20 AM Justine Olshan <jols...@confluent.io> >>> wrote: >>> >>>> Hey Omnia, >>>> >>>> Thanks for the response. I think I understand your explanations here >>>> with respect to principal and clientId usage. >>>> >>>> For the throttling -- handleInitProducerIdRequest will allocate the ID >>>> to the producer, but we don't actually store it on the broker or increment >>>> our metric until the first produce request for that producer is sent (or >>>> sent again after previously expiring). Would you consider throttling the >>>> produce request instead? It may be hard to get any metrics from the >>>> transaction coordinator where the initProducerId request is handled. >>>> >>>> Justine >>>> >>>> On Tue, Feb 14, 2023 at 9:29 AM Omnia Ibrahim <o.g.h.ibra...@gmail.com> >>>> wrote: >>>> >>>>> Hey Justine, >>>>> > If I understand your message correctly, there are issues with >>>>> identifying the source of the rogue clients? So you propose to add a new >>>>> metric for that? >>>>> > And also proposing to throttle based on clientId as a potential >>>>> follow up? >>>>> I want to identify rogue clients by KafkaPrincipal (and/or clientId) >>>>> similarly to how we identify clients in Fetch/Produce/Request >>>>> QuotaManagers. Using KafkaPrincipal should give cluster admin the ability >>>>> to throttle later based on principal which is most likely to be a smaller >>>>> set than clientIds. My initial thought was to add a metrics that represent >>>>> how many InitProducerIDRequest are sent by KafkaPrincipal (and/or >>>>> clientId) >>>>> similar to Fetch/Produce QuotaManagers. >>>>> Then as a follow-up, we can throttle based on either KafkaPrinciple or >>>>> clientId (maybe default as well to align this with other QuotaManagers in >>>>> Kafka). >>>>> >>>>> >1. Does we rely on the client using the same ID? What if there are >>>>> many clients that all use different client IDs? >>>>> This is why I want to use the combination of KafkaPrincipal or >>>>> clientId similar to some other quotas we have in Kafka already. This will >>>>> be a similar risk to Fetch/Produce quota in Kafka which also relay on the >>>>> client to use the same clientId and KafkaPrincipal. >>>>> >>>>> >2. Are there places where high cardinality of this metric is a >>>>> concern? I can imagine many client IDs in the system. Would we treat this >>>>> as a rate metric (ie, when we get an init producer ID and return a new >>>>> producer ID we emit a count for that client id?) Or something else? >>>>> My initial thought here was to follow the steps of ClientQuotaManager >>>>> and ClientRequestQuotaManager and use a rate metric. However, I think we >>>>> can emit it either >>>>> >>>>> 1. when we return the new PID. However, I have concerns that we >>>>> may circle back to the previous concerns with OMM due to keeping track >>>>> of >>>>> ACTIVE PIDs per KafkaPrincipal(and/or) clientId in the future. Also >>>>> this >>>>> would be the first time Kafka throttle IDs for any client. >>>>> 2. or once we recieve initProducerIDRequest and throttle before >>>>> even hitting `handleInitProducerIdRequest`. Going this direction we may >>>>> need to throttle it within a different quota window than ` >>>>> quota.window.size.seconds ` as throttling INIT_PRODUCER_ID request >>>>> per second wouldn't be efficient for most cases. I personally think >>>>> this >>>>> direction is easier as it seems more consistent with the existing quota >>>>> implementation. Specially that Kafka has already the concept of >>>>> throttling >>>>> subset of requests (in ControllerMutationQuotaManager) but never had >>>>> any >>>>> concept of throttling active IDs for any client. >>>>> >>>>> >>>>> What do you think? >>>>> >>>>> Thanks >>>>> Omnia >>>>> >>>>> On Thu, Feb 2, 2023 at 5:39 PM Justine Olshan <jols...@confluent.io> >>>>> wrote: >>>>> >>>>>> Hey Omnia, >>>>>> Sorry for losing track of this. >>>>>> >>>>>> If I understand your message correctly, there are issues with >>>>>> identifying the source of the rogue clients? So you propose to add a new >>>>>> metric for that? >>>>>> And also proposing to throttle based on clientId as a potential >>>>>> follow up? >>>>>> >>>>>> I think both of these make sense. The only things I can think of for >>>>>> the metric are: >>>>>> 1. Does we rely on the client using the same ID? What if there are >>>>>> many clients that all use different client IDs? >>>>>> 2. Are there places where high cardinality of this metric is a >>>>>> concern? I can imagine many client IDs in the system. Would we treat this >>>>>> as a rate metric (ie, when we get an init producer ID and return a new >>>>>> producer ID we emit a count for that client id?) Or something else? >>>>>> >>>>>> Thanks, >>>>>> Justine >>>>>> >>>>>> On Thu, Feb 2, 2023 at 8:44 AM Omnia Ibrahim <o.g.h.ibra...@gmail.com> >>>>>> wrote: >>>>>> >>>>>>> Hi Luke and Justine, >>>>>>> Are there any thoughts or updates on this? I would love to help with >>>>>>> this as we are hitting this more frequently now. >>>>>>> >>>>>>> best, >>>>>>> >>>>>>> On Mon, Oct 31, 2022 at 6:15 PM Omnia Ibrahim < >>>>>>> o.g.h.ibra...@gmail.com> wrote: >>>>>>> >>>>>>>> Hi Luke and Justine, >>>>>>>> >>>>>>>>> For (3), you said: >>>>>>>>> > - I have some concerns about the impact of this option on the >>>>>>>>> transactional >>>>>>>>> producers, for example, what will happen to an ongoing transaction >>>>>>>>> associated with an expired PID? Would this leave the transactions >>>>>>>>> in a >>>>>>>>> "hanging" state? >>>>>>>>> >>>>>>>>> - How will we notify the client that the transaction can't >>>>>>>>> continue due to >>>>>>>>> an expired PID? >>>>>>>>> >>>>>>>>> - If PID got marked as `expired` this will mean that >>>>>>>>> `admin.DescribeProducers` will not list them which will make >>>>>>>>> *`kafka-transactions.sh >>>>>>>>> --list`* a bit tricky as we can't identify if there are >>>>>>>>> transactions linked >>>>>>>>> to this expired PID or not. The same concern applies to >>>>>>>>> *`kafka-transactions.sh >>>>>>>>> --find-hanging`*. >>>>>>>>> >>>>>>>>> --> Yes, you're right. Those are also concerns for this solution. >>>>>>>>> Currently, there's no way to notify clients about the expiration. >>>>>>>>> Also, the ongoing transactions will be hanging. For the admin cli, >>>>>>>>> we've >>>>>>>>> never thought about it. Good point. >>>>>>>>> In summary, to adopt this solution, there are many issues needed >>>>>>>>> to get >>>>>>>>> fixed. >>>>>>>>> >>>>>>>> >>>>>>>> Justin already clarified that if PID is attached to a transaction >>>>>>>> it will not expire so identifying the transactions shouldn't be a >>>>>>>> concern >>>>>>>> anymore. >>>>>>>> The only concern here will be that this solution will not solve the >>>>>>>> problem if the rouge producer is a transactional producer with hanging >>>>>>>> transactions. >>>>>>>> If anyone faced this situation they will need to abort the hanging >>>>>>>> transactions manually and then the solution to expire a PID can then >>>>>>>> work. >>>>>>>> >>>>>>>> --> Yes, I mean KafkaPrinciple (sorry, I didn't make it clear) >>>>>>>>> Yes, We were thinking about throttling by KafkaPrinciple. Client >>>>>>>>> Id is also >>>>>>>>> workable. >>>>>>>>> It's just these 2 attributes are not required. >>>>>>>>> That is, it's possible we take all clients as the same one: >>>>>>>>> {default >>>>>>>>> KafkaPrinciple + default clientID}, and apply throttling on it. >>>>>>>>> Do you have any thoughts about it? >>>>>>>>> Maybe skip throttling for {default KafkaPrinciple + default >>>>>>>>> clientID} >>>>>>>>> >>>>>>>> >>>>>>>> Throttling for default KafkaPrinciple and default ClientID is >>>>>>>> useful when we need to have a hard limit on the whole cluster and >>>>>>>> whoever >>>>>>>> is running the cluster doesn't knowclientsntIDs or if a KafkaPrinciple >>>>>>>> is >>>>>>>> reused between different producer applications. >>>>>>>> I usually find it helpful to have a way to apply throttling only on >>>>>>>> the rough clients only once I identify them without punishing everyone >>>>>>>> on >>>>>>>> the cluster. However, there are two problems with this >>>>>>>> - There's no easy way at the moment to link PIDs to clientId or >>>>>>>> KafkaPrinciple. This need to be addressed first. >>>>>>>> - Is Justin's comment on the throttling, and the fact that will >>>>>>>> mean we either block all requests or have to store the request in >>>>>>>> memory >>>>>>>> which in both cases has side downs on the producer experince. >>>>>>>> >>>>>>>> I recently had another discussion with my team and it does seem >>>>>>>>> like there >>>>>>>>> should be a way to make it more clear to the clients what is going >>>>>>>>> on. A >>>>>>>>> lot of this protocol is implicit. I'm wondering if maybe there is >>>>>>>>> a way to >>>>>>>>> improve the story for newer clients. (Ie if we choose to expire >>>>>>>>> based on a >>>>>>>>> size limit, we should include a response indicating the ID has >>>>>>>>> expired.) We >>>>>>>>> also discussed ways to redefine the guarantees so that users who >>>>>>>>> have >>>>>>>>> stronger idempotency requirements can ensure them (over >>>>>>>>> availability/memory >>>>>>>>> concerns). Let me know if you have any ideas here. >>>>>>>>> >>>>>>>> >>>>>>>> It may be easier to improve the experience for new clients. >>>>>>>> However, if we improved only the new clients we may need a way to help >>>>>>>> teams who run Kafka with rough clients on old versions by at least >>>>>>>> giving >>>>>>>> them an easy way to identify the clientId/ or KafkaPrinciple that >>>>>>>> generated >>>>>>>> these PIDs. >>>>>>>> >>>>>>>> For context, it's very tricky to even identify which clientId is >>>>>>>> creating all these PIDs that caused OOM, which is a contributing part >>>>>>>> of >>>>>>>> the issue at the moment. So maybe one option here could be adding a new >>>>>>>> metric that tracks the number of generated PIDs per clientId. This will >>>>>>>> help the team who runs the Kafka cluster to >>>>>>>> - contact these rough clients and ask them to fix their clients or >>>>>>>> upgrade to a new client if the new client version has a better >>>>>>>> experience. >>>>>>>> - or if ended with a throttling solution this may help identify >>>>>>>> which clientId needs to be throttled. >>>>>>>> >>>>>>>> Maybe we can start with a solution for identifying the rough >>>>>>>> clients first and keep looking for a solution to limit them, what do >>>>>>>> you >>>>>>>> think? >>>>>>>> >>>>>>>> Thanks >>>>>>>> >>>>>>>> On Tue, Oct 18, 2022 at 5:24 PM Justine Olshan >>>>>>>> <jols...@confluent.io.invalid> wrote: >>>>>>>> >>>>>>>>> Oops. I realized I just replied to Omnia 🤦♀️ >>>>>>>>> >>>>>>>>> Here was my response for the mailing thread: >>>>>>>>> >>>>>>>>> Hey Omnia, >>>>>>>>> Sorry to hear this is a problem for you as well. :( >>>>>>>>> > * I have some concerns about the impact of this option on the >>>>>>>>> transactional producers, for example, what will happen to an >>>>>>>>> ongoing >>>>>>>>> transaction associated with an expired PID? Would this leave the >>>>>>>>> transactions in a "hanging" state?* >>>>>>>>> We currently check if a transaction is ongoing and do not expire >>>>>>>>> the >>>>>>>>> producer ID if it has an ongoing transaction. I suspect we will >>>>>>>>> continue to >>>>>>>>> do this with any solution we pick. >>>>>>>>> >>>>>>>>> My team members and I looked a bit into the throttling case and it >>>>>>>>> can get >>>>>>>>> a bit tricky since it means we need to throttle the produce >>>>>>>>> request before >>>>>>>>> it is processed. This means we either block all requests or have >>>>>>>>> to store >>>>>>>>> the request in memory (which is not great if we are trying to save >>>>>>>>> memory). >>>>>>>>> >>>>>>>>> I recently had another discussion with my team and it does seem >>>>>>>>> like there >>>>>>>>> should be a way to make it more clear to the clients what is going >>>>>>>>> on. A >>>>>>>>> lot of this protocol is implicit. I'm wondering if maybe there is >>>>>>>>> a way to >>>>>>>>> improve the story for newer clients. (Ie if we choose to expire >>>>>>>>> based on a >>>>>>>>> size limit, we should include a response indicating the ID has >>>>>>>>> expired.) We >>>>>>>>> also discussed ways to redefine the guarantees so that users who >>>>>>>>> have >>>>>>>>> stronger idempotency requirements can ensure them (over >>>>>>>>> availability/memory >>>>>>>>> concerns). Let me know if you have any ideas here. >>>>>>>>> >>>>>>>>> Thanks again for commenting here, hopefully we can come to a good >>>>>>>>> solution. >>>>>>>>> >>>>>>>>> On Tue, Oct 18, 2022 at 1:11 AM Luke Chen <show...@gmail.com> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>> > Hi Omnia, >>>>>>>>> > >>>>>>>>> > Thanks for your reply. >>>>>>>>> > >>>>>>>>> > For (3), you said: >>>>>>>>> > > - I have some concerns about the impact of this option on the >>>>>>>>> > transactional >>>>>>>>> > producers, for example, what will happen to an ongoing >>>>>>>>> transaction >>>>>>>>> > associated with an expired PID? Would this leave the >>>>>>>>> transactions in a >>>>>>>>> > "hanging" state? >>>>>>>>> > >>>>>>>>> > - How will we notify the client that the transaction can't >>>>>>>>> continue due to >>>>>>>>> > an expired PID? >>>>>>>>> > >>>>>>>>> > - If PID got marked as `expired` this will mean that >>>>>>>>> > `admin.DescribeProducers` will not list them which will make >>>>>>>>> > *`kafka-transactions.sh >>>>>>>>> > --list`* a bit tricky as we can't identify if there are >>>>>>>>> transactions linked >>>>>>>>> > to this expired PID or not. The same concern applies to >>>>>>>>> > *`kafka-transactions.sh >>>>>>>>> > --find-hanging`*. >>>>>>>>> > >>>>>>>>> > --> Yes, you're right. Those are also concerns for this solution. >>>>>>>>> > Currently, there's no way to notify clients about the expiration. >>>>>>>>> > Also, the ongoing transactions will be hanging. For the admin >>>>>>>>> cli, we've >>>>>>>>> > never thought about it. Good point. >>>>>>>>> > In summary, to adopt this solution, there are many issues needed >>>>>>>>> to get >>>>>>>>> > fixed. >>>>>>>>> > >>>>>>>>> > >>>>>>>>> > For (5), you said: >>>>>>>>> > > I am assuming you mean KafkaPrincipal here! If so is your >>>>>>>>> concern here >>>>>>>>> > that >>>>>>>>> > those good clients that use the same principal as a rogue one >>>>>>>>> will get >>>>>>>>> > throttled? >>>>>>>>> > >>>>>>>>> > If this is the case, then I believe it should be okay as other >>>>>>>>> throttling >>>>>>>>> > in Kafka on *`/config/users/<user>`* has the same behaviour. >>>>>>>>> > >>>>>>>>> > >>>>>>>>> > What about applying limit/throttling to >>>>>>>>> > *`/config/users/<user>/clients/<client-id>` >>>>>>>>> > *similar to what we have with client quota? This should reduce >>>>>>>>> the concern >>>>>>>>> > about throttling good clients, right? >>>>>>>>> > >>>>>>>>> > --> Yes, I mean KafkaPrinciple (sorry, I didn't make it clear) >>>>>>>>> > Yes, We were thinking about throttling by KafkaPrinciple. Client >>>>>>>>> Id is >>>>>>>>> > also workable. >>>>>>>>> > It's just these 2 attributes are not required. >>>>>>>>> > That is, it's possible we take all clients as the same one: >>>>>>>>> {default >>>>>>>>> > KafkaPrinciple + default clientID}, and apply throttling on it. >>>>>>>>> > Do you have any thoughts about it? >>>>>>>>> > Maybe skip throttling for {default KafkaPrinciple + default >>>>>>>>> clientID} ? >>>>>>>>> > >>>>>>>>> > Luke >>>>>>>>> > >>>>>>>>> > >>>>>>>>> > >>>>>>>>> > On Sat, Oct 15, 2022 at 2:33 AM Omnia Ibrahim < >>>>>>>>> o.g.h.ibra...@gmail.com> >>>>>>>>> > wrote: >>>>>>>>> > >>>>>>>>> >> Hi Luke & Justine, >>>>>>>>> >> Thanks for looking into this issue, we have been experiencing >>>>>>>>> this because >>>>>>>>> >> of rouge clients as well. >>>>>>>>> >> >>>>>>>>> >> > 3. Having a limit to the number of active producer IDs (sort >>>>>>>>> of like an >>>>>>>>> >> LRU >>>>>>>>> >> >cache) >>>>>>>>> >> >-> The idea here is that if we hit a misconfigured client, we >>>>>>>>> will expire >>>>>>>>> >> >the older entries. The concern here is we have risks to lose >>>>>>>>> idempotency >>>>>>>>> >> >guarantees, and currently, we don't have a way to notify >>>>>>>>> clients about >>>>>>>>> >> >losing idempotency guarantees. Besides, the least recently >>>>>>>>> used entries >>>>>>>>> >> >got removed are not always from the "bad" clients. >>>>>>>>> >> >>>>>>>>> >> - I have some concerns about the impact of this option on the >>>>>>>>> >> transactional >>>>>>>>> >> producers, for example, what will happen to an ongoing >>>>>>>>> transaction >>>>>>>>> >> associated with an expired PID? Would this leave the >>>>>>>>> transactions in a >>>>>>>>> >> "hanging" state? >>>>>>>>> >> >>>>>>>>> >> - How will we notify the client that the transaction can't >>>>>>>>> continue due to >>>>>>>>> >> an expired PID? >>>>>>>>> >> >>>>>>>>> >> - If PID got marked as `expired` this will mean that >>>>>>>>> >> `admin.DescribeProducers` will not list them which will make >>>>>>>>> >> *`kafka-transactions.sh >>>>>>>>> >> --list`* a bit tricky as we can't identify if there are >>>>>>>>> transactions >>>>>>>>> >> linked >>>>>>>>> >> to this expired PID or not. The same concern applies to >>>>>>>>> >> *`kafka-transactions.sh >>>>>>>>> >> --find-hanging`*. >>>>>>>>> >> >>>>>>>>> >> >>>>>>>>> >> >5. limit/throttling the producer id based on the principle >>>>>>>>> >> >-> Although we can limit the impact to a certain principle >>>>>>>>> with this >>>>>>>>> >> idea, >>>>>>>>> >> >same concern still exists as solution #1 #2. >>>>>>>>> >> >>>>>>>>> >> I am assuming you mean KafkaPrincipal here! If so is your >>>>>>>>> concern here >>>>>>>>> >> that >>>>>>>>> >> those good clients that use the same principal as a rogue one >>>>>>>>> will get >>>>>>>>> >> throttled? >>>>>>>>> >> >>>>>>>>> >> If this is the case, then I believe it should be okay as other >>>>>>>>> throttling >>>>>>>>> >> in Kafka on *`/config/users/<user>`* has the same behaviour. >>>>>>>>> >> >>>>>>>>> >> >>>>>>>>> >> What about applying limit/throttling to >>>>>>>>> >> *`/config/users/<user>/clients/<client-id>` >>>>>>>>> >> *similar to what we have with client quota? This should reduce >>>>>>>>> the concern >>>>>>>>> >> about throttling good clients, right? >>>>>>>>> >> >>>>>>>>> >> best, >>>>>>>>> >> >>>>>>>>> >> Omnia >>>>>>>>> >> >>>>>>>>> >> On Tue, Oct 11, 2022 at 4:18 AM Luke Chen <show...@gmail.com> >>>>>>>>> wrote: >>>>>>>>> >> >>>>>>>>> >> > Bump this thread to see if there are any comments/thoughts. >>>>>>>>> >> > Thanks. >>>>>>>>> >> > >>>>>>>>> >> > Luke >>>>>>>>> >> > >>>>>>>>> >> > On Mon, Sep 26, 2022 at 11:06 AM Luke Chen <show...@gmail.com> >>>>>>>>> wrote: >>>>>>>>> >> > >>>>>>>>> >> > > Hi devs, >>>>>>>>> >> > > >>>>>>>>> >> > > As stated in the motivation section in KIP-854 >>>>>>>>> >> > > < >>>>>>>>> >> > >>>>>>>>> >> >>>>>>>>> https://cwiki.apache.org/confluence/display/KAFKA/KIP-854+Separate+configuration+for+producer+ID+expiry >>>>>>>>> >> > >: >>>>>>>>> >> > > >>>>>>>>> >> > > With idempotent producers becoming the default in Kafka, >>>>>>>>> this means >>>>>>>>> >> that >>>>>>>>> >> > > unless otherwise specified, all new producers will be given >>>>>>>>> producer >>>>>>>>> >> IDs. >>>>>>>>> >> > > Some (inefficient) applications may now create many >>>>>>>>> non-transactional >>>>>>>>> >> > > idempotent producers. Each of these producers will be >>>>>>>>> assigned a >>>>>>>>> >> producer >>>>>>>>> >> > > ID and these IDs and their metadata are stored in the >>>>>>>>> broker memory, >>>>>>>>> >> > which >>>>>>>>> >> > > might cause brokers out of memory. >>>>>>>>> >> > > >>>>>>>>> >> > > Justine (in cc.) and I and some other team members are >>>>>>>>> working on the >>>>>>>>> >> > > solutions for this issue. But none of them solves it >>>>>>>>> completely >>>>>>>>> >> without >>>>>>>>> >> > > side effects. Among them, "availability" VS "idempotency >>>>>>>>> guarantees" >>>>>>>>> >> is >>>>>>>>> >> > > what we can't decide which to sacrifice. Some of these >>>>>>>>> solutions >>>>>>>>> >> > sacrifice >>>>>>>>> >> > > availability of produce (1,2,5) and others sacrifice >>>>>>>>> idempotency >>>>>>>>> >> > guarantees >>>>>>>>> >> > > (3). It could be useful to know if people generally have a >>>>>>>>> preference >>>>>>>>> >> one >>>>>>>>> >> > > way or the other. Or what other better solutions there >>>>>>>>> might be. >>>>>>>>> >> > > >>>>>>>>> >> > > Here are the proposals we came up with: >>>>>>>>> >> > > >>>>>>>>> >> > > 1. Limit the total active producer ID allocation number. >>>>>>>>> >> > > -> This is the simplest solution. But since the OOM issue >>>>>>>>> is usually >>>>>>>>> >> > > caused by a rogue or misconfigured client, and this >>>>>>>>> solution might >>>>>>>>> >> > "punish" >>>>>>>>> >> > > the good client from sending messages. >>>>>>>>> >> > > >>>>>>>>> >> > > 2. Throttling the producer ID allocation rate >>>>>>>>> >> > > -> Same concern as the solution #1. >>>>>>>>> >> > > >>>>>>>>> >> > > 3. Having a limit to the number of active producer IDs >>>>>>>>> (sort of like >>>>>>>>> >> an >>>>>>>>> >> > > LRU cache) >>>>>>>>> >> > > -> The idea here is that if we hit a misconfigured client, >>>>>>>>> we will >>>>>>>>> >> expire >>>>>>>>> >> > > the older entries. The concern here is we have risks to lose >>>>>>>>> >> idempotency >>>>>>>>> >> > > guarantees, and currently, we don't have a way to notify >>>>>>>>> clients about >>>>>>>>> >> > > losing idempotency guarantees. Besides, the least recently >>>>>>>>> used >>>>>>>>> >> entries >>>>>>>>> >> > > got removed are not always from the "bad" clients. >>>>>>>>> >> > > >>>>>>>>> >> > > 4. allow clients to "close" the producer ID usage >>>>>>>>> >> > > -> We can provide a way for producer to "close" producerID >>>>>>>>> usage. >>>>>>>>> >> > > Currently, we only have a way to INIT_PRODUCER_ID requested >>>>>>>>> to >>>>>>>>> >> allocate >>>>>>>>> >> > > one. After that, we'll keep the producer ID metadata in >>>>>>>>> broker even if >>>>>>>>> >> > the >>>>>>>>> >> > > producer is "closed". Having a closed API (ex: >>>>>>>>> END_PRODUCER_ID), we >>>>>>>>> >> can >>>>>>>>> >> > > remove the entry from broker side. In client side, we can >>>>>>>>> send it when >>>>>>>>> >> > > producer closing. The concern is, the old clients >>>>>>>>> (including non-java >>>>>>>>> >> > > clients) will still suffer from the OOM issue. >>>>>>>>> >> > > >>>>>>>>> >> > > 5. limit/throttling the producer id based on the principle >>>>>>>>> >> > > -> Although we can limit the impact to a certain principle >>>>>>>>> with this >>>>>>>>> >> > idea, >>>>>>>>> >> > > same concern still exists as solution #1 #2. >>>>>>>>> >> > > >>>>>>>>> >> > > Any thoughts/feedback are welcomed. >>>>>>>>> >> > > >>>>>>>>> >> > > Thank you. >>>>>>>>> >> > > Luke >>>>>>>>> >> > > >>>>>>>>> >> > >>>>>>>>> >> >>>>>>>>> > >>>>>>>>> >>>>>>>>
