Thanks for your clarification on 2)/3), that makes sense. On Tue, Jan 19, 2021 at 10:16 AM Boyang Chen <reluctanthero...@gmail.com> wrote:
> Thanks for the input Guozhang, replied inline. > > On Mon, Jan 18, 2021 at 8:57 PM Guozhang Wang <wangg...@gmail.com> wrote: > > > Hello Boyang, > > > > Thanks for the updated KIP. I read it again and have the following > > thoughts: > > > > 0. I'm a bit concerned that if commitTxn does not throw any non-fatal > > exception, and instead we rely on the subsequent beginTxn call to throw, > it > > may violate some callers with a pattern that relying on commitTxn to > > succeed to make some non-rollback operations. For example: > > > > beginTxn() > > // do some read-write on my local DB > > commitTxn() > > // if commitTxn succeeds, then commit the DB > > > > ------------- > > > > The issue is that, committing DB is a non-rollback operation, and users > may > > just rely on commitTxn to return without error to make this non-rollback > > call. Of course we can just claim this pattern is not legitimate and is > not > > the right way of doing things, but many users may naturally adopt this > > pattern. > > > > So maybe we should still let commitTxn also throw non-fatal exceptions, > in > > which case we would then call abortTxn again. > > > > But if we do this, the pattern becomes: > > > > try { > > beginTxn() > > // do something > > } catch (Exception) { > > shouldCommit = false; > > } > > > > if (shouldCommit) { > > try { > > commitTxn() > > } catch (...) { // enumerate all fatal exceptions > > shutdown() > > } catch (KafkaException) { > > // non-fatal > > shouldCommit = false; > > } > > } > > > > if (!shouldCommit && running()) { > > try { > > abortTxn() > > } catch (KafkaException) { > > // only throw fatal > > shutdown() > > } > > } > > > > --------------------- > > > > Which is much more complicated. > > > > Thank makes me think, the alternative we have discussed offline may be > > better: let commitTxn() to return a boolean flag. > > > > * If it returns true, it means the commit succeeded. Users can > comfortably > > continue and do any external non-rollback operations if they like. > > * If it returns false, it means the commit failed with non-fatal error > *AND > > the txn has been aborted*. Users do not need to call abortTxn again. > > * If it throws, then it means fatal errors. Users should shut down the > > client. > > > > In this case, the pattern becomes: > > > > try { > > beginTxn() > > // do something > > } catch (Exception) { > > shouldCommit = false; > > } > > > > try { > > if (shouldCommit) { > > commitSucceeded = commitTxn() > > } else { > > // reset offsets, rollback operations, etc > > abortTxn() > > } > > } catch (KafkaException) { > > // only throw fatal > > shutdown() > > } > > > > if (commitSucceeded) > > // do other non-rollbackable things > > else > > // reset offsets, rollback operations, etc > > > > ------------------------- > > > > Of course, if we want to have better visibility into what caused the > commit > > to fail and txn to abort. We can let the return type be an enum instead > of > > a flag. But my main idea is to still let the commitTxn be the final point > > users can tell whether this txn succeeded or not, instead of relying on > the > > next beginTxn() call. > > > > I agree that adding a boolean flag is indeed useful in this case. Will > update the KIP. > > 1. Re: "while maintaining the behavior to throw fatal exception in raw" not > > sure what do you mean by "throw" here. Are you proposing the callback > would > > *pass* (not throw) in any fatal exceptions when triggered without > wrapping? > > > > Yes, I want to say *pass*, the benefit is to make the end user's > expectation consistent > regarding exception handling. > > > > 2. I'm not sure I understand the claim regarding the callback that "In > EOS > > setup, it is not required to handle the exception". Are you proposing > that, > > e.g. in Streams, we do not try to handle any exceptions if EOS is enabled > > in the callback anymore, but just let commitTxn() itself to fail to be > > notified about the problem? Personally I think in Streams we should just > > make the handling logic of the callback to be consistent regardless of > the > > EOS settings (today we have different logic depending on this logic, > which > > I think could be unified as well). > > > > My idea originates from the claim on send API: > "When used as part of a transaction, it is not necessary to define a > callback or check the result of the future in order to detect errors from > <code>send</code>. " > My understanding is that for EOS, the exception will be detected by calling > transactional APIs either way, so it is a duplicate handling to track > all the sendExceptions in RecordCollector. However, I looked up > sendException is being used today as follow: > > 1. Pass to "ProductionExceptionHandler" for any default or customized > exception handler to handle > 2. Stop collecting offset info or new exceptions > 3. Check and rethrow exceptions in close(), flush() or new send() calls > > To my understanding, we should still honor the commitment to #1 for any > user customized implementation. The #2 does not really affect our decision > upon EOS. The #3 here is still valuable as it could help us fail fast in > new send() instead of waiting to later stage of processing. In that sense, > I agree we should continue to record send exceptions even under EOS case to > ensure the strength of stream side Producer logic. On the safer side, we no > longer need to wrap certain fatal exceptions like ProducerFenced as > TaskMigrated, since they should not crash the stream thread if thrown in > raw format, once we adopt the new processing model in the send phase. > > > > > > Guozhang > > > > > > > > On Thu, Dec 17, 2020 at 8:42 PM Boyang Chen <reluctanthero...@gmail.com> > > wrote: > > > > > Thanks for everyone's feedback so far. I have polished the KIP after > > > offline discussion with some folks working on EOS to make the exception > > > handling more lightweight. The essential change is that we are not > > > inventing a new intermediate exception type, but instead separating a > > full > > > transaction into two phases: > > > > > > 1. The data transmission phase > > > 2. The commit phase > > > > > > This way for any exception thrown from phase 1, will be an indicator in > > > phase 2 whether we should do commit or abort, and from now on > > > `commitTransaction` should only throw fatal exceptions, similar to > > > `abortTransaction`, so that any KafkaException caught in the commit > phase > > > will be definitely fatal to crash the app. For more advanced users such > > as > > > Streams, we have the ability to further wrap selected types of fatal > > > exceptions to trigger task migration if necessary. > > > > > > More details in the KIP, feel free to take another look, thanks! > > > > > > On Thu, Dec 17, 2020 at 8:36 PM Boyang Chen < > reluctanthero...@gmail.com> > > > wrote: > > > > > > > Thanks Bruno for the feedback. > > > > > > > > On Mon, Dec 7, 2020 at 5:26 AM Bruno Cadonna <br...@confluent.io> > > wrote: > > > > > > > >> Thanks Boyang for the KIP! > > > >> > > > >> Like Matthias, I do also not know the producer internal well enough > to > > > >> comment on the categorization. However, I think having a super > > exception > > > >> (e.g. RetriableException) that encodes if an exception is fatal or > > not > > > >> is cleaner, better understandable and less error-prone, because > > ideally > > > >> when you add a new non-fatal exception in future you just need to > > think > > > >> about letting it inherit from the super exception and all the rest > of > > > >> the code will just behave correctly without the need to wrap the new > > > >> exception into another exception each time it is thrown (maybe it is > > > >> thrown at different location in the code). > > > >> > > > >> As far as I understand the following statement from your previous > > e-mail > > > >> is the reason that currently such a super exception is not possible: > > > >> > > > >> "Right now we have RetriableException type, if we are going to add a > > > >> `ProducerRetriableException` type, we have to put this new interface > > as > > > >> the parent of the RetriableException, because not all thrown > non-fatal > > > >> exceptions are `retriable` in general for producer" > > > >> > > > >> > > > >> In the list of exceptions in your KIP, I found non-fatal exceptions > > that > > > >> do not inherit from RetriableException. I guess those are the ones > you > > > >> are referring to in your statement: > > > >> > > > >> InvalidProducerEpochException > > > >> InvalidPidMappingException > > > >> TransactionAbortedException > > > >> > > > >> All of those exceptions are non-fatal and do not inherit from > > > >> RetriableException. Is there a reason for that? If they depended > from > > > >> RetriableException we would be a bit closer to a super exception I > > > >> mention above. > > > >> > > > >> The reason is that sender may catch those exceptions in the > > > > ProduceResponse, and it currently does infinite > > > > retries on RetriableException. To make sure we could trigger the > > > > abortTransaction() by catching non-fatal thrown > > > > exceptions and reinitialize the txn state, we chose not to let those > > > > exceptions inherit RetriableException so that > > > > they won't cause infinite retry on sender. > > > > > > > > > > > >> With OutOfOrderSequenceException and UnknownProducerIdException, I > > think > > > >> to understand that their fatality depends on the type (i.e. > > > >> configuration) of the producer. That makes it difficult to have a > > super > > > >> exception that encodes the retriability as mentioned above. Would it > > be > > > >> possible to introduce exceptions that inherit from > RetriableException > > > >> and that are thrown when those exceptions are caught from the > brokers > > > >> and the type of the producer is such that the exceptions are > > retriable? > > > >> > > > >> Yea, I think in general the exception type mixing is difficult to > get > > > > them all right. I have already proposed another solution based on my > > > > offline discussion with some folks working on EOS > > > > to make the handling more straightforward for end users without the > > need > > > > to distinguish exception fatality. > > > > > > > >> Best, > > > >> Bruno > > > >> > > > >> > > > >> On 04.12.20 19:34, Guozhang Wang wrote: > > > >> > Thanks Boyang for the proposal! I made a pass over the list and > here > > > are > > > >> > some thoughts: > > > >> > > > > >> > 0) Although this is not part of the public API, I think we should > > make > > > >> sure > > > >> > that the suggested pattern (i.e. user can always call abortTxn() > > when > > > >> > handling non-fatal errors) are indeed supported. E.g. if the txn > is > > > >> already > > > >> > aborted by the broker side, then users can still call abortTxn > which > > > >> would > > > >> > not throw another exception but just be treated as a no-op. > > > >> > > > > >> > 1) *ConcurrentTransactionsException*: I think this error can also > be > > > >> > returned but not documented yet. This should be a non-fatal error. > > > >> > > > > >> > 2) *InvalidTxnStateException*: this error is returned from broker > > when > > > >> txn > > > >> > state transition failed (e.g. it is trying to transit to > > > complete-commit > > > >> > while the current state is not prepare-commit). This error could > > > >> indicates > > > >> > a bug on the client internal code or the broker code, OR a user > > error > > > >> --- a > > > >> > similar error is ConcurrentTransactionsException, i.e. if Kafka is > > > >> bug-free > > > >> > these exceptions would only be returned if users try to do > something > > > >> wrong, > > > >> > e.g. calling abortTxn right after a commitTxn, etc. So I'm > thinking > > it > > > >> > should be a non-fatal error instead of a fatal error, wdyt? > > > >> > > > > >> > 3) *KafkaException*: case i "indicates fatal transactional > sequence > > > >> > (Fatal)", this is a bit conflicting with the > > *OutOfSequenceException* > > > >> that > > > >> > is treated as non-fatal. I guess your proposal is that > > > >> > OutOfOrderSequenceException would be treated either as fatal with > > > >> > transactional producer, or non-fatal with idempotent producer, is > > that > > > >> > right? If the producer is only configured with idempotency but not > > > >> > transaction, then throwing a TransactionStateCorruptedException > for > > > >> > non-fatal errors would be confusing since users are not using > > > >> transactions > > > >> > at all.. So I suggest we always throw OutOfSequenceException as-is > > > (i.e. > > > >> > not wrapped) no matter how the producer is configured, and let the > > > >> caller > > > >> > decide how to handle it based on whether it is only idempotent or > > > >> > transactional itself. > > > >> > > > > >> > 4) Besides all the txn APIs, the `send()` callback / future can > also > > > >> throw > > > >> > txn-related exceptions, I think this KIP should also cover this > API > > as > > > >> well? > > > >> > > > > >> > 5) This is related to 1/2) above: sometimes those non-fatal errors > > > like > > > >> > ConcurrentTxn or InvalidTxnState are not due to the state being > > > >> corrupted > > > >> > at the broker side, but maybe users are doing something wrong. So > > I'm > > > >> > wondering if we should further distinguish those non-fatal errors > > > >> between > > > >> > a) those that are caused by Kafka itself, e.g. a broker timed out > > and > > > >> > aborted a txn and later an endTxn request is received, and b) the > > > user's > > > >> > API call pattern is incorrect, causing the request to be rejected > > with > > > >> an > > > >> > error code from the broker. *TransactionStateCorruptedException* > > feels > > > >> to > > > >> > me more like for case a), but not case b). > > > >> > > > > >> > > > > >> > Guozhang > > > >> > > > > >> > > > > >> > On Wed, Dec 2, 2020 at 4:50 PM Boyang Chen < > > > reluctanthero...@gmail.com> > > > >> > wrote: > > > >> > > > > >> >> Thanks Matthias, I think your proposal makes sense as well, on > the > > > pro > > > >> side > > > >> >> we could have a universally agreed exception type to be caught by > > the > > > >> user, > > > >> >> without having an extra layer on top of the actual exceptions. I > > > could > > > >> see > > > >> >> some issue on downsides: > > > >> >> > > > >> >> 1. The exception hierarchy will be more complex. Right now we > have > > > >> >> RetriableException type, if we are going to add a > > > >> >> `ProducerRetriableException` type, we have to put this new > > interface > > > >> as the > > > >> >> parent of the RetriableException, because not all thrown > non-fatal > > > >> >> exceptions are `retriable` in general for producer, for example > > > >> >> < > > > >> >> > > > >> > > > > > > https://github.com/apache/kafka/blob/e275742f850af4a1b79b0d1bd1ac9a1d2e89c64e/clients/src/main/java/org/apache/kafka/clients/producer/internals/Sender.java#L745 > > > >> >>> . > > > >> >> We could have an empty interface `ProducerRetriableException` to > > let > > > >> all > > > >> >> the thrown exceptions implement for sure, but it's a bit > unorthodox > > > in > > > >> the > > > >> >> way we deal with exceptions in general. > > > >> >> > > > >> >> 2. There are cases where we throw a KafkaException wrapping > another > > > >> >> KafkaException as either fatal or non-fatal. If we use an > interface > > > to > > > >> >> solve #1, it is also required to implement another bloated > > exception > > > >> class > > > >> >> which could replace KafkaException type, as we couldn't mark > > > >> KafkaException > > > >> >> as retriable for sure. > > > >> >> > > > >> >> 3. In terms of the encapsulation, wrapping means we could limit > the > > > >> scope > > > >> >> of affection to the producer only, which is important since we > > don't > > > >> want > > > >> >> shared exception types to implement a producer-related interface, > > > such > > > >> >> as UnknownTopicOrPartitionException. > > > >> >> > > > >> >> Best, > > > >> >> Boyang > > > >> >> > > > >> >> On Wed, Dec 2, 2020 at 3:38 PM Matthias J. Sax <mj...@apache.org > > > > > >> wrote: > > > >> >> > > > >> >>> Thanks for the KIP Boyang! > > > >> >>> > > > >> >>> Overall, categorizing exceptions makes a lot of sense. As I > don't > > > know > > > >> >>> the producer internals well enough, I cannot comment on the > > > >> >>> categorization in detail though. > > > >> >>> > > > >> >>> What I am wondering is, if we should introduce an exception > > > interface > > > >> >>> that non-fatal exception would implement instead of creating a > new > > > >> class > > > >> >>> that will wrap non-fatal exceptions? What would be the pros/cons > > for > > > >> >>> both designs? > > > >> >>> > > > >> >>> > > > >> >>> -Matthias > > > >> >>> > > > >> >>> > > > >> >>> On 12/2/20 11:35 AM, Boyang Chen wrote: > > > >> >>>> Hey there, > > > >> >>>> > > > >> >>>> I would like to start a discussion thread for KIP-691: > > > >> >>>> > > > >> >>> > > > >> >> > > > >> > > > > > > https://cwiki.apache.org/confluence/display/KAFKA/KIP-691%3A+Enhance+Transactional+Producer+Exception+Handling > > > >> >>>> > > > >> >>>> The KIP is aiming to simplify the exception handling logic for > > > >> >>>> transactional Producer users by classifying fatal and non-fatal > > > >> >>> exceptions > > > >> >>>> and throw them correspondingly for easier catch and retry. Let > me > > > >> know > > > >> >>> what > > > >> >>>> you think. > > > >> >>>> > > > >> >>>> Best, > > > >> >>>> Boyang > > > >> >>>> > > > >> >>> > > > >> >> > > > >> > > > > >> > > > > >> > > > > > > > > > > > > > -- > > -- Guozhang > > > -- -- Guozhang