Thanks Andrew. Done :) @Chris: I changed the config parameter type from boolean to integer, which defines the timeout for retrying. I thought reusing `max.block.ms` was not reasonable as you mentioned.
So if the KIP looks good, let 's skip to the good part ;-) VOTING :) Bests, Alieh On Tue, May 14, 2024 at 12:26 PM Andrew Schofield <andrew_schofi...@live.com> wrote: > Hi Alieh, > Just one final comment. > > [AJS5] Existing classes use Retriable, not Retryable. For example: > > https://kafka.apache.org/21/javadoc/org/apache/kafka/common/errors/RetriableException.html > > I suggest RetriableResponse and NonRetriableResponse. > > Thanks, > Andrew > > > On 13 May 2024, at 23:17, Alieh Saeedi <asae...@confluent.io.INVALID> > wrote: > > > > Hi all, > > > > > > Thanks for all the valid points you listed. > > > > > > KIP updates and addressing concerns: > > > > > > 1) The KIP now suggests two Response types: `RetryableResponse` and > > `NonRetryableResponse` > > > > > > 2) `custom.exception.handler` is changed to > `custom.exception.handler.class` > > > > > > 3) The KIP clarifies that `In the case of an implemented handler for the > > specified exception, the handler takes precedence.` > > > > > > 4) There is now a `default` implementation for both handle() methods. > > > > > > 5) @Chris: for `UnknownTopicOrPartition`, the default is already > retrying > > for 60s. (In fact, the default value of `max.block.ms`). If the handler > > instructs to FAIL or SWALLOW, there will be no retry, and if the handler > > instructs to RETRY, that will be the default behavior, which follows the > > values in already existing config parameters such as `max.block.ms`. > Does > > that make sense? > > > > > > Hope the changes and explanations are convincing :) > > > > > > Cheers, > > > > Alieh > > > > On Mon, May 13, 2024 at 6:40 PM Justine Olshan > <jols...@confluent.io.invalid> > > wrote: > > > >> Oh I see. The type isn't the error type but a newly defined type for the > >> response. Makes sense and works for me. > >> > >> Justine > >> > >> On Mon, May 13, 2024 at 9:13 AM Chris Egerton <fearthecel...@gmail.com> > >> wrote: > >> > >>> If we have dedicated methods for each kind of exception > >>> (handleRecordTooLarge, handleUnknownTopicOrPartition, etc.), doesn't > that > >>> provide sufficient constraint? I'm not suggesting we eliminate these > >>> methods, just that we change their return types to something more > >> flexible. > >>> > >>> On Mon, May 13, 2024, 12:07 Justine Olshan > <jols...@confluent.io.invalid > >>> > >>> wrote: > >>> > >>>> I'm not sure I agree with the Retriable and NonRetriableResponse > >> comment. > >>>> This doesn't limit the blast radius or enforce certain errors are > used. > >>>> I think we might disagree on how controlled these interfaces can be... > >>>> > >>>> Justine > >>>> > >>>> On Mon, May 13, 2024 at 8:40 AM Chris Egerton <chr...@aiven.io.invalid > >>> > >>>> wrote: > >>>> > >>>>> Hi Alieh, > >>>>> > >>>>> Thanks for the updates! I just have a few more thoughts: > >>>>> > >>>>> - I don't think a boolean property is sufficient to dictate retries > >> for > >>>>> unknown topic partitions, though. These errors can occur if a topic > >> has > >>>>> just been created, which can occur if, for example, automatic topic > >>>>> creation is enabled for a multi-task connector. This is why I > >> proposed > >>> a > >>>>> timeout instead of a boolean (and see my previous email for why > >>> reducing > >>>>> max.block.ms for a producer is not a viable alternative). If it > >> helps, > >>>> one > >>>>> way to reproduce this yourself is to add the line > >>>>> `fooProps.put(TASKS_MAX_CONFIG, "10");` to the integration test here: > >>>>> > >>>>> > >>>> > >>> > >> > https://github.com/apache/kafka/blob/5439914c32fa00d634efa7219699f1bc21add839/connect/runtime/src/test/java/org/apache/kafka/connect/integration/SourceConnectorsIntegrationTest.java#L134 > >>>>> and then check the logs afterward for messages like "Error while > >>> fetching > >>>>> metadata with correlation id <n> : > >>>> {foo-topic=UNKNOWN_TOPIC_OR_PARTITION}". > >>>>> > >>>>> - I also don't think we need custom XxxResponse enums for every > >>> possible > >>>>> method; it seems like this will lead to a lot of duplication and > >>>> cognitive > >>>>> overhead if we want to expand the error handler in the future. > >>> Something > >>>>> more flexible like RetriableResponse and NonRetriableResponse could > >>>>> suffice. > >>>>> > >>>>> - Finally, the KIP still doesn't state how the handler will or won't > >>> take > >>>>> precedence over existing retry properties. If I set `retries` or ` > >>>>> delivery.timeout.ms` or `max.block.ms` to low values, will that > >> cause > >>>>> retries to cease even if my custom handler would otherwise keep > >>> returning > >>>>> RETRY for an error? > >>>>> > >>>>> Cheers, > >>>>> > >>>>> Chris > >>>>> > >>>>> On Mon, May 13, 2024 at 11:02 AM Andrew Schofield < > >>>>> andrew_schofi...@live.com> > >>>>> wrote: > >>>>> > >>>>>> Hi Alieh, > >>>>>> Just a few more comments on the KIP. It is looking much less risky > >>> now > >>>>> the > >>>>>> scope > >>>>>> is tighter. > >>>>>> > >>>>>> [AJS1] It would be nice to have default implementations of the > >> handle > >>>>>> methods > >>>>>> so an implementor would not need to implement both themselves. > >>>>>> > >>>>>> [AJS2] Producer configurations which are class names usually end in > >>>>>> “.class”. > >>>>>> I suggest “custom.exception.handler.class”. > >>>>>> > >>>>>> [AJS3] If I implemented a handler, and I set a non-default value > >> for > >>>> one > >>>>>> of the > >>>>>> new configuations, what happens? I would expect that the handler > >>> takes > >>>>>> precedence. I wasn’t quite clear what “the control will follow the > >>>>> handler > >>>>>> instructions” meant. > >>>>>> > >>>>>> [AJS4] Because you now have an enum for the > >>>>>> RecordTooLargeExceptionResponse, > >>>>>> I don’t think you need to state in the comment for > >>>>>> ProducerExceptionHandler that > >>>>>> RETRY will be interpreted as FAIL. > >>>>>> > >>>>>> Thanks, > >>>>>> Andrew > >>>>>> > >>>>>>> On 13 May 2024, at 14:53, Alieh Saeedi > >>> <asae...@confluent.io.INVALID > >>>>> > >>>>>> wrote: > >>>>>>> > >>>>>>> Hi all, > >>>>>>> > >>>>>>> > >>>>>>> Thanks for the very interesting discussion during my PTO. > >>>>>>> > >>>>>>> > >>>>>>> KIP updates and addressing concerns: > >>>>>>> > >>>>>>> > >>>>>>> 1) Two handle() methods are defined in ProducerExceptionHandler > >> for > >>>> the > >>>>>> two > >>>>>>> exceptions with different input parameters so that we have > >>>>>>> handle(RecordTooLargeException e, ProducerRecord record) and > >>>>>>> handle(UnknownTopicOrPartitionException e, ProducerRecord record) > >>>>>>> > >>>>>>> > >>>>>>> 2) The ProducerExceptionHandler extends `Closable` as well. > >>>>>>> > >>>>>>> > >>>>>>> 3) The KIP suggests having two more configuration parameters with > >>>>> boolean > >>>>>>> values: > >>>>>>> > >>>>>>> - `drop.invalid.large.records` with a default value of `false` > >> for > >>>>>>> swallowing too large records. > >>>>>>> > >>>>>>> - `retry.unknown.topic.partition` with a default value of `true` > >>> that > >>>>>>> performs RETRY for `max.block.ms` ms, encountering the > >>>>>>> UnknownTopicOrPartitionException. > >>>>>>> > >>>>>>> > >>>>>>> Hope the main concerns are addressed so that we can go forward > >> with > >>>>>> voting. > >>>>>>> > >>>>>>> > >>>>>>> Cheers, > >>>>>>> > >>>>>>> Alieh > >>>>>>> > >>>>>>> On Thu, May 9, 2024 at 11:25 PM Artem Livshits > >>>>>>> <alivsh...@confluent.io.invalid> wrote: > >>>>>>> > >>>>>>>> Hi Mathias, > >>>>>>>> > >>>>>>>>> [AL1] While I see the point, I would think having a different > >>>>> callback > >>>>>>>> for every exception might not really be elegant? > >>>>>>>> > >>>>>>>> I'm not sure how to assess the level of elegance of the > >> proposal, > >>>> but > >>>>> I > >>>>>> can > >>>>>>>> comment on the technical characteristics: > >>>>>>>> > >>>>>>>> 1. Having specific interfaces that codify the logic that is > >>>> currently > >>>>>>>> prescribed in the comments reduce the chance of making a > >> mistake. > >>>>>>>> Commments may get ignored, misuderstood or etc. but if the > >>> contract > >>>> is > >>>>>>>> codified, the compilier will help to enforce the contract. > >>>>>>>> 2. Given that the logic is trickier than it seems (the > >>>>> record-too-large > >>>>>> is > >>>>>>>> an example that can easily confuse someone who's not intimately > >>>>> familiar > >>>>>>>> with the nuances of the batching logic), having a little more > >>> hoops > >>>> to > >>>>>> jump > >>>>>>>> would give a greater chance that whoever tries to add a new > >> cases > >>>>> pauses > >>>>>>>> and thinks a bit more. > >>>>>>>> 3. As Justine pointed out, having different method will be a > >>> forcing > >>>>>>>> function to go through a KIP rather than smuggle new cases > >> through > >>>>>>>> implementation. > >>>>>>>> 4. Sort of a consequence of the previous 3 -- all those things > >>>> reduce > >>>>>> the > >>>>>>>> chance of someone writing the code that works with 2 errors and > >>> then > >>>>>> when > >>>>>>>> more errors are added in the future will suddenly incorrectly > >>> ignore > >>>>> new > >>>>>>>> errors (the example I gave in the previous email). > >>>>>>>> > >>>>>>>>> [AL2 cont.] Similar to AL1, I see such a handler to some extend > >>> as > >>>>>>>> business logic. If a user puts a bad filter condition in their > >> KS > >>>> app, > >>>>>> and > >>>>>>>> drops messages > >>>>>>>> > >>>>>>>> I agree that there is always a chance to get a bug and lose > >>>> messages, > >>>>>> but > >>>>>>>> there are generally separation of concerns that has different > >> risk > >>>>>> profile: > >>>>>>>> the filtering logic may be more rigorously tested and rarely > >>> changed > >>>>>> (say > >>>>>>>> an application developer does it), but setting the topics to > >>> produce > >>>>>> may be > >>>>>>>> done via configuration (e.g. a user of the application does it) > >>> and > >>>>> it's > >>>>>>>> generally an expectation that users would get an error when > >>>>>> configuration > >>>>>>>> is incorrect. > >>>>>>>> > >>>>>>>> What could be worse is that UnknownTopicOrPartitionException can > >>> be > >>>> an > >>>>>>>> intermittent error, i.e. with a generally correct configuration, > >>>> there > >>>>>>>> could be metadata propagation problem on the cluster and then a > >>>> random > >>>>>> set > >>>>>>>> of records could get lost. > >>>>>>>> > >>>>>>>>> [AL3] Maybe I misunderstand what you are saying, but to me, > >>>> checking > >>>>>> the > >>>>>>>> size of the record upfront is exactly what the KIP proposes? No? > >>>>>>>> > >>>>>>>> It achieves the same result but solves it differently, my > >>> proposal: > >>>>>>>> > >>>>>>>> 1. Application checks the validity of a record (maybe via a new > >>>>>>>> validateRecord method) before producing it, and can just exclude > >>> it > >>>> or > >>>>>>>> return an error to the user. > >>>>>>>> 2. Application produces the record -- at this point there are no > >>>>> records > >>>>>>>> that could return record too large, they were either skipped at > >>>> step 1 > >>>>>> or > >>>>>>>> we didn't get here because step 1 failed. > >>>>>>>> > >>>>>>>> Vs. KIP's proposal > >>>>>>>> > >>>>>>>> 1. Application produces the record. > >>>>>>>> 2. Application gets a callback. > >>>>>>>> 3. Application returns the action on how to proceed. > >>>>>>>> > >>>>>>>> The advantage of the former is the clarity of semantics -- the > >>>> record > >>>>> is > >>>>>>>> invalid (property of the record, not a function of server state > >> or > >>>>>> server > >>>>>>>> configuration) and we can clearly know that it is the record > >> that > >>> is > >>>>> bad > >>>>>>>> and can never succeed. > >>>>>>>> > >>>>>>>> The KIP-proposed way actually has a very tricky point: it > >> actually > >>>>>> handles > >>>>>>>> a subset of record-too-large exceptions. The broker can return > >>>>>>>> record-too-large and reject the whole batch (but we don't want > >> to > >>>>> ignore > >>>>>>>> those because then we can skip random records that just happened > >>> to > >>>> be > >>>>>> in > >>>>>>>> the same batch), in some sense we use the same error for 2 > >>> different > >>>>>>>> conditions and understanding that requires pretty deep > >>> understanding > >>>>> of > >>>>>>>> Kafka internals. > >>>>>>>> > >>>>>>>> -Artem > >>>>>>>> > >>>>>>>> > >>>>>>>> On Wed, May 8, 2024 at 9:47 AM Justine Olshan > >>>>>> <jols...@confluent.io.invalid > >>>>>>>>> > >>>>>>>> wrote: > >>>>>>>> > >>>>>>>>> My concern with respect to it being fragile: the code that > >>> ensures > >>>>> the > >>>>>>>>> error type is internal to the producer. Someone may see it and > >>>> say, I > >>>>>>>> want > >>>>>>>>> to add such and such error. This looks like internal code, so I > >>>> don't > >>>>>>>> need > >>>>>>>>> a KIP, and then they can change it to whatever they want > >> thinking > >>>> it > >>>>> is > >>>>>>>>> within the typical kafka improvement protocol. > >>>>>>>>> > >>>>>>>>> Relying on an internal change to enforce an external API is > >>> fragile > >>>>> in > >>>>>> my > >>>>>>>>> opinion. That's why I sort of agreed with Artem with enforcing > >>> the > >>>>>> error > >>>>>>>> in > >>>>>>>>> the method signature -- part of the public API. > >>>>>>>>> > >>>>>>>>> Chris's comments on requiring more information to handler again > >>>> makes > >>>>>> me > >>>>>>>>> wonder if we are solving a problem of lack of information at > >> the > >>>>>>>>> application level with a more powerful solution than we need. > >>> (Ie, > >>>> if > >>>>>> we > >>>>>>>>> had more information, could the application close and restart > >> the > >>>>>>>>> transaction rather than having to drop records) But I am happy > >> to > >>>>>>>>> compromise with a handler that we can agree is sufficiently > >>>>> controlled > >>>>>>>> and > >>>>>>>>> documented. > >>>>>>>>> > >>>>>>>>> Justine > >>>>>>>>> > >>>>>>>>> On Wed, May 8, 2024 at 7:20 AM Chris Egerton > >>>> <chr...@aiven.io.invalid > >>>>>> > >>>>>>>>> wrote: > >>>>>>>>> > >>>>>>>>>> Hi Alieh, > >>>>>>>>>> > >>>>>>>>>> Continuing prior discussions: > >>>>>>>>>> > >>>>>>>>>> 1) Regarding the "flexibility" discussion, my overarching > >> point > >>> is > >>>>>>>> that I > >>>>>>>>>> don't see the point in allowing for this kind of pluggable > >> logic > >>>>>>>> without > >>>>>>>>>> also covering more scenarios. Take example 2 in the KIP: if > >>> we're > >>>>>> going > >>>>>>>>> to > >>>>>>>>>> implement retries only on "important" topics when a topic > >>>> partition > >>>>>>>> isn't > >>>>>>>>>> found, why wouldn't we also want to be able to do this for > >> other > >>>>>>>> errors? > >>>>>>>>>> Again, taking authorization errors as an example, why wouldn't > >>> we > >>>>> want > >>>>>>>> to > >>>>>>>>>> be able to fail when we can't write to "important" topics > >>> because > >>>>> the > >>>>>>>>>> producer principal lacks sufficient ACLs, and drop the record > >> if > >>>> the > >>>>>>>>> topic > >>>>>>>>>> isn't "important"? In a security-conscious environment with > >>>>>>>>>> runtime-dependent topic routing (which is a common feature of > >>> many > >>>>>>>> source > >>>>>>>>>> connectors, such as the Debezium connectors), this seems > >> fairly > >>>>>> likely. > >>>>>>>>>> > >>>>>>>>>> 2) As far as changing the shape of the API goes, I like > >> Artem's > >>>> idea > >>>>>> of > >>>>>>>>>> splitting out the interface based on specific exceptions. This > >>> may > >>>>> be > >>>>>> a > >>>>>>>>>> little laborious to expand in the future, but if we really > >> want > >>> to > >>>>>>>>>> limit the exceptions that we cover with the handler and move > >>>> slowly > >>>>>> and > >>>>>>>>>> cautiously, then IMO it'd be reasonable to reflect that in the > >>>>>>>>> interface. I > >>>>>>>>>> also acknowledge that there's no way to completely prevent > >>> people > >>>>> from > >>>>>>>>>> shooting themselves in the foot by implementing the API > >>>> incorrectly, > >>>>>>>> but > >>>>>>>>> I > >>>>>>>>>> think it's worth it to do what we can--including leveraging > >> the > >>>> Java > >>>>>>>>>> language's type system--to help them, so IMO there's value to > >>>>>>>> eliminating > >>>>>>>>>> the implicit behavior of failing when a policy returns RETRY > >>> for a > >>>>>>>>>> non-retriable error. This can take a variety of shapes and I'm > >>> not > >>>>>>>> going > >>>>>>>>> to > >>>>>>>>>> insist on anything specific, but I do want to again raise my > >>>>> concerns > >>>>>>>>> with > >>>>>>>>>> the current proposal and request that we find something a > >> little > >>>>>>>> better. > >>>>>>>>>> > >>>>>>>>>> 3) Concerning the default implementation--actually, I meant > >>> what I > >>>>>>>> wrote > >>>>>>>>> :) > >>>>>>>>>> I don't want a "second" default, I want an implementation of > >>> this > >>>>>>>>> interface > >>>>>>>>>> to be used as the default if no others are specified. The > >>> behavior > >>>>> of > >>>>>>>>> this > >>>>>>>>>> default implementation would be identical to existing behavior > >>> (so > >>>>>>>> there > >>>>>>>>>> would be no backwards compatibility concerns like the ones > >>> raised > >>>> by > >>>>>>>>>> Matthias), but it would be possible to configure this default > >>>>> handler > >>>>>>>>> class > >>>>>>>>>> to behave differently for a basic set of scenarios. This would > >>>>> mirror > >>>>>>>>> (pun > >>>>>>>>>> intended) the approach we've taken with Mirror Maker 2 and its > >>>>>>>>>> ReplicationPolicy interface [1]. There is a default > >>> implementation > >>>>>>>>>> available [2] that recognizes a handful of basic configuration > >>>>>>>> properties > >>>>>>>>>> [3] for simple tweaks, but if users want, they can also > >>> implement > >>>>>> their > >>>>>>>>> own > >>>>>>>>>> replication policy for more fine-grained logic if those > >>> properties > >>>>>>>> aren't > >>>>>>>>>> flexible enough. > >>>>>>>>>> > >>>>>>>>>> More concretely, I'm imagining something like this for the > >>>> producer > >>>>>>>>>> exception handler: > >>>>>>>>>> > >>>>>>>>>> - Default implementation class > >>>>>>>>>> of > >>>> org.apache.kafka.clients.producer.DefaultProducerExceptionHandler > >>>>>>>>>> - This class would recognize two properties: > >>>>>>>>>> - drop.invalid.large.records: Boolean property, defaults to > >>>> false. > >>>>> If > >>>>>>>>>> "false", then causes the handler to return FAIL whenever > >>>>>>>>>> a RecordTooLargeException is encountered; if "true", then > >> causes > >>>>>>>>>> SWALLOW/SKIP/DROP to be returned instead > >>>>>>>>>> - unknown.topic.partition.retry.timeout.ms: Integer > >> property, > >>>>>>>> defaults > >>>>>>>>>> to > >>>>>>>>>> INT_MAX. Whenever an UnknownTopicOrPartitionException is > >>>>> encountered, > >>>>>>>>>> causes the handler to return FAIL if that record has been > >>> pending > >>>>> for > >>>>>>>>> more > >>>>>>>>>> than the retry timeout; otherwise, causes RETRY to be returned > >>>>>>>>>> > >>>>>>>>>> I think this is worth addressing now instead of later because > >> it > >>>>>> forces > >>>>>>>>> us > >>>>>>>>>> to evaluate the usefulness of this interface and it addresses > >> a > >>>>>>>>>> long-standing issue not just with Kafka Connect, but with the > >>> Java > >>>>>>>>> producer > >>>>>>>>>> in general. For reference, here are a few tickets I collected > >>>> after > >>>>>>>>> briefly > >>>>>>>>>> skimming our Jira showing that this is a real pain point for > >>>> users: > >>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-10340, > >>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-12990, > >>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-13634. Although > >>> this > >>>> is > >>>>>>>>>> frequently reported with Kafka Connect, it applies to anyone > >> who > >>>>>>>>> configures > >>>>>>>>>> a producer to use a high retry timeout. I am aware of the > >>>>>> max.block.ms > >>>>>>>>>> property, but it's painful and IMO poor behavior to require > >>> users > >>>> to > >>>>>>>>> reduce > >>>>>>>>>> the value of this property just to handle the single scenario > >>> when > >>>>>>>> trying > >>>>>>>>>> to write to topics that don't exist, since it would also limit > >>> the > >>>>>>>> retry > >>>>>>>>>> timeout for other operations that are legitimately retriable. > >>>>>>>>>> > >>>>>>>>>> Raising new points: > >>>>>>>>>> > >>>>>>>>>> 5) I don't see the interplay between this handler and existing > >>>>>>>>>> retry-related properties mentioned anywhere in the KIP. I'm > >>>> assuming > >>>>>>>> that > >>>>>>>>>> properties like "retries", "max.block.ms", and " > >>>> delivery.timeout.ms > >>>>> " > >>>>>>>>> would > >>>>>>>>>> take precedence over the handler and once they are exhausted, > >>> the > >>>>>>>>>> record/batch will fail no matter what? If so, it's probably > >>> worth > >>>>>>>> briefly > >>>>>>>>>> mentioning this (no more than a sentence or two) in the KIP, > >> and > >>>> if > >>>>>>>> not, > >>>>>>>>>> I'm curious what you have in mind. > >>>>>>>>>> > >>>>>>>>>> 6) I also wonder if the API provides enough information in its > >>>>> current > >>>>>>>>>> form. Would it be possible to provide handlers with some way > >> of > >>>>>>>> tracking > >>>>>>>>>> how long a record has been pending for (i.e., how long it's > >> been > >>>>> since > >>>>>>>>> the > >>>>>>>>>> record was provided as an argument to Producer::send)? One way > >>> to > >>>> do > >>>>>>>> this > >>>>>>>>>> could be to add a method like `onNewRecord(ProducerRecord)` > >> and > >>>>>>>>>> allow/require the handler to do its own bookkeeping, probably > >>>> with a > >>>>>>>>>> matching `onRecordSuccess(ProducerRecord)` method so that the > >>>>> handler > >>>>>>>>>> doesn't eat up an ever-increasing amount of memory trying to > >>> track > >>>>>>>>> records. > >>>>>>>>>> An alternative could be to include information about the > >> initial > >>>>> time > >>>>>>>> the > >>>>>>>>>> record was received by the producer and the number of retries > >>> that > >>>>>> have > >>>>>>>>>> been performed for it as parameters in the handle method(s), > >> but > >>>> I'm > >>>>>>>> not > >>>>>>>>>> sure how easy this would be to implement and it might clutter > >>>> things > >>>>>>>> up a > >>>>>>>>>> bit too much. > >>>>>>>>>> > >>>>>>>>>> 7) A small request--can we add Closeable (or, if you prefer, > >>>>>>>>> AutoCloseable) > >>>>>>>>>> as a superinterface for the handler interface? > >>>>>>>>>> > >>>>>>>>>> [1] - > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>> > >>>>> > >>>> > >>> > >> > https://kafka.apache.org/37/javadoc/org/apache/kafka/connect/mirror/ReplicationPolicy.html > >>>>>>>>>> [2] - > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>> > >>>>> > >>>> > >>> > >> > https://kafka.apache.org/37/javadoc/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.html > >>>>>>>>>> [3] - > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>> > >>>>> > >>>> > >>> > >> > https://kafka.apache.org/37/javadoc/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.html#SEPARATOR_CONFIG > >>>>>>>>>> , > >>>>>>>>>> > >>>>>>>>>> > >>>>>>>>> > >>>>>>>> > >>>>>> > >>>>> > >>>> > >>> > >> > https://kafka.apache.org/37/javadoc/org/apache/kafka/connect/mirror/DefaultReplicationPolicy.html#INTERNAL_TOPIC_SEPARATOR_ENABLED_CONFIG > >>>>>>>>>> > >>>>>>>>>> Cheers, > >>>>>>>>>> > >>>>>>>>>> Chris > >>>>>>>>>> > >>>>>>>>>> On Wed, May 8, 2024 at 12:37 AM Matthias J. Sax < > >>> mj...@apache.org > >>>>> > >>>>>>>>> wrote: > >>>>>>>>>> > >>>>>>>>>>> Very interesting discussion. > >>>>>>>>>>> > >>>>>>>>>>> Seems a central point is the question about "how generic" we > >>>>> approach > >>>>>>>>>>> this, and some people think we need to be conservative and > >>> others > >>>>>>>> think > >>>>>>>>>>> we should try to be as generic as possible. > >>>>>>>>>>> > >>>>>>>>>>> Personally, I think following a limited scope for this KIP > >> (by > >>>>>>>>>>> explicitly saying we only cover RecordTooLarge and > >>>>>>>>>>> UnknownTopicOrPartition) might be better. We have concrete > >>>> tickets > >>>>>>>> that > >>>>>>>>>>> we address, while for other exception (like authorization) we > >>>> don't > >>>>>>>>> know > >>>>>>>>>>> if people want to handle it to begin with. Boiling the ocean > >>>> might > >>>>>>>> not > >>>>>>>>>>> get us too far, and being somewhat pragmatic might help to > >> move > >>>>> this > >>>>>>>>> KIP > >>>>>>>>>>> forward. -- I also agree with Justin and Artem, that we want > >> to > >>>> be > >>>>>>>>>>> careful anyway to not allow users to break stuff too easily. > >>>>>>>>>>> > >>>>>>>>>>> As the same time, I agree that we should setup this change > >> in a > >>>>>>>> forward > >>>>>>>>>>> looking way, and thus having a single generic handler allows > >> us > >>>> to > >>>>>>>>> later > >>>>>>>>>>> extend the handler more easily. This should also simplify > >>> follow > >>>> up > >>>>>>>> KIP > >>>>>>>>>>> that might add new error cases (I actually mentioned one more > >>> to > >>>>>>>> Alieh > >>>>>>>>>>> already, but we both agreed that it might be best to exclude > >> it > >>>>> from > >>>>>>>>> the > >>>>>>>>>>> KIP right now, to make the 3.8 deadline. Doing a follow up > >> KIP > >>> is > >>>>> not > >>>>>>>>>>> the end of the world.) > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> @Chris: > >>>>>>>>>>> > >>>>>>>>>>> (2) This sounds fair to me, but not sure how "bad" it > >> actually > >>>>> would > >>>>>>>>> be? > >>>>>>>>>>> If the contract is clearly defined, it seems to be fine what > >>> the > >>>>> KIP > >>>>>>>>>>> proposes, and given that such a handler is an expert API, and > >>> we > >>>>> can > >>>>>>>>>>> provide "best practices" (cf my other comment below in > >> [AL1]), > >>>>> being > >>>>>>>> a > >>>>>>>>>>> little bit pragmatic sound fine to me. > >>>>>>>>>>> > >>>>>>>>>>> Not sure if I understand Justin's argument on this question? > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> (3) About having a default impl or not. I am fine with adding > >>>> one, > >>>>>>>> even > >>>>>>>>>>> if I am not sure why Connect could not just add its own one > >> and > >>>>> plug > >>>>>>>> it > >>>>>>>>>>> in (and we would add corresponding configs for Connect, but > >> not > >>>> for > >>>>>>>> the > >>>>>>>>>>> producer itself)? For this case, we could also do this as a > >>>> follow > >>>>> up > >>>>>>>>>>> KIP, but happy to include it in this KIP to provide value to > >>>>> Connect > >>>>>>>>>>> right away (even if the value might not come right away if we > >>>> miss > >>>>>>>> the > >>>>>>>>>>> 3.8 deadline due to expanded KIP scope...) -- For KS, we > >> would > >>>> for > >>>>>>>>> sure > >>>>>>>>>>> plugin our own impl, and lock down the config such that users > >>>>> cannot > >>>>>>>>> set > >>>>>>>>>>> their own handler on the internal producer to begin with. > >> Might > >>>> be > >>>>>>>> good > >>>>>>>>>>> to elaborate why the producer should have a default? We might > >>>>>>>> actually > >>>>>>>>>>> want to add this to the KIP right away? > >>>>>>>>>>> > >>>>>>>>>>> The key for a default impl would be, to not change the > >> current > >>>>>>>>> behavior, > >>>>>>>>>>> and having no default seems to achieve this. For the two > >> cases > >>>> you > >>>>>>>>>>> mentioned, it's unclear to me what default value on "upper > >>> bound > >>>> on > >>>>>>>>>>> retires" for UnkownTopicOrPartitionException we should set? > >>> Seems > >>>>> it > >>>>>>>>>>> would need to be the same as `delivery.timeout.ms`? However, > >>> if > >>>>>>>> users > >>>>>>>>>>> have `delivery.timeout.ms` actually overwritten we would > >> need > >>> to > >>>>> set > >>>>>>>>>>> this config somewhat "dynamic"? Is this feasible? If we > >>>> hard-code 2 > >>>>>>>>>>> minutes, it might not be backward compatible. I have the > >>>> impression > >>>>>>>> we > >>>>>>>>>>> might introduce some undesired coupling? -- For the "record > >> too > >>>>>>>> large" > >>>>>>>>>>> case, the config seems to be boolean and setting it to > >> `false` > >>> by > >>>>>>>>>>> default seems to provide backward compatibility. > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> @Artem: > >>>>>>>>>>> > >>>>>>>>>>> [AL1] While I see the point, I would think having a different > >>>>>>>> callback > >>>>>>>>>>> for every exception might not really be elegant? In the end, > >>> the > >>>>>>>>> handler > >>>>>>>>>>> is an very advanced feature anyway, and if it's implemented > >> in > >>> a > >>>>> bad > >>>>>>>>>>> way, well, it's a user error -- we cannot protect users from > >>>>>>>>> everything. > >>>>>>>>>>> To me, a handler like this, is to some extend "business > >> logic" > >>>> and > >>>>>>>> if a > >>>>>>>>>>> user gets business logic wrong, it's hard to protect them. -- > >>> We > >>>>>>>> would > >>>>>>>>>>> of course provide best practice guidance in the JaveDocs, and > >>>>> explain > >>>>>>>>>>> that a handler should have explicit `if` statements for stuff > >>> it > >>>>> want > >>>>>>>>> to > >>>>>>>>>>> handle, and only a single default which return FAIL. > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> [AL2] Yes, but for KS we would retry at the application > >> layer. > >>>> Ie, > >>>>>>>> the > >>>>>>>>>>> TX is not completed, we clean up and setup out task from > >>> scratch, > >>>>> to > >>>>>>>>>>> ensure the pending transaction is completed before we resume. > >>> If > >>>>> the > >>>>>>>> TX > >>>>>>>>>>> was indeed aborted, we would retry from older offset and thus > >>>> just > >>>>>>>> hit > >>>>>>>>>>> the same error again and the loop begins again. > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> [AL2 cont.] Similar to AL1, I see such a handler to some > >> extend > >>>> as > >>>>>>>>>>> business logic. If a user puts a bad filter condition in > >> their > >>> KS > >>>>>>>> app, > >>>>>>>>>>> and drops messages, it nothing we can do about it, and this > >>>> handler > >>>>>>>>>>> IMHO, has a similar purpose. This is also the line of > >> thinking > >>> I > >>>>>>>> apply > >>>>>>>>>>> to EOS, to address Justin's concern about "should we allow to > >>>> drop > >>>>>>>> for > >>>>>>>>>>> EOS", and my answer is "yes", because it's more business > >> logic > >>>> than > >>>>>>>>>>> actual error handling IMHO. And by default, we fail... So > >> users > >>>>>>>> opt-in > >>>>>>>>>>> to add business logic to drop records. It's an application > >>> level > >>>>>>>>>>> decision how to write the code. > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> [AL3] Maybe I misunderstand what you are saying, but to me, > >>>>> checking > >>>>>>>>> the > >>>>>>>>>>> size of the record upfront is exactly what the KIP proposes? > >>> No? > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> @Justin: > >>>>>>>>>>> > >>>>>>>>>>>> I saw the sample > >>>>>>>>>>>> code -- is it just an if statement checking for the error > >>> before > >>>>>>>> the > >>>>>>>>>>>> handler is invoked? That seems a bit fragile. > >>>>>>>>>>> > >>>>>>>>>>> What do you mean by fragile? Not sure if I see your point. > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> > >>>>>>>>>>> -Matthias > >>>>>>>>>>> > >>>>>>>>>>> On 5/7/24 5:33 PM, Artem Livshits wrote: > >>>>>>>>>>>> Hi Alieh, > >>>>>>>>>>>> > >>>>>>>>>>>> Thanks for the KIP. The motivation talks about very > >> specific > >>>>>>>> cases, > >>>>>>>>>> but > >>>>>>>>>>>> the interface is generic. > >>>>>>>>>>>> > >>>>>>>>>>>> [AL1] > >>>>>>>>>>>> If the interface evolves in the future I think we could have > >>> the > >>>>>>>>>>> following > >>>>>>>>>>>> confusion: > >>>>>>>>>>>> > >>>>>>>>>>>> 1. A user implemented SWALLOW action for both > >>>>>>>> RecordTooLargeException > >>>>>>>>>> and > >>>>>>>>>>>> UnknownTopicOrPartitionException. For simpicity they just > >>>> return > >>>>>>>>>> SWALLOW > >>>>>>>>>>>> from the function, because it elegantly handles all known > >>> cases. > >>>>>>>>>>>> 2. The interface has evolved to support a new exception. > >>>>>>>>>>>> 3. The user has upgraded their Kafka client. > >>>>>>>>>>>> > >>>>>>>>>>>> Now a new kind of error gets dropped on the floor without > >>> user's > >>>>>>>>>>> intention > >>>>>>>>>>>> and it would be super hard to detect and debug. > >>>>>>>>>>>> > >>>>>>>>>>>> To avoid the confusion, I think we should use handlers for > >>>>> specific > >>>>>>>>>>>> exceptions. Then if a new exception is added it won't get > >>>>> silently > >>>>>>>>>>> swalled > >>>>>>>>>>>> because the user would need to add new functionality to > >> handle > >>>> it. > >>>>>>>>>>>> > >>>>>>>>>>>> I also have some higher level comments: > >>>>>>>>>>>> > >>>>>>>>>>>> [AL2] > >>>>>>>>>>>>> it throws a TimeoutException, and the user can only blindly > >>>>> retry, > >>>>>>>>>> which > >>>>>>>>>>>> may result in an infinite retry loop > >>>>>>>>>>>> > >>>>>>>>>>>> If the TimeoutException happens during transactional > >>> processing > >>>>>>>>>> (exactly > >>>>>>>>>>>> once is the desired sematnics), then the client should not > >>> retry > >>>>>>>> when > >>>>>>>>>> it > >>>>>>>>>>>> gets TimeoutException because without knowing the reason for > >>>>>>>>>>>> TimeoutExceptions, the client cannot know whether the > >> message > >>>> got > >>>>>>>>>>> actually > >>>>>>>>>>>> produced or not and retrying the message may result in > >>>>> duplicatees. > >>>>>>>>>>>> > >>>>>>>>>>>>> The thrown TimeoutException "cuts" the connection to the > >>>>>>>> underlying > >>>>>>>>>> root > >>>>>>>>>>>> cause of missing metadata > >>>>>>>>>>>> > >>>>>>>>>>>> Maybe we should fix the error handling and return the proper > >>>>>>>>> underlying > >>>>>>>>>>>> message? Then the application can properly handle the > >> message > >>>>>>>> based > >>>>>>>>> on > >>>>>>>>>>>> preferences. > >>>>>>>>>>>> > >>>>>>>>>>>> From the product perspective, it's not clear how safe it is > >> to > >>>>>>>>> blindly > >>>>>>>>>>>> ignore UnknownTopicOrPartitionException. This could lead to > >>>>>>>>> situations > >>>>>>>>>>>> when a simple typo could lead to massive data loss (part of > >>> the > >>>>>>>> data > >>>>>>>>>>> would > >>>>>>>>>>>> effectively be produced to a "black hole" and the user may > >> not > >>>>>>>> notice > >>>>>>>>>> it > >>>>>>>>>>>> for a while). > >>>>>>>>>>>> > >>>>>>>>>>>> In which situations would you recommend the user to "black > >>> hole" > >>>>>>>>>> messages > >>>>>>>>>>>> in case of misconfiguration? > >>>>>>>>>>>> > >>>>>>>>>>>> [AL3] > >>>>>>>>>>>> > >>>>>>>>>>>>> If the custom handler decides on SWALLOW for > >>>>>>>>> RecordTooLargeException, > >>>>>>>>>>>> > >>>>>>>>>>>> Is it my understanding that this KIP proposes that > >>> functionality > >>>>>>>> that > >>>>>>>>>>> would > >>>>>>>>>>>> only be able to SWALLOW RecordTooLargeException that happen > >>>>> because > >>>>>>>>> the > >>>>>>>>>>>> producer cannot produce the record (if the broker rejects > >> the > >>>>>>>> batch, > >>>>>>>>>> the > >>>>>>>>>>>> error won't get to the handler, because we cannot know which > >>>> other > >>>>>>>>>>> records > >>>>>>>>>>>> get ignored). In this case, why not just check the locally > >>>>>>>>> configured > >>>>>>>>>>> max > >>>>>>>>>>>> record size upfront and not produce the recrord in the first > >>>>> place? > >>>>>>>>>>> Maybe > >>>>>>>>>>>> we can expose a validation function from the producer that > >>> could > >>>>>>>>>> validate > >>>>>>>>>>>> the records locally, so we don't need to produce the record > >> in > >>>>>>>> order > >>>>>>>>> to > >>>>>>>>>>>> know that it's invalid. > >>>>>>>>>>>> > >>>>>>>>>>>> -Artem > >>>>>>>>>>>> > >>>>>>>>>>>> On Tue, May 7, 2024 at 2:07 PM Justine Olshan > >>>>>>>>>>> <jols...@confluent.io.invalid> > >>>>>>>>>>>> wrote: > >>>>>>>>>>>> > >>>>>>>>>>>>> Alieh and Chris, > >>>>>>>>>>>>> > >>>>>>>>>>>>> Thanks for clarifying 1) but I saw the motivation. I guess > >> I > >>>> just > >>>>>>>>>> didn't > >>>>>>>>>>>>> understand how that would be ensured on the producer side. > >> I > >>>> saw > >>>>>>>> the > >>>>>>>>>>> sample > >>>>>>>>>>>>> code -- is it just an if statement checking for the error > >>>> before > >>>>>>>> the > >>>>>>>>>>>>> handler is invoked? That seems a bit fragile. > >>>>>>>>>>>>> > >>>>>>>>>>>>> Can you clarify what you mean by `since the code does not > >>> reach > >>>>>>>> the > >>>>>>>>> KS > >>>>>>>>>>>>> interface and breaks somewhere in producer.` If we surfaced > >>>> this > >>>>>>>>> error > >>>>>>>>>>> to > >>>>>>>>>>>>> the application in a better way would that also be a > >> solution > >>>> to > >>>>>>>> the > >>>>>>>>>>> issue? > >>>>>>>>>>>>> > >>>>>>>>>>>>> Justine > >>>>>>>>>>>>> > >>>>>>>>>>>>> On Tue, May 7, 2024 at 1:55 PM Alieh Saeedi > >>>>>>>>>>> <asae...@confluent.io.invalid> > >>>>>>>>>>>>> wrote: > >>>>>>>>>>>>> > >>>>>>>>>>>>>> Hi, > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Thank you, Chris and Justine, for the feedback. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> @Chris > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> 1) Flexibility: it has two meanings. The first meaning is > >>> the > >>>>> one > >>>>>>>>> you > >>>>>>>>>>>>>> mentioned. We are going to cover more exceptions in the > >>>> future, > >>>>>>>> but > >>>>>>>>>> as > >>>>>>>>>>>>>> Justine mentioned, we must be very conservative about > >> adding > >>>>> more > >>>>>>>>>>>>>> exceptions. Additionally, flexibility mainly means that > >> the > >>>> user > >>>>>>>> is > >>>>>>>>>>> able > >>>>>>>>>>>>> to > >>>>>>>>>>>>>> develop their own code. As mentioned in the motivation > >>> section > >>>>>>>> and > >>>>>>>>>> the > >>>>>>>>>>>>>> examples, sometimes the user decides on dropping a record > >>>> based > >>>>>>>> on > >>>>>>>>>> the > >>>>>>>>>>>>>> topic, for example. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> 2) Defining two separate methods for retriable and > >>>> non-retriable > >>>>>>>>>>>>>> exceptions: although the idea is brilliant, the user may > >>> still > >>>>>>>>> make a > >>>>>>>>>>>>>> mistake by implementing the wrong method and see a > >>>> non-expecting > >>>>>>>>>>>>> behaviour. > >>>>>>>>>>>>>> For example, he may implement handleRetriable() for > >>>>>>>>>>>>> RecordTooLargeException > >>>>>>>>>>>>>> and define SWALLOW for the exception, but in practice, he > >>> sees > >>>>> no > >>>>>>>>>>> change > >>>>>>>>>>>>> in > >>>>>>>>>>>>>> default behaviour since he implemented the wrong method. I > >>>> think > >>>>>>>> we > >>>>>>>>>> can > >>>>>>>>>>>>>> never reduce the user’s mistakes to 0. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> 3) Default implementation for Handler: the default > >> behaviour > >>>> is > >>>>>>>>>> already > >>>>>>>>>>>>>> preserved with NO need of implementing any handler or > >>> setting > >>>>> the > >>>>>>>>>>>>>> corresponding config parameter `custom.exception.handler`. > >>>> What > >>>>>>>> you > >>>>>>>>>>> mean > >>>>>>>>>>>>> is > >>>>>>>>>>>>>> actually having a second default, which requires having > >> both > >>>>>>>>>> interface > >>>>>>>>>>>>> and > >>>>>>>>>>>>>> config parameters. About UnknownTopicOrPartitionException: > >>> the > >>>>>>>>>> producer > >>>>>>>>>>>>>> already offers the config parameter `max.block.ms` which > >>>>>>>>> determines > >>>>>>>>>>> the > >>>>>>>>>>>>>> duration of retrying. The main purpose of the user who > >> needs > >>>> the > >>>>>>>>>>> handler > >>>>>>>>>>>>> is > >>>>>>>>>>>>>> to get the root cause of TimeoutException and handle it in > >>> the > >>>>>>>> way > >>>>>>>>> he > >>>>>>>>>>>>>> intends. The KIP explains the necessity of it for KS > >> users. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> 4) Naming issue: By SWALLOW, we meant actually swallow the > >>>>> error, > >>>>>>>>>> while > >>>>>>>>>>>>>> SKIP means skip the record, I think. If it makes sense for > >>>> more > >>>>>>>>> ppl, > >>>>>>>>>> I > >>>>>>>>>>>>> can > >>>>>>>>>>>>>> change it to SKIP > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> @Justine > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> 1) was addressed by Chris. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> 2 and 3) The problem is exactly what you mentioned. > >>> Currently, > >>>>>>>>> there > >>>>>>>>>> is > >>>>>>>>>>>>> no > >>>>>>>>>>>>>> way to handle these issues application-side. Even KS users > >>> who > >>>>>>>>>>> implement > >>>>>>>>>>>>> KS > >>>>>>>>>>>>>> ProductionExceptionHandler are not able to handle the > >>>> exceptions > >>>>>>>> as > >>>>>>>>>>> they > >>>>>>>>>>>>>> intend since the code does not reach the KS interface and > >>>> breaks > >>>>>>>>>>>>> somewhere > >>>>>>>>>>>>>> in producer. > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>> Alieh > >>>>>>>>>>>>>> > >>>>>>>>>>>>>> On Tue, May 7, 2024 at 8:43 PM Chris Egerton < > >>>>>>>>>> fearthecel...@gmail.com> > >>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Hi Justine, > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> The method signatures for the interface are indeed > >>>> open-ended, > >>>>>>>> but > >>>>>>>>>> the > >>>>>>>>>>>>>> KIP > >>>>>>>>>>>>>>> states that its uses will be limited. See the motivation > >>>>>>>> section: > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> We believe that the user should be able to develop > >> custom > >>>>>>>>> exception > >>>>>>>>>>>>>>> handlers for managing producer exceptions. On the other > >>> hand, > >>>>>>>> this > >>>>>>>>>>> will > >>>>>>>>>>>>>> be > >>>>>>>>>>>>>>> an expert-level API, and using that may result in strange > >>>>>>>>> behaviour > >>>>>>>>>> in > >>>>>>>>>>>>>> the > >>>>>>>>>>>>>>> system, making it hard to find the root cause. Therefore, > >>> the > >>>>>>>>> custom > >>>>>>>>>>>>>>> handler is currently limited to handling > >>>>> RecordTooLargeException > >>>>>>>>> and > >>>>>>>>>>>>>>> UnknownTopicOrPartitionException. > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> Chris > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>> On Tue, May 7, 2024, 14:37 Justine Olshan > >>>>>>>>>>> <jols...@confluent.io.invalid > >>>>>>>>>>>>>> > >>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Hi Alieh, > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> I was out for KSB and then was also sick. :( > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> To your point 1) Chris, I don't think it is limited to > >> two > >>>>>>>>> specific > >>>>>>>>>>>>>>>> scenarios, since the interface accepts a generic > >>> Exception e > >>>>>>>> and > >>>>>>>>>> can > >>>>>>>>>>>>> be > >>>>>>>>>>>>>>>> implemented to check if that e is an instanceof any > >>>> exception. > >>>>>>>> I > >>>>>>>>>>>>> didn't > >>>>>>>>>>>>>>> see > >>>>>>>>>>>>>>>> anywhere that specific errors are enforced. I'm a bit > >>>>> concerned > >>>>>>>>>> about > >>>>>>>>>>>>>>> this > >>>>>>>>>>>>>>>> actually. I'm concerned about the opened-endedness and > >> the > >>>>>>>>> contract > >>>>>>>>>>>>> we > >>>>>>>>>>>>>>> have > >>>>>>>>>>>>>>>> with transactions. We are allowing the client to make > >>>>> decisions > >>>>>>>>>> that > >>>>>>>>>>>>>> are > >>>>>>>>>>>>>>>> somewhat invisible to the server. As an aside, can we > >>> build > >>>> in > >>>>>>>>> log > >>>>>>>>>>>>>>> messages > >>>>>>>>>>>>>>>> when the handler decides to skip etc a message. I'm > >> really > >>>>>>>>>> concerned > >>>>>>>>>>>>>>> about > >>>>>>>>>>>>>>>> messages being silently dropped. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> I do think Chris's point 2) about retriable vs non > >>> retriable > >>>>>>>>> errors > >>>>>>>>>>>>> is > >>>>>>>>>>>>>>>> fair. I'm a bit concerned about skipping a unknown topic > >>> or > >>>>>>>>>> partition > >>>>>>>>>>>>>>>> exception too early, as there are cases where it can be > >>>>>>>>> transient. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> I'm still a little bit wary of allowing dropping records > >>> as > >>>>>>>> part > >>>>>>>>> of > >>>>>>>>>>>>> EOS > >>>>>>>>>>>>>>>> generally as in many cases, these errors signify an > >> issue > >>>> with > >>>>>>>>> the > >>>>>>>>>>>>>>> original > >>>>>>>>>>>>>>>> data. I understand that streams and connect/mirror maker > >>> may > >>>>>>>> have > >>>>>>>>>>>>>> reasons > >>>>>>>>>>>>>>>> they want to progress past these messages, but wondering > >>> if > >>>>>>>> there > >>>>>>>>>> is > >>>>>>>>>>>>> a > >>>>>>>>>>>>>>> way > >>>>>>>>>>>>>>>> that can be done application-side. I'm willing to accept > >>>> this > >>>>>>>>> sort > >>>>>>>>>> of > >>>>>>>>>>>>>>>> proposal if we can make it clear that this sort of thing > >>> is > >>>>>>>>>> happening > >>>>>>>>>>>>>> and > >>>>>>>>>>>>>>>> we limit the blast radius for what we can do. > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> Justine > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> On Tue, May 7, 2024 at 9:55 AM Chris Egerton > >>>>>>>>>> <chr...@aiven.io.invalid > >>>>>>>>>>>>>> > >>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Hi Alieh, > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Sorry for the delay, I've been out sick. I still have > >>> some > >>>>>>>>>> thoughts > >>>>>>>>>>>>>>> that > >>>>>>>>>>>>>>>>> I'd like to see addressed before voting. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> 1) If flexibility is the motivation for a pluggable > >>>>> interface, > >>>>>>>>> why > >>>>>>>>>>>>>> are > >>>>>>>>>>>>>>> we > >>>>>>>>>>>>>>>>> only limiting the uses for this interface to two very > >>>>> specific > >>>>>>>>>>>>>>> scenarios? > >>>>>>>>>>>>>>>>> Why not also allow, e.g., authorization errors to be > >>>> handled > >>>>>>>> as > >>>>>>>>>>>>> well > >>>>>>>>>>>>>>>>> (allowing users to drop records destined for some > >>>> off-limits > >>>>>>>>>>>>> topics, > >>>>>>>>>>>>>> or > >>>>>>>>>>>>>>>>> retry for a limited duration in case there's a delay in > >>> the > >>>>>>>>>>>>>> propagation > >>>>>>>>>>>>>>>> of > >>>>>>>>>>>>>>>>> ACL updates)? It'd be nice to see some analysis of > >> other > >>>>>>>> errors > >>>>>>>>>>>>> that > >>>>>>>>>>>>>>>> could > >>>>>>>>>>>>>>>>> be handled with this new API, both to avoid the > >> follow-up > >>>>> work > >>>>>>>>> of > >>>>>>>>>>>>>>> another > >>>>>>>>>>>>>>>>> KIP to address them in the future, and to make sure > >> that > >>>>> we're > >>>>>>>>> not > >>>>>>>>>>>>>>>> painting > >>>>>>>>>>>>>>>>> ourselves into a corner with the current API in a way > >>> that > >>>>>>>> would > >>>>>>>>>>>>> make > >>>>>>>>>>>>>>>>> future modifications difficult. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> 2) Something feels a bit off with how retriable vs. > >>>>>>>>> non-retriable > >>>>>>>>>>>>>>> errors > >>>>>>>>>>>>>>>>> are handled with the interface. Why not introduce two > >>>>> separate > >>>>>>>>>>>>>> methods > >>>>>>>>>>>>>>> to > >>>>>>>>>>>>>>>>> handle each case separately? That way there's no > >>> ambiguity > >>>> or > >>>>>>>>>>>>>> implicit > >>>>>>>>>>>>>>>>> behavior when, e.g., attempting to retry on a > >>>>>>>>>>>>>> RecordTooLargeException. > >>>>>>>>>>>>>>>> This > >>>>>>>>>>>>>>>>> could be something like `NonRetriableResponse > >>>>>>>>>>>>> handle(ProducerRecord, > >>>>>>>>>>>>>>>>> Exception)` and `RetriableResponse > >>>>>>>>> handleRetriable(ProducerRecord, > >>>>>>>>>>>>>>>>> Exception)`, though the exact names and shape can > >>> obviously > >>>>> be > >>>>>>>>>>>>> toyed > >>>>>>>>>>>>>>>> with a > >>>>>>>>>>>>>>>>> bit. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> 3) Although the flexibility of a pluggable interface > >> may > >>>>>>>> benefit > >>>>>>>>>>>>> some > >>>>>>>>>>>>>>>>> users' custom producer applications and Kafka Streams > >>>>>>>>>> applications, > >>>>>>>>>>>>>> it > >>>>>>>>>>>>>>>>> comes at significant deployment cost for other > >>> low-/no-code > >>>>>>>>>>>>>>> environments, > >>>>>>>>>>>>>>>>> including but not limited to Kafka Connect and > >>> MirrorMaker > >>>> 2. > >>>>>>>>> Can > >>>>>>>>>>>>> we > >>>>>>>>>>>>>>> add > >>>>>>>>>>>>>>>> a > >>>>>>>>>>>>>>>>> default implementation of the exception handler that > >>> allows > >>>>>>>> for > >>>>>>>>>>>>> some > >>>>>>>>>>>>>>>> simple > >>>>>>>>>>>>>>>>> behavior to be tweaked via configuration property? Two > >>>> things > >>>>>>>>> that > >>>>>>>>>>>>>>> would > >>>>>>>>>>>>>>>> be > >>>>>>>>>>>>>>>>> nice to have would be A) an upper bound on the retry > >> time > >>>> for > >>>>>>>>>>>>>>>>> unknown-topic-partition exceptions and B) an option to > >>> drop > >>>>>>>>>> records > >>>>>>>>>>>>>>> that > >>>>>>>>>>>>>>>>> are large enough to trigger a record-too-large > >> exception. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> 4) I'd still prefer to see "SKIP" or "DROP" instead of > >>> the > >>>>>>>>>> proposed > >>>>>>>>>>>>>>>>> "SWALLOW" option, which IMO is opaque and non-obvious, > >>>>>>>>> especially > >>>>>>>>>>>>>> when > >>>>>>>>>>>>>>>>> trying to guess the behavior for retriable errors. > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> Chris > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> On Fri, May 3, 2024 at 11:23 AM Alieh Saeedi > >>>>>>>>>>>>>>>> <asae...@confluent.io.invalid > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> Hi all, > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> A summary of the KIP and the discussions: > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> The KIP introduces a handler interface for Producer in > >>>> order > >>>>>>>> to > >>>>>>>>>>>>>>> handle > >>>>>>>>>>>>>>>>> two > >>>>>>>>>>>>>>>>>> exceptions: RecordTooLargeException and > >>>>>>>>>>>>>>>> UnknownTopicOrPartitionException. > >>>>>>>>>>>>>>>>>> The handler handles the exceptions per-record. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> - Do we need this handler? [Motivation and Examples > >>>>>>>> sections] > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> RecordTooLargeException: 1) In transactions, the > >>> producer > >>>>>>>>>>>>> collects > >>>>>>>>>>>>>>>>> multiple > >>>>>>>>>>>>>>>>>> records in batches. Then a RecordTooLargeException > >>> related > >>>>>>>> to a > >>>>>>>>>>>>>>> single > >>>>>>>>>>>>>>>>>> record leads to failing the entire batch. A custom > >>>> exception > >>>>>>>>>>>>>> handler > >>>>>>>>>>>>>>> in > >>>>>>>>>>>>>>>>>> this case may decide on dropping the record and > >>> continuing > >>>>>>>> the > >>>>>>>>>>>>>>>>> processing. > >>>>>>>>>>>>>>>>>> See Example 1, please. 2) More over, in Kafka > >> Streams, a > >>>>>>>> record > >>>>>>>>>>>>>> that > >>>>>>>>>>>>>>> is > >>>>>>>>>>>>>>>>> too > >>>>>>>>>>>>>>>>>> large is a poison pill record, and there is no way to > >>> skip > >>>>>>>> over > >>>>>>>>>>>>>> it. A > >>>>>>>>>>>>>>>>>> handler would allow us to react to this error inside > >> the > >>>>>>>>>>>>> producer, > >>>>>>>>>>>>>>>> i.e., > >>>>>>>>>>>>>>>>>> local to where the error happens, and thus simplify > >> the > >>>>>>>> overall > >>>>>>>>>>>>>> code > >>>>>>>>>>>>>>>>>> significantly. Please read the Motivation section for > >>> more > >>>>>>>>>>>>>>> explanation. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> UnknownTopicOrPartitionException: For this case, the > >>>>> producer > >>>>>>>>>>>>>> handles > >>>>>>>>>>>>>>>>> this > >>>>>>>>>>>>>>>>>> exception internally and only issues a WARN log about > >>>>> missing > >>>>>>>>>>>>>>> metadata > >>>>>>>>>>>>>>>>> and > >>>>>>>>>>>>>>>>>> retries internally. Later, when the producer hits " > >>>>>>>>>>>>>>> deliver.timeout.ms" > >>>>>>>>>>>>>>>>> it > >>>>>>>>>>>>>>>>>> throws a TimeoutException, and the user can only > >> blindly > >>>>>>>> retry, > >>>>>>>>>>>>>> which > >>>>>>>>>>>>>>>> may > >>>>>>>>>>>>>>>>>> result in an infinite retry loop. The thrown > >>>>> TimeoutException > >>>>>>>>>>>>>> "cuts" > >>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>> connection to the underlying root cause of missing > >>>> metadata > >>>>>>>>>>>>> (which > >>>>>>>>>>>>>>>> could > >>>>>>>>>>>>>>>>>> indeed be a transient error but is persistent for a > >>>>>>>>> non-existing > >>>>>>>>>>>>>>>> topic). > >>>>>>>>>>>>>>>>>> Thus, there is no programmatic way to break the > >> infinite > >>>>>>>> retry > >>>>>>>>>>>>>> loop. > >>>>>>>>>>>>>>>>> Kafka > >>>>>>>>>>>>>>>>>> Streams also blindly retries for this case, and the > >>>>>>>> application > >>>>>>>>>>>>>> gets > >>>>>>>>>>>>>>>>> stuck. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> - Having interface vs configuration option: > >> [Motivation, > >>>>>>>>>>>>> Examples, > >>>>>>>>>>>>>>> and > >>>>>>>>>>>>>>>>>> Rejected Alternatives sections] > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> Our solution is introducing an interface due to the > >> full > >>>>>>>>>>>>>> flexibility > >>>>>>>>>>>>>>>> that > >>>>>>>>>>>>>>>>>> it offers. Sometimes users, especially Kafka Streams > >>> ones, > >>>>>>>>>>>>>> determine > >>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>> handler's behaviour based on the situation. For > >>> example, f > >>>>>>>>>>>>>>>>>> acing UnknownTopicOrPartitionException*, *the user may > >>>> want > >>>>>>>> to > >>>>>>>>>>>>>> raise > >>>>>>>>>>>>>>> an > >>>>>>>>>>>>>>>>>> error for some topics but retry it for other topics. > >>>> Having > >>>>> a > >>>>>>>>>>>>>>>>> configuration > >>>>>>>>>>>>>>>>>> option with a fixed set of possibilities does not > >> serve > >>>> the > >>>>>>>>>>>>> user's > >>>>>>>>>>>>>>>>>> needs. See Example 2, please. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> - Note on RecordTooLargeException: [Public Interfaces > >>>>>>>> section] > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> If the custom handler decides on SWALLOW for > >>>>>>>>>>>>>> RecordTooLargeException, > >>>>>>>>>>>>>>>>> then > >>>>>>>>>>>>>>>>>> this record will not be a part of the batch of > >>>> transactions > >>>>>>>> and > >>>>>>>>>>>>>> will > >>>>>>>>>>>>>>>> also > >>>>>>>>>>>>>>>>>> not be sent to the broker in non-transactional mode. > >> So > >>> no > >>>>>>>>>>>>> worries > >>>>>>>>>>>>>>>> about > >>>>>>>>>>>>>>>>>> getting a RecordTooLargeException from the broker in > >>> this > >>>>>>>> case, > >>>>>>>>>>>>> as > >>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>> record will never ever be sent to the broker. SWALLOW > >>>> means > >>>>>>>>> drop > >>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>> record > >>>>>>>>>>>>>>>>>> and continue/swallow the error. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> - What if the handle() method implements RETRY for > >>>>>>>>>>>>>>>>> RecordTooLargeException? > >>>>>>>>>>>>>>>>>> [Proposed Changes section] > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> We have to limit the user to only have FAIL or SWALLOW > >>> for > >>>>>>>>>>>>>>>>>> RecordTooLargeException. Actually, RETRY must be equal > >>> to > >>>>>>>> FAIL. > >>>>>>>>>>>>>> This > >>>>>>>>>>>>>>> is > >>>>>>>>>>>>>>>>>> well documented/informed in javadoc. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> - What if the handle() method of the handler throws an > >>>>>>>>> exception? > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> The handler is expected to have correct code. If it > >>> throws > >>>>> an > >>>>>>>>>>>>>>>> exception, > >>>>>>>>>>>>>>>>>> everything fails. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> This is a PoC PR < > >>>>> https://github.com/apache/kafka/pull/15846 > >>>>>>>>> > >>>>>>>>>>>>> ONLY > >>>>>>>>>>>>>>> for > >>>>>>>>>>>>>>>>>> RecordTooLargeException. The code changes related to > >>>>>>>>>>>>>>>>>> UnknownTopicOrPartitionException will be added to this > >>> PR > >>>>>>>>> LATER. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> Looking forward to your feedback again. > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> Alieh > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>> On Thu, Apr 25, 2024 at 11:46 PM Kirk True < > >>>>>>>> k...@kirktrue.pro> > >>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Hi Alieh, > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Thanks for the updates! > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Comments inline... > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> On Apr 25, 2024, at 1:10 PM, Alieh Saeedi > >>>>>>>>>>>>>>>>> <asae...@confluent.io.INVALID > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> wrote: > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> Hi all, > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> Thanks a lot for the constructive feedbacks! > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> Addressing some of the main concerns: > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> - The `RecordTooLargeException` can be thrown by > >>> broker, > >>>>>>>>>>>>>> producer > >>>>>>>>>>>>>>>> and > >>>>>>>>>>>>>>>>>>>> consumer. Of course, the `ProducerExceptionHandler` > >>>>>>>> interface > >>>>>>>>>>>>>> is > >>>>>>>>>>>>>>>>>>> introduced > >>>>>>>>>>>>>>>>>>>> to affect only the exceptions thrown from the > >>> producer. > >>>>>>>> This > >>>>>>>>>>>>>> KIP > >>>>>>>>>>>>>>>> very > >>>>>>>>>>>>>>>>>>>> specifically means to provide a possibility to > >> manage > >>>> the > >>>>>>>>>>>>>>>>>>>> `RecordTooLargeException` thrown from the > >>>> Producer.send() > >>>>>>>>>>>>>> method. > >>>>>>>>>>>>>>>>>> Please > >>>>>>>>>>>>>>>>>>>> see “Proposed Changes” section for more clarity. I > >>>>>>>>>>>>> investigated > >>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>> issue > >>>>>>>>>>>>>>>>>>>> there thoroughly. I hope it can explain the concern > >>>> about > >>>>>>>> how > >>>>>>>>>>>>>> we > >>>>>>>>>>>>>>>>> handle > >>>>>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>>> errors as well. > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> - The problem with Callback: Methods of Callback are > >>>>> called > >>>>>>>>>>>>>> when > >>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>> record > >>>>>>>>>>>>>>>>>>>> sent to the server is acknowledged, while this is > >> not > >>>> the > >>>>>>>>>>>>>> desired > >>>>>>>>>>>>>>>>> time > >>>>>>>>>>>>>>>>>>> for > >>>>>>>>>>>>>>>>>>>> all exceptions. We intend to handle exceptions > >>>> beforehand. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> I guess it makes sense to keep the expectation for > >> when > >>>>>>>>>>>>> Callback > >>>>>>>>>>>>>> is > >>>>>>>>>>>>>>>>>>> invoked as-is vs. shoehorning more into it. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> - What if the custom handler returns RETRY for > >>>>>>>>>>>>>>>>>>> `RecordTooLargeException`? I > >>>>>>>>>>>>>>>>>>>> assume changing the producer configuration at > >> runtime > >>> is > >>>>>>>>>>>>>>> possible. > >>>>>>>>>>>>>>>> If > >>>>>>>>>>>>>>>>>> so, > >>>>>>>>>>>>>>>>>>>> RETRY for a too large record is valid because maybe > >> in > >>>> the > >>>>>>>>>>>>> next > >>>>>>>>>>>>>>>> try, > >>>>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>>> too large record is not poisoning any more. I am not > >>>> 100% > >>>>>>>>>>>>> sure > >>>>>>>>>>>>>>>> about > >>>>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>>> technical details, though. Otherwise, we can > >> consider > >>>> the > >>>>>>>>>>>>> RETRY > >>>>>>>>>>>>>>> as > >>>>>>>>>>>>>>>>> FAIL > >>>>>>>>>>>>>>>>>>> for > >>>>>>>>>>>>>>>>>>>> this exception. Another solution would be to > >> consider > >>> a > >>>>>>>>>>>>>> constant > >>>>>>>>>>>>>>>>> number > >>>>>>>>>>>>>>>>>>> of > >>>>>>>>>>>>>>>>>>>> times for RETRY which can be useful for other > >>> exceptions > >>>>> as > >>>>>>>>>>>>>> well. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> It’s not presently possible to change the > >> configuration > >>>> of > >>>>>>>> an > >>>>>>>>>>>>>>>> existing > >>>>>>>>>>>>>>>>>>> Producer at runtime. So if a record hits a > >>>>>>>>>>>>>> RecordTooLargeException > >>>>>>>>>>>>>>>>> once, > >>>>>>>>>>>>>>>>>> no > >>>>>>>>>>>>>>>>>>> amount of retrying (with the current Producer) will > >>>> change > >>>>>>>>> that > >>>>>>>>>>>>>>> fact. > >>>>>>>>>>>>>>>>> So > >>>>>>>>>>>>>>>>>>> I’m still a little stuck on how to handle a response > >> of > >>>>>>>> RETRY > >>>>>>>>>>>>> for > >>>>>>>>>>>>>>> an > >>>>>>>>>>>>>>>>>>> “oversized” record. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> - What if the handle() method itself throws an > >>>> exception? > >>>>> I > >>>>>>>>>>>>>> think > >>>>>>>>>>>>>>>>>>>> rationally and pragmatically, the behaviour must be > >>>>> exactly > >>>>>>>>>>>>>> like > >>>>>>>>>>>>>>>> when > >>>>>>>>>>>>>>>>>> no > >>>>>>>>>>>>>>>>>>>> custom handler is defined since the user actually > >> did > >>>> not > >>>>>>>>>>>>> have > >>>>>>>>>>>>>> a > >>>>>>>>>>>>>>>>>> working > >>>>>>>>>>>>>>>>>>>> handler. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> I’m not convinced that ignoring an errant handler is > >>> the > >>>>>>>> right > >>>>>>>>>>>>>>>> choice. > >>>>>>>>>>>>>>>>> It > >>>>>>>>>>>>>>>>>>> then becomes a silent failure that might have > >>>>> repercussions, > >>>>>>>>>>>>>>>> depending > >>>>>>>>>>>>>>>>> on > >>>>>>>>>>>>>>>>>>> the business logic. A user would have to proactively > >>>> trawls > >>>>>>>>>>>>>> through > >>>>>>>>>>>>>>>> the > >>>>>>>>>>>>>>>>>>> logs for WARN/ERROR messages to catch it. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Throwing a hard error is pretty draconian, though… > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> - Why not use config parameters instead of an > >>> interface? > >>>>> As > >>>>>>>>>>>>>>>> explained > >>>>>>>>>>>>>>>>>> in > >>>>>>>>>>>>>>>>>>>> the “Rejected Alternatives” section, we assume that > >>> the > >>>>>>>>>>>>> handler > >>>>>>>>>>>>>>>> will > >>>>>>>>>>>>>>>>> be > >>>>>>>>>>>>>>>>>>>> used for a greater number of exceptions in the > >> future. > >>>>>>>>>>>>>> Defining a > >>>>>>>>>>>>>>>>>>>> configuration parameter for each exception may make > >>> the > >>>>>>>>>>>>>>>>> configuration a > >>>>>>>>>>>>>>>>>>> bit > >>>>>>>>>>>>>>>>>>>> messy. Moreover, the handler offers more > >> flexibility. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Agreed that the logic-via-configuration approach is > >>> weird > >>>>>>>> and > >>>>>>>>>>>>>>>> limiting. > >>>>>>>>>>>>>>>>>>> Forget I ever suggested it ;) > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> I’d think additional background in the Motivation > >>> section > >>>>>>>>> would > >>>>>>>>>>>>>>> help > >>>>>>>>>>>>>>>> me > >>>>>>>>>>>>>>>>>>> understand how users might use this feature beyond a) > >>>>>>>> skipping > >>>>>>>>>>>>>>>>>> “oversized” > >>>>>>>>>>>>>>>>>>> records, and b) not retrying missing topics. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> Small change: > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> -ProductionExceptionHandlerResponse -> Response for > >>>>> brevity > >>>>>>>>>>>>> and > >>>>>>>>>>>>>>>>>>> simplicity. > >>>>>>>>>>>>>>>>>>>> Could’ve been HandlerResponse too I think! > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> The name change sounds good to me. > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>> Thanks Alieh! > >>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> I thank you all again for your useful > >>>>>>>> questions/suggestions. > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> I would be happy to hear more of your concerns, as > >>>> stated > >>>>>>>> in > >>>>>>>>>>>>>> some > >>>>>>>>>>>>>>>>>>> feedback. > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> Cheers, > >>>>>>>>>>>>>>>>>>>> Alieh > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>> On Wed, Apr 24, 2024 at 12:31 AM Justine Olshan > >>>>>>>>>>>>>>>>>>>> <jols...@confluent.io.invalid> wrote: > >>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> Thanks Alieh for the updates. > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> I'm a little concerned about the design pattern > >> here. > >>>> It > >>>>>>>>>>>>> seems > >>>>>>>>>>>>>>>> like > >>>>>>>>>>>>>>>>> we > >>>>>>>>>>>>>>>>>>> want > >>>>>>>>>>>>>>>>>>>>> specific usages, but we are packaging it as a > >> generic > >>>>>>>>>>>>> handler. > >>>>>>>>>>>>>>>>>>>>> I think we tried to narrow down on the specific > >>> errors > >>>> we > >>>>>>>>>>>>> want > >>>>>>>>>>>>>>> to > >>>>>>>>>>>>>>>>>>> handle, > >>>>>>>>>>>>>>>>>>>>> but it feels a little clunky as we have a generic > >>> thing > >>>>>>>> for > >>>>>>>>>>>>>> two > >>>>>>>>>>>>>>>>>> specific > >>>>>>>>>>>>>>>>>>>>> errors. > >>>>>>>>>>>>>>>>>>>>> > >>>>>>>>>>>>>>>>>>>>> I'm wondering if we are using the right patterns to > >>>> solve > >>>>>>>>>>>>>> these > >>>>>>>>>>>>>>>>>>> problems. I > >>>>>>>>>>>>>>>>>>>>> agree though that we will need something more than > >>> the > >>>>>>>> error > >>>>>>>>>>>>>>>> classes > >>>>>>>>>>>>>>>>>> I'm > >>>>>>>>>>>>>>>>>>>>> proposing if we want to have different handling be > >>>>>>>>>>>>>> configurable. > >>>>>>>>>>>>>>>>>>>>> My concern is that the open-endedness of a handler > >>>> means > >>>>>>>>>>>>> that > >>>>>>>>>>>>>> we > >>>>>>>>>>>>>>>> are > >>>>>>>>>>>>>>>>>>>>> creating more problems than we are solving. It is > >>> still > >>>>>>>>>>>>>> unclear > >>>>>>>>>>>>>>> to > >>>>>>>>>>>>>>>>> me > >>>>>>>>>>>>>>>>>>> how > >>>>>>>>>>>>>>>>>>>>> we expect to handle the errors. Perhaps we could > >>>> include > >>>>>>>> an > >>>>>>>>>>>>>>>> example? > >>>>>>>>>>>>>>>>>> It > >>>>>>>>>>>>>>>>>>>>> seems like there is a specific use case in mind and > >>>> maybe > >>>>>>>> we > >>>>>>>>>>>>>> can > > >
