Hi Artem,
Yes, I completely agree that by default, special action shouldn't be
required from users to prevent transactions from being committed when one
or more records can't be sent. The behavior I was suggesting was only
relevant to the new API we're discussing where we allow users to
intentionally bypass that logic when invoking commitTransaction.
Cheers,
Chris
On Tue, Jun 25, 2024, 01:44 Artem Livshits <alivsh...@confluent.io.invalid>
wrote:
> Hey folks,
>
> Great discussion!
>
> Re: throwing exceptions from send(). send() is documented to throw
> KafkaException, so if the application doesn't handle it, it should be a
> bug. Now, it does have a note that API exceptions wouldn't be thrown, not
> sure if we have code that relies on that. There is a reason exceptions
> have classes, they are designed to express a "class of errors" that can be
> handled, so that we don't have to add a flag or a new method every time we
> have a new exception to throw. But if there is consensus that it's still
> too risky (especially if we have examples of code that gets broken), then I
> agree that we shouldn't do it.
>
> Re: various ways to communicate semantics change. If we must have 2
> different behaviors, I think passing options to "ignore errors" to
> commitTransaction is probably the most intuitive way to do it. What I
> really don't like (in any of the options), is that we cannot really
> document it in a way that articulates a value in the product. There are
> tons of nuances that require understanding some buggy behavior, then fixed
> behavior, then an option to sometimes turn on buggy behavior, and etc.
>
> > if a user invokes Producer::abortTransaction from within a producer
> callback today
>
> I think we would get invalid state exception. Which we could probably fix,
> but even if we supported it, I think it would be good if doing send +
> commit would lead to aborted transaction without special action from the
> application -- the simple things should be really simple, any failure
> during send or commit should abort send + commit sequence without special
> handling.
>
> -Artem
>
> On Mon, Jun 24, 2024 at 6:37 PM Chris Egerton <fearthecel...@gmail.com>
> wrote:
>
> > One quick thought: if a user invokes Producer::abortTransaction from
> within
> > a producer callback today, even in the midst of an ongoing call to
> > Producer::commitTransaction, what is the behavior? Would it be reasonable
> > to support this behavior as a way to allow error handling to take place
> > during implicit flushes, via producer callback?
> >
> > On Mon, Jun 24, 2024 at 9:21 PM Matthias J. Sax <mj...@apache.org>
> wrote:
> >
> > > My point it, that it does not seem to be safe to allow users to ignore
> > > errors with an implicit flush, and I think it's better to only allow it
> > > with (ie, after) an explicit flush().
> > >
> > > My reasoning is, that users should make a decision to ignore errors or
> > > not, before calling `commitTx()`, but after inspecting all potential
> > > send errors. With an implicit flush, users need to "blindly" decide to
> > > ignore send errors, because there are pending sends and potential
> errors
> > > are not known yet, when calling `commitTx()`.
> > >
> > >
> > >
> > > > In the documentation of commitTransaction, we say if any send throws
> an
> > > > error, commitTransaction will too.
> > >
> > > Yes. And I think we should keep it this way for an implicit flush. With
> > > an explicit flush, `commitTransaction()` cannot encounter any send
> > > errors any longer.
> > >
> > >
> > >
> > > > It says that all callbacks will be executed, but we ignore the errors
> > of
> > > > the callbacks.
> > >
> > > Ah. Thanks for pointing this out. For this case it's even worse (for
> > > case (2)), because the user cannot inspect any errors and make any
> > > decision to ignore or not during an implicit flush...
> > >
> > >
> > >
> > > > We shouldn't be relying on errors in the callback unless we are
> > > > calling flush, which we can still do. It seems this has always been
> the
> > > > case as well.
> > >
> > > Yes, has always been this way, and my point is to keep it this way
> > > (option (2) would change it), and not start to allow to ignore errors
> > > with an implicit flush.
> > >
> > >
> > >
> > > -Matthias
> > >
> > >
> > >
> > > On 6/24/24 4:57 PM, Justine Olshan wrote:
> > > > Transaction verification is a concept from KIP-890 referring to the
> > > > verification that a partition has been added to the transaction. It's
> > > not a
> > > > huge deal, but maybe we don't want to overload the terminology.
> > > >
> > > > For option 2, I was a little confused by this
> > > >
> > > >> when commitTx is called, there is still pending Futures and not
> > > > all Callbacks are executed yet -- with the implicit flush, we know
> that
> > > > all Callbacks are executed, but even for this case, the user could
> only
> > > > throw an exception inside the Callback to stop the TX to eventually
> > > > commit -- Futures cannot be used to make a decision to ignore error
> and
> > > > commit or not.
> > > >
> > > > In the documentation of commitTransaction, we say if any send throws
> an
> > > > error, commitTransaction will too.
> > > >
> > > > *Further, if any of the {@link #send(ProducerRecord)} calls which
> were
> > > part
> > > > of the transaction hit irrecoverable errors, this method will throw
> the
> > > > last received exception immediately and the transaction will not be
> > > > committed.*
> > > >
> > > > It says that all callbacks will be executed, but we ignore the errors
> > of
> > > > the callbacks.
> > > >
> > > > *If the transaction is committed successfully and this method returns
> > > > without throwing an exception, it is guaranteed that all {@link
> > Callback
> > > > callbacks} for records in the transaction will have been invoked and
> > > > completed. Note that exceptions thrown by callbacks are ignored; the
> > > > producer proceeds to commit the transaction in any case.*
> > > >
> > > > Is it fair to say though that for the send errors, we can choose to
> > > ignore
> > > > them? II wasn't understanding where the callbacks come in with your
> > > > comment. We shouldn't be relying on errors in the callback unless we
> > are
> > > > calling flush, which we can still do. It seems this has always been
> the
> > > > case as well.
> > > >
> > > > Justine
> > > >
> > > > On Mon, Jun 24, 2024 at 11:07 AM Andrew Schofield <
> > > andrew_schofi...@live.com>
> > > > wrote:
> > > >
> > > >> Agreed. Options 1 and 3 are safe. Option 2 is not. I’d be happy with
> > 3a
> > > as
> > > >> the way.
> > > >>
> > > >> I suggest “TRANSACTION VERIFIED”.
> > > >>
> > > >> There isn’t really precedent for options in the producer API. We
> could
> > > use
> > > >> an enum,
> > > >> which is easy to use and not very future-proof. Or we could use a
> > class
> > > >> like the
> > > >> admin API does, which is cumbersome and flexible.
> > > >>
> > > >> CommitTransactionOptions.TRANSACTION_VERIFIED
> > > >>
> > > >> or
> > > >>
> > > >> public class CommitTransactionOptions {
> > > >> public CommitTransactionOptions();
> > > >>
> > > >> CommitTransactionOptions transactionVerified(boolean
> > > >> transactionVerified);
> > > >>
> > > >> boolean transactionVerified();
> > > >> }
> > > >>
> > > >>
> > > >> Then 3b is:
> > > >>
> > > >> send(…)
> > > >> send(…)
> > > >> flush()
> > > >> commitTransaction(new
> > > >> CommitTransactionOptions().transactionVerified(true))
> > > >>
> > > >>
> > > >> I’d tend towards the enum here because I doubt we need as much
> > > flexibility
> > > >> as the admin API requires.
> > > >>
> > > >> Thanks,
> > > >> Andrew
> > > >>
> > > >>
> > > >>> On 24 Jun 2024, at 18:39, Matthias J. Sax <mj...@apache.org>
> wrote:
> > > >>>
> > > >>> I am ok either way (ie, flush or commit), but I think we need to
> > define
> > > >> exact semantics, and I think there is some subtle thing to consider:
> > > >>>
> > > >>>
> > > >>>
> > > >>> 1) flush(Options)
> > > >>>
> > > >>> Example:
> > > >>>
> > > >>> send(...)
> > > >>> send(...)
> > > >>>
> > > >>> flush(ignoreErrors)
> > > >>>
> > > >>> // at this point, we know that all Futures are completed and all
> > > >> Callbacks are executed, and we can assume that all user code
> checking
> > > for
> > > >> errors did execute, before `commitTx` is called
> > > >>>
> > > >>> // I consider this option as safe
> > > >>>
> > > >>> commitTx()
> > > >>>
> > > >>>
> > > >>> 2) commitTx(Option)
> > > >>>
> > > >>> Example:
> > > >>>
> > > >>> send(...)
> > > >>> send(...)
> > > >>>
> > > >>> // when commitTx is called, there is still pending Futures and
> not
> > > all
> > > >> Callbacks are executed yet -- with the implicit flush, we know that
> > all
> > > >> Callbacks are executed, but even for this case, the user could only
> > > throw
> > > >> an exception inside the Callback to stop the TX to eventually commit
> > --
> > > >> Futures cannot be used to make a decision to ignore error and commit
> > or
> > > not.
> > > >>>
> > > >>> // I consider this option not as safe
> > > >>>
> > > >>> commitTx(igrnoreErrors)
> > > >>>
> > > >>>
> > > >>>
> > > >>> 3a) required flush + commitTx(Option)
> > > >>>
> > > >>> Example:
> > > >>>
> > > >>> send(...)
> > > >>> send(...)
> > > >>>
> > > >>> flush()
> > > >>>
> > > >>> // at this point, we know that all Future are completed and all
> > > >> Callbacks are executed, and we can assume that all user code
> checking
> > > for
> > > >> error did execute, before `commitTx` is called
> > > >>>
> > > >>> // I consider this option as safe
> > > >>>
> > > >>> commitTx(ignoreErrors)
> > > >>>
> > > >>>
> > > >>> 3b) missing flush + commitTx(Option)
> > > >>>
> > > >>> Example:
> > > >>>
> > > >>> send(...)
> > > >>> send(...)
> > > >>>
> > > >>> // as flush() was not called explicitly, we should ignore
> > > >> `ignoreErrors` flag and always throw an exception if the producer is
> > in
> > > >> error state, because we cannot be sure that the user did all
> required
> > > check
> > > >> for error handling
> > > >>>
> > > >>> commitTx(ignoreErrors)
> > > >>>
> > > >>>
> > > >>>
> > > >>> The only issue with option (3) is, that it's more complex and
> > semantics
> > > >> are more subtle. But it might be the a good (necessary?) bridge
> > between
> > > (1)
> > > >> and (2): (3) is semantically sound (we ignore errors via passing a
> > flag
> > > >> into commitTx() instead of flush()), and at the same time safe (we
> > force
> > > >> users to explicitly flush() and [hopefully] do proper error
> handling,
> > > and
> > > >> don't rely in am implicit flush() during commitTx() which might be
> > error
> > > >> prone).
> > > >>>
> > > >>> (Also need to find a good and descriptive name for the flag we pass
> > > into
> > > >> `commitTx()` for this case.)
> > > >>>
> > > >>>
> > > >>> -Matthias
> > > >>>
> > > >>>
> > > >>>
> > > >>> On 6/24/24 8:51 AM, Andrew Schofield wrote:
> > > >>>> Hi Chris,
> > > >>>> That works for me too. I slightly prefer an option on flush(), but
> > > what
> > > >> you suggested
> > > >>>> works too.
> > > >>>> Thanks,
> > > >>>> Andrew
> > > >>>>> On 24 Jun 2024, at 15:14, Chris Egerton <chr...@aiven.io.INVALID
> >
> > > >> wrote:
> > > >>>>>
> > > >>>>> Hi Andrew,
> > > >>>>>
> > > >>>>> I like a lot of what you said, but I still believe it's better to
> > > >> override
> > > >>>>> commitTransaction than flush. Users will already have to manually
> > opt
> > > >> in to
> > > >>>>> ignoring errors encountered during transactions, and we can
> > document
> > > >>>>> recommended usage (i.e., explicitly invoking flush() before
> > invoking
> > > >>>>> commitTransaction(ignoreRecordErrors)) in the newly-introduced
> > > method.
> > > >> I
> > > >>>>> don't believe it's worth the increased cognitive load on users
> with
> > > >>>>> non-transactional producers to introduce an overloaded flush()
> > > variant.
> > > >>>>>
> > > >>>>> Cheers,
> > > >>>>>
> > > >>>>> Chris
> > > >>>>>
> > > >>>>> On Mon, Jun 24, 2024 at 9:39 AM Andrew Schofield <
> > > >> andrew_schofi...@live.com>
> > > >>>>> wrote:
> > > >>>>>
> > > >>>>>> Hi Alieh,
> > > >>>>>> Thanks for driving this. Unfortunately, there are many parts of
> > the
> > > >> API
> > > >>>>>> which
> > > >>>>>> are a bit unfortunate and it’s tricky to make small improvements
> > > that
> > > >>>>>> don’t have
> > > >>>>>> downsides.
> > > >>>>>>
> > > >>>>>> I don’t like the idea of using a configuration because
> > configuration
> > > >> is
> > > >>>>>> often
> > > >>>>>> outside the application and changing the behaviour of someone
> > else’s
> > > >>>>>> application
> > > >>>>>> without understanding it is risky. Anything which embeds a
> > > >> transactional
> > > >>>>>> producer
> > > >>>>>> could have its behaviour changed unexpectedly.
> > > >>>>>>
> > > >>>>>> It would be been much nicer if send() didn’t fail silently and
> > > change
> > > >> the
> > > >>>>>> transaction
> > > >>>>>> state. But, because it’s an asynchronous operation, I don’t
> really
> > > >> think
> > > >>>>>> we can
> > > >>>>>> just make it throw all exceptions, even though I really think
> that
> > > >>>>>> `send()` is the
> > > >>>>>> method with the problem here.
> > > >>>>>>
> > > >>>>>> The contract of `flush()` is that it makes sure that all
> preceding
> > > >> sends
> > > >>>>>> will have
> > > >>>>>> completed, so it should be true that a well written application
> > > would
> > > >> be
> > > >>>>>> able to
> > > >>>>>> know which records were OK because of the Future<RecordMetadata>
> > > >> returned
> > > >>>>>> by the `send()` method. It should be able to determine whether
> it
> > > >> wants to
> > > >>>>>> commit
> > > >>>>>> the transaction even if some of the intended operations didn’t
> > > >> succeed.
> > > >>>>>>
> > > >>>>>> What we don’t currently have is a way for the application to say
> > to
> > > >> the
> > > >>>>>> KafkaProducer
> > > >>>>>> that it knows the outcome of sending the records and to confirm
> > that
> > > >> it
> > > >>>>>> wants to proceed.
> > > >>>>>> Then it would not be necessary for `commitTransaction()` to
> throw
> > an
> > > >>>>>> exception to
> > > >>>>>> report a historical error which the application might choose to
> > > >> ignore.
> > > >>>>>>
> > > >>>>>> Having read the comments, I think the KIP is on the right lines
> > > >> focusing
> > > >>>>>> on the `flush()`
> > > >>>>>> method. My suggestion is that we introduce an option on
> `flush()`
> > to
> > > >> be
> > > >>>>>> used before
> > > >>>>>> `commitTransaction()` for applications that want to be able to
> > > commit
> > > >>>>>> transactions which
> > > >>>>>> had known failed operations.
> > > >>>>>>
> > > >>>>>> The code would be:
> > > >>>>>>
> > > >>>>>> producer.beginTransaction();
> > > >>>>>>
> > > >>>>>> future1 = producer.send(goodRecord1);
> > > >>>>>> future2 = producer.send(badRecord); // The future from this
> > call
> > > >> will
> > > >>>>>> complete exceptionally
> > > >>>>>> future3 = producer.send(goodRecord2);
> > > >>>>>>
> > > >>>>>> producer.flush(FlushOption.TRANSACTION_READY);
> > > >>>>>>
> > > >>>>>> // At this point, we know that all 3 futures are complete and
> > the
> > > >>>>>> transaction contains 2 records
> > > >>>>>> producer.commitTransaction();
> > > >>>>>>
> > > >>>>>> I wouldn’t deprecate `flush()` with no option. It just uses the
> > > >> default
> > > >>>>>> option which behaves
> > > >>>>>> like today.
> > > >>>>>>
> > > >>>>>> Why did I suggest an option on `flush()` rather than
> > > >>>>>> `commitTransaction()`? Because with
> > > >>>>>> `flush()`, it’s clear when the application is stating that it’s
> > seen
> > > >> all
> > > >>>>>> of the results from its
> > > >>>>>> `send()` calls and it’s ready to proceed. If it has to rely on
> > > >> flushing
> > > >>>>>> that occurs inside
> > > >>>>>> `commitTransaction()`, I don’t see it’s as clear-cut.
> > > >>>>>>
> > > >>>>>> Thanks,
> > > >>>>>> Andrew
> > > >>>>>>
> > > >>>>>>
> > > >>>>>>
> > > >>>>>>> On 24 Jun 2024, at 13:44, Alieh Saeedi
> > > <asae...@confluent.io.INVALID
> > > >>>
> > > >>>>>> wrote:
> > > >>>>>>>
> > > >>>>>>> Hi all,
> > > >>>>>>> Thanks for the interesting discussion.
> > > >>>>>>>
> > > >>>>>>> I assume that now the main questions are as follows:
> > > >>>>>>>
> > > >>>>>>> 1. Do we need to transit the transcation to the error state for
> > API
> > > >>>>>>> exceptions?
> > > >>>>>>> 2. Should we throw the API exception in `send()` instead of
> > > >> returning a
> > > >>>>>>> future error?
> > > >>>>>>> 3. If the answer to question (1) is NO and to question (2) is
> > YES,
> > > >> do we
> > > >>>>>>> need to change the current `flush` or `commitTnx` at all?
> > > >>>>>>>
> > > >>>>>>> Cheers,
> > > >>>>>>> Alieh
> > > >>>>>>>
> > > >>>>>>> On Sat, Jun 22, 2024 at 3:21 AM Matthias J. Sax <
> > mj...@apache.org>
> > > >>>>>> wrote:
> > > >>>>>>>
> > > >>>>>>>> Hey Kirk,
> > > >>>>>>>>
> > > >>>>>>>> can you elaborate on a few points?
> > > >>>>>>>>
> > > >>>>>>>>> Otherwise users would have to know to explicitly change their
> > > code
> > > >> to
> > > >>>>>>>> invoke flush().
> > > >>>>>>>>
> > > >>>>>>>> Why? If we would add an option to `flush(FlushOption)`, the
> > > existing
> > > >>>>>>>> `flush()` w/o any option will still be there, right? If we
> would
> > > >> really
> > > >>>>>>>> deprecate existing `flush()`, it would just mean that we would
> > > pass
> > > >>>>>>>> "default FlushOption" into an implicit flush (and yes, we
> would
> > > >> need to
> > > >>>>>>>> define what this would be).
> > > >>>>>>>>
> > > >>>>>>>> I think there is no clear winner (as pointed out in my last
> > > reply),
> > > >> and
> > > >>>>>>>> both `flush(FlushOption)` and `commitTx(CommitOption)` has
> > > >> advantages
> > > >>>>>>>> and drawbacks. Guess we need to just agree on which tradeoff
> we
> > > >> want to
> > > >>>>>>>> move forward with?
> > > >>>>>>>>
> > > >>>>>>>>
> > > >>>>>>>> Not sure if your database example is a 1:1 fit? I think, the
> > > better
> > > >>>>>>>> comparison would be:
> > > >>>>>>>>
> > > >>>>>>>> BEGIN TX;
> > > >>>>>>>> INSERT INTO foo VALUES (’a’);
> > > >>>>>>>> INSERT INTO foo VALUES (’b’);
> > > >>>>>>>> INSERT INTO foo VALUES (’c’);
> > > >>>>>>>> INSERT INTO foo VALUES (’not sure’);
> > > >>>>>>>>
> > > >>>>>>>> For this case, the full TX would roll back, right? I still
> think
> > > >> that
> > > >>>>>>>> allowing users to just skip over the last error, and continue
> > the
> > > TX
> > > >>>>>>>> would be ok. In the end, we provide a programmatic API, and
> not
> > a
> > > >>>>>>>> declarative one as SQL. Of course, default behavior would
> still
> > be
> > > >> to
> > > >>>>>>>> put the producer into error state, and the user would need to
> > call
> > > >>>>>>>> `abortTransaction()` to move forward.
> > > >>>>>>>>
> > > >>>>>>>>
> > > >>>>>>>> -Matthias
> > > >>>>>>>>
> > > >>>>>>>> On 6/21/24 5:26 PM, Kirk True wrote:
> > > >>>>>>>>> Hi Matthias,
> > > >>>>>>>>>
> > > >>>>>>>>>> On Jun 21, 2024, at 12:28 PM, Matthias J. Sax <
> > mj...@apache.org
> > > >
> > > >>>>>> wrote:
> > > >>>>>>>>>>
> > > >>>>>>>>>> If we want to limit it to `RecordTooLargeException` throwing
> > > from
> > > >>>>>>>> `send()` directly make sense. Thanks for calling it out.
> > > >>>>>>>>>>
> > > >>>>>>>>>> It's still a question of backward compatibility? `send()`
> does
> > > >> throw
> > > >>>>>>>> exceptions already, including generic `KafkaException`. Not
> sure
> > > if
> > > >> this
> > > >>>>>>>> helps with backward compatibility? Could we just add a new
> > > exception
> > > >>>>>> type
> > > >>>>>>>> (which is a child of `KafkaException`)?
> > > >>>>>>>>>>
> > > >>>>>>>>>> The Producer JavaDocs are not totally explicit about it
> IMHO.
> > > >>>>>>>>>>
> > > >>>>>>>>>> I think we could expect that some generic error handling
> path
> > > gets
> > > >>>>>>>> executed. For the TX-case, I would assume that a TX would be
> > > >> aborted if
> > > >>>>>>>> `send()` throws or that the producer would be `closed()`.
> > Overall
> > > >> this
> > > >>>>>>>> might be safe?
> > > >>>>>>>>>>
> > > >>>>>>>>>>> It would be a little less flexible
> > > >>>>>>>>>>>> though, since (as you note) it would still be impossible
> to
> > > >> commit
> > > >>>>>>>>>>>> transactions after errors have been reported from brokers.
> > > >>>>>>>>>>
> > > >>>>>>>>>> KS would still need a way to clear the error state of the
> > > >> producer. We
> > > >>>>>>>> could catch a `RecordTooLargeException` from `send()`, call
> the
> > > >> handler
> > > >>>>>> and
> > > >>>>>>>> let it decide what to do next. But if it does return
> `CONTINUE`
> > to
> > > >>>>>> swallow
> > > >>>>>>>> the error and drop the poison pill record on the floor, we
> would
> > > >> want to
> > > >>>>>>>> move forward and commit the transaction.
> > > >>>>>>>>>>
> > > >>>>>>>>>> But the question is: if we cannot add a record to the tx,
> does
> > > the
> > > >>>>>>>> producer need to go into error state? In the end, we did throw
> > and
> > > >>>>>> inform
> > > >>>>>>>> the app that the record was _not_ added, and it's up to the
> app
> > to
> > > >>>>>> decide
> > > >>>>>>>> what to do next?
> > > >>>>>>>>>
> > > >>>>>>>>> That’s an excellent question…
> > > >>>>>>>>>
> > > >>>>>>>>> Imagine the user’s application is writing information to a
> > > database
> > > >>>>>>>> instead of Kafka. If there’s a table with a CHAR(1) column and
> > > this
> > > >> SQL
> > > >>>>>>>> statement was attempted, what should happen?
> > > >>>>>>>>>
> > > >>>>>>>>> INSERT INTO foo VALUES (’not sure’);
> > > >>>>>>>>>
> > > >>>>>>>>> Yes, that DML would fail, sure, but would the user expect
> that
> > > the
> > > >>>>>>>> connection used by database library would get stuck in some
> kind
> > > of
> > > >>>>>> error
> > > >>>>>>>> state? A user would be able catch the error and either
> continue
> > or
> > > >>>>>> abort,
> > > >>>>>>>> based on their business rules.
> > > >>>>>>>>>
> > > >>>>>>>>> So I agree with what I believe you’re implying: we shouldn’t
> > > >> poison the
> > > >>>>>>>> Producer/TransactionManager on certain types of
> > application-level
> > > >>>>>> errors in
> > > >>>>>>>> send().
> > > >>>>>>>>>
> > > >>>>>>>>> Kirk
> > > >>>>>>>>>
> > > >>>>>>>>>> If we report the error only via the `Callback` it's a
> > different
> > > >> story,
> > > >>>>>>>> because the contract for this case is clearly specified on the
> > > >> JavaDocs:
> > > >>>>>>>>>>
> > > >>>>>>>>>>> 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>. If any of
> > the
> > > >> send
> > > >>>>>>>> calls failed with an irrecoverable error,
> > > >>>>>>>>>>> the final {@link #commitTransaction()} call will fail and
> > throw
> > > >> the
> > > >>>>>>>> exception from the last failed send. When
> > > >>>>>>>>>>> this happens, your application should call {@link
> > > >>>>>> #abortTransaction()}
> > > >>>>>>>> to reset the state and continue to send
> > > >>>>>>>>>>> data.
> > > >>>>>>>>>>
> > > >>>>>>>>>>
> > > >>>>>>>>>>
> > > >>>>>>>>>> -Matthias
> > > >>>>>>>>>>
> > > >>>>>>>>>>
> > > >>>>>>>>>> On 6/21/24 11:42 AM, Chris Egerton wrote:
> > > >>>>>>>>>>> Hi Artem,
> > > >>>>>>>>>>> I think it'd make sense to throw directly from send
> whenever
> > > >>>>>> possible,
> > > >>>>>>>>>>> instead of returning an already-completed future. I didn't
> do
> > > >> that in
> > > >>>>>>>> my
> > > >>>>>>>>>>> bug fix to try to be conservative about breaking changes
> but
> > > this
> > > >>>>>>>> seems to
> > > >>>>>>>>>>> have caused its own set of headaches. It would be a little
> > less
> > > >>>>>>>> flexible
> > > >>>>>>>>>>> though, since (as you note) it would still be impossible to
> > > >> commit
> > > >>>>>>>>>>> transactions after errors have been reported from brokers.
> > > >>>>>>>>>>> I'll leave it up to the Kafka Streams folks to decide if
> that
> > > >>>>>>>> flexibility
> > > >>>>>>>>>>> is required. If it is, then users could explicitly call
> > flush()
> > > >>>>>> before
> > > >>>>>>>>>>> committing (and ignoring errors for) or aborting a
> > transaction,
> > > >> if
> > > >>>>>> they
> > > >>>>>>>>>>> want to implement fine-grained error handling logic such as
> > > >> allowing
> > > >>>>>>>> errors
> > > >>>>>>>>>>> for a subset of topics to be ignored.
> > > >>>>>>>>>>> Hi Matthias,
> > > >>>>>>>>>>> Most of the time you're right and we can't throw from
> send();
> > > >>>>>> however,
> > > >>>>>>>> in
> > > >>>>>>>>>>> this case (client-side record-too-large exception), the
> error
> > > is
> > > >>>>>>>> actually
> > > >>>>>>>>>>> noticed by the producer before send() returns, so it should
> > be
> > > >>>>>>>> possible to
> > > >>>>>>>>>>> throw directly.
> > > >>>>>>>>>>> Cheers,
> > > >>>>>>>>>>> Chris
> > > >>>>>>>>>>> On Fri, Jun 21, 2024, 14:25 Matthias J. Sax <
> > mj...@apache.org>
> > > >>>>>> wrote:
> > > >>>>>>>>>>>> Not sure if we can change send and make it throw, given
> that
> > > >> send()
> > > >>>>>> is
> > > >>>>>>>>>>>> async? That is why users can register a `Callback` to
> begin
> > > >> with,
> > > >>>>>>>> right?
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>> And Alieh's point about backward compatibility is also a
> > fair
> > > >>>>>> concern.
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>>> Actually, this would potentially be even
> > > >>>>>>>>>>>>> worse than the original buggy behavior because the bug
> was
> > > >> that we
> > > >>>>>>>>>>>> ignored
> > > >>>>>>>>>>>>> errors that happened in the "send()" method itself, not
> > > >> necessarily
> > > >>>>>>>> the
> > > >>>>>>>>>>>>> ones that we got from the broker.
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>> My understanding was that `commitTx(swallowError)` would
> > only
> > > >>>>>> swallow
> > > >>>>>>>>>>>> `send()` errors, not errors about the actually commit. I
> > agree
> > > >> that
> > > >>>>>> it
> > > >>>>>>>>>>>> would be very bad to swallow errors about the actual tx
> > > >> commit...
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>> It's a fair question if this might be too subtle; to make
> it
> > > >>>>>> explicit,
> > > >>>>>>>>>>>> we could use `CommitOpions#ignorePendingSendErors()`
> > [working
> > > >> name]
> > > >>>>>> to
> > > >>>>>>>>>>>> make it clear.
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>> If we think it's too subtle to change commit to swallow
> > send()
> > > >>>>>> errors,
> > > >>>>>>>>>>>> maybe going with `flush(FlushOptions)` would be clearer
> (and
> > > we
> > > >> can
> > > >>>>>>>> use
> > > >>>>>>>>>>>> `FlushOption#swallowSendErrorsForTransactions()` [working
> > > name]
> > > >> to
> > > >>>>>> be
> > > >>>>>>>>>>>> explicitly that the `FlushOption` for now has only an
> effect
> > > for
> > > >>>>>> TX).
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>> Thoughts?
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>> -Matthias
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>>
> > > >>>>>>>>>>>> On 6/21/24 4:10 AM, Alieh Saeedi wrote:
> > > >>>>>>>>>>>>> Hi all,
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> It is very exciting to see all the experts here raising
> > very
> > > >> good
> > > >>>>>>>> points.
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> As we go further, we see more and more options to improve
> > our
> > > >>>>>>>> solution,
> > > >>>>>>>>>>>>> which makes concluding and updating the KIP impossible.
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> The main suggestions so far are:
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> 1. `flush` with `flushOptions` as input parameter
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> 2. `commitTx` with `commitOptions` as input parameter
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> 3. `send` must throw the exception
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> My concern about the 3rd suggestion:
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> 1. Does the change cause any issue with backward
> > > compatibility?
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> 2. The `send (bad record)` already transits the
> transaction
> > > to
> > > >> the
> > > >>>>>>>> error
> > > >>>>>>>>>>>>> state. No user, including Streams is able to transit the
> > > >>>>>> transaction
> > > >>>>>>>> back
> > > >>>>>>>>>>>>> from the error state. Do you mean we remove the
> > > >>>>>>>>>>>>> `maybeTransitionToErrorState(e)` from here
> > > >>>>>>>>>>>>> <
> > > >>>>>>>>>>>>
> > > >>>>>>>>
> > > >>>>>>
> > > >>
> > >
> >
> https://github.com/apache/kafka/blob/9b5b434e2a6b2d5290ea403fc02859b1c523d8aa/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java#L1112
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> as well?
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> Cheers,
> > > >>>>>>>>>>>>> Alieh
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>> On Fri, Jun 21, 2024 at 8:45 AM Andrew Schofield <
> > > >>>>>>>>>>>> andrew_schofi...@live.com>
> > > >>>>>>>>>>>>> wrote:
> > > >>>>>>>>>>>>>
> > > >>>>>>>>>>>>>> Hi Artem,
> > > >>>>>>>>>>>>>> I think you make a good point which is worth further
> > > >>>>>> consideration.
> > > >>>>>>>> If
> > > >>>>>>>>>>>>>> any of the existing methods is really ripe for a change
> > > here,
> > > >> it’s
> > > >>>>>>>> the
> > > >>>>>>>>>>>>>> send() that actually caused the problem. If that can be
> > > fixed
> > > >> so
> > > >>>>>>>> there
> > > >>>>>>>>>>>> are
> > > >>>>>>>>>>>>>> no situations in which a lurking error breaks a
> > transaction,
> > > >> that
> > > >>>>>>>> might
> > > >>>>>>>>>>>> be
> > > >>>>>>>>>>>>>> the best.
> > > >>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>> Thanks,
> > > >>>>>>>>>>>>>> Andrew
> > > >>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>> On 21 Jun 2024, at 01:51, Artem Livshits <
> > > >> alivsh...@confluent.io
> > > >>>>>>>>>>>> .INVALID>
> > > >>>>>>>>>>>>>> wrote:
> > > >>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>> I thought we still wait for requests (and their
> errors)
> > to
> > > >> come
> > > >>>>>>>> in and
> > > >>>>>>>>>>>>>>> could handle fatal errors appropriately.
> > > >>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>> We do wait for requests, but my understanding is that
> > when
> > > >>>>>>>>>>>>>>> commitTransaction("ignore send errors") we want to
> ignore
> > > >> errors.
> > > >>>>>>>> So
> > > >>>>>>>>>>>> if
> > > >>>>>>>>>>>>>> we
> > > >>>>>>>>>>>>>>> do
> > > >>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>> 1. send
> > > >>>>>>>>>>>>>>> 2. commitTransaction("ignore send errors")
> > > >>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>> the commit will succeed. You can look at the example
> in
> > > >>>>>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-9279 and
> > just
> > > >>>>>>>> substitute
> > > >>>>>>>>>>>>>>> commitTransaction with commitTransaction("ignore send
> > > >> errors")
> > > >>>>>> and
> > > >>>>>>>> we
> > > >>>>>>>>>>>> get
> > > >>>>>>>>>>>>>>> the buggy behavior back :-). Actually, this would
> > > >> potentially be
> > > >>>>>>>> even
> > > >>>>>>>>>>>>>>> worse than the original buggy behavior because the bug
> > was
> > > >> that
> > > >>>>>> we
> > > >>>>>>>>>>>>>> ignored
> > > >>>>>>>>>>>>>>> errors that happened in the "send()" method itself, not
> > > >>>>>>>> necessarily the
> > > >>>>>>>>>>>>>>> ones that we got from the broker.
> > > >>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>> Actually, looking at
> > > >>>>>>>> https://github.com/apache/kafka/pull/11508/files,
> > > >>>>>>>>>>>>>>> wouldn't a better solution be to just throw the error
> > from
> > > >> the
> > > >>>>>>>> "send"
> > > >>>>>>>>>>>>>>> method itself, rather than trying to set it to be
> thrown
> > > >> during
> > > >>>>>>>> commit?
> > > >>>>>>>>>>>>>>> This way the example in
> > > >>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-9279
> > > >>>>>>>>>>>>>>> would be fixed, and at the same time it would give an
> > > >> opportunity
> > > >>>>>>>> for
> > > >>>>>>>>>>>> KS
> > > >>>>>>>>>>>>>> to
> > > >>>>>>>>>>>>>>> catch the error and ignore it if needed. Not sure if
> we
> > > >> need a
> > > >>>>>>>> KIP for
> > > >>>>>>>>>>>>>>> that, just do a better fix of the old bug.
> > > >>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>> -Artem
> > > >>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>> On Thu, Jun 20, 2024 at 4:58 PM Justine Olshan
> > > >>>>>>>>>>>>>> <jols...@confluent.io.invalid>
> > > >>>>>>>>>>>>>>> wrote:
> > > >>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>> I'm a bit late to the party, but the discussion here
> > looks
> > > >>>>>>>> reasonable.
> > > >>>>>>>>>>>>>>>> Moving the logic to a transactional method makes sense
> > to
> > > >> me and
> > > >>>>>>>> makes
> > > >>>>>>>>>>>>>> me
> > > >>>>>>>>>>>>>>>> feel a bit better about keeping the complexity in the
> > > >> methods
> > > >>>>>>>> relevant
> > > >>>>>>>>>>>>>> to
> > > >>>>>>>>>>>>>>>> the issue.
> > > >>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>> One minor concern is that if we set "ignore send
> > > >>>>>>>>>>>>>>>> errors" (or whatever we decide to name it) option
> > without
> > > >>>>>> explicit
> > > >>>>>>>>>>>>>> flush,
> > > >>>>>>>>>>>>>>>> it'll actually lead to broken behavior as the
> > application
> > > >> won't
> > > >>>>>> be
> > > >>>>>>>>>>>> able
> > > >>>>>>>>>>>>>> to
> > > >>>>>>>>>>>>>>>> stop a commit from proceeding even on fatal errors.
> > > >>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>> Is this with respect to the case a request is still
> > > inflight
> > > >>>>>> when
> > > >>>>>>>> we
> > > >>>>>>>>>>>>>> call
> > > >>>>>>>>>>>>>>>> commitTransaction? I thought we still wait for
> requests
> > > (and
> > > >>>>>> their
> > > >>>>>>>>>>>>>> errors)
> > > >>>>>>>>>>>>>>>> to come in and could handle fatal errors
> appropriately.
> > > >>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>> Justine
> > > >>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>> On Thu, Jun 20, 2024 at 4:32 PM Artem Livshits
> > > >>>>>>>>>>>>>>>> <alivsh...@confluent.io.invalid> wrote:
> > > >>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>> Hi Matthias (and other folks who suggested ideas),
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>> maybe `commitTransaction(CommitOptions)` or similar
> > > could
> > > >> be a
> > > >>>>>>>> good
> > > >>>>>>>>>>>>>>>> way
> > > >>>>>>>>>>>>>>>>> forward?
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>> I like this approach. One minor concern is that if
> we
> > > set
> > > >>>>>>>> "ignore
> > > >>>>>>>>>>>> send
> > > >>>>>>>>>>>>>>>>> errors" (or whatever we decide to name it) option
> > without
> > > >>>>>>>> explicit
> > > >>>>>>>>>>>>>> flush,
> > > >>>>>>>>>>>>>>>>> it'll actually lead to broken behavior as the
> > application
> > > >> won't
> > > >>>>>>>> be
> > > >>>>>>>>>>>> able
> > > >>>>>>>>>>>>>>>> to
> > > >>>>>>>>>>>>>>>>> stop a commit from proceeding even on fatal errors.
> > But
> > > I
> > > >>>>>> guess
> > > >>>>>>>>>>>> we'll
> > > >>>>>>>>>>>>>>>> just
> > > >>>>>>>>>>>>>>>>> have to clearly document it.
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>> In some way we are basically adding a flag to
> > optionally
> > > >>>>>> restore
> > > >>>>>>>> the
> > > >>>>>>>>>>>>>>>>> https://issues.apache.org/jira/browse/KAFKA-9279
> bug,
> > > >> which is
> > > >>>>>>>> the
> > > >>>>>>>>>>>>>>>>> motivation for all these changes, anyway :-).
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>> -Artem
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>> On Thu, Jun 20, 2024 at 2:18 PM Matthias J. Sax <
> > > >>>>>>>> mj...@apache.org>
> > > >>>>>>>>>>>>>>>> wrote:
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>> Seems the option to use a config does not get a lot
> of
> > > >>>>>> support.
> > > >>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>> So we need to go with some form or "overload / new
> > > >> method". I
> > > >>>>>>>> think
> > > >>>>>>>>>>>>>>>>>> Chris' point about not coupling it to `flush()` but
> > > rather
> > > >>>>>>>>>>>>>>>>>> `commitTransaction()` is actually a very good one;
> for
> > > >> non-tx
> > > >>>>>>>> case,
> > > >>>>>>>>>>>>>> the
> > > >>>>>>>>>>>>>>>>>> different flush variants would not make sense.
> > > >>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>> I also like Lianet's idea to pass in some "options"
> > > >> object, so
> > > >>>>>>>> maybe
> > > >>>>>>>>>>>>>>>>>> `commitTransaction(CommitOptions)` or similar could
> > be a
> > > >> good
> > > >>>>>>>> way
> > > >>>>>>>>>>>>>>>>>> forward? It's much better than a `boolean`
> parameter,
> > > >>>>>>>> aesthetically,
> > > >>>>>>>>>>>>>> as
> > > >>>>>>>>>>>>>>>>>> we as extendable in the future if necessary.
> > > >>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>> Given that we would pass in an optional parameter,
> we
> > > >> might
> > > >>>>>> not
> > > >>>>>>>> even
> > > >>>>>>>>>>>>>>>>>> need to deprecate the existing `commitTransaction()`
> > > >> method?
> > > >>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>> -Matthias
> > > >>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>> On 6/20/24 9:12 AM, Andrew Schofield wrote:
> > > >>>>>>>>>>>>>>>>>>> Hi Alieh,
> > > >>>>>>>>>>>>>>>>>>> Thanks for the KIP.
> > > >>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>> I *really* don’t like adding a config which changes
> > the
> > > >>>>>>>> behaviour
> > > >>>>>>>>>>>> of
> > > >>>>>>>>>>>>>>>>> the
> > > >>>>>>>>>>>>>>>>>>> flush() method. We already have too many configs.
> > But I
> > > >>>>>> totally
> > > >>>>>>>>>>>>>>>>>> understand
> > > >>>>>>>>>>>>>>>>>>> the problem that you’re trying to solve and some of
> > the
> > > >> other
> > > >>>>>>>>>>>>>>>>> suggestions
> > > >>>>>>>>>>>>>>>>>>> in this thread seem neater.
> > > >>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>> Personally, I would add another method to
> > > KafkaProducer.
> > > >> Not
> > > >>>>>> an
> > > >>>>>>>>>>>>>>>>> overload
> > > >>>>>>>>>>>>>>>>>>> on flush() because this is not flush() at all.
> Using
> > > >>>>>> Matthias’s
> > > >>>>>>>>>>>>>>>>> options,
> > > >>>>>>>>>>>>>>>>>>> I prefer (3).
> > > >>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>> Thanks,
> > > >>>>>>>>>>>>>>>>>>> Andrew
> > > >>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> On 20 Jun 2024, at 15:08, Lianet M. <
> > > liane...@gmail.com
> > > >>>
> > > >>>>>>>> wrote:
> > > >>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> Hi all, thanks for the KIP Alieh!
> > > >>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> LM1. Totally agree with Artem's point about the
> > config
> > > >> not
> > > >>>>>>>> being
> > > >>>>>>>>>>>> the
> > > >>>>>>>>>>>>>>>>>> most
> > > >>>>>>>>>>>>>>>>>>>> explicit/flexible way to express this capability.
> > > >> Getting
> > > >>>>>>>> then to
> > > >>>>>>>>>>>>>>>>>> Matthias
> > > >>>>>>>>>>>>>>>>>>>> 4 options, what I don't like about 3 and 4 is that
> > it
> > > >> seems
> > > >>>>>>>> they
> > > >>>>>>>>>>>>>>>> might
> > > >>>>>>>>>>>>>>>>>> not
> > > >>>>>>>>>>>>>>>>>>>> age very well? Aren't we going to be wanting some
> > > other
> > > >>>>>> twist
> > > >>>>>>>> to
> > > >>>>>>>>>>>> the
> > > >>>>>>>>>>>>>>>>>> flush
> > > >>>>>>>>>>>>>>>>>>>> semantics that will have us adding yet another
> param
> > > to
> > > >> it,
> > > >>>>>> or
> > > >>>>>>>>>>>>>>>> another
> > > >>>>>>>>>>>>>>>>>>>> overloaded method? I truly don't have the context
> to
> > > >> answer
> > > >>>>>>>> that,
> > > >>>>>>>>>>>>>>>> but
> > > >>>>>>>>>>>>>>>>>> if it
> > > >>>>>>>>>>>>>>>>>>>> feels like a realistic future maybe adding some
> kind
> > > >>>>>>>> FlushOptions
> > > >>>>>>>>>>>>>>>>>> params to
> > > >>>>>>>>>>>>>>>>>>>> the flush would be better from an extensibility
> > point
> > > of
> > > >>>>>>>> view. It
> > > >>>>>>>>>>>>>>>>> would
> > > >>>>>>>>>>>>>>>>>>>> only have the clearErrors option available for now
> > but
> > > >> could
> > > >>>>>>>>>>>> accept
> > > >>>>>>>>>>>>>>>>> any
> > > >>>>>>>>>>>>>>>>>>>> other we may need. I find that this would remove
> the
> > > >>>>>>>> "ugliness"
> > > >>>>>>>>>>>>>>>>> Matthias
> > > >>>>>>>>>>>>>>>>>>>> pointed out for 3. and 4.
> > > >>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> LM2. No matter how we end up expressing the
> > different
> > > >>>>>>>> semantics
> > > >>>>>>>>>>>> for
> > > >>>>>>>>>>>>>>>>>> flush,
> > > >>>>>>>>>>>>>>>>>>>> let's make sure we update the KIP on the flush and
> > > >>>>>>>>>>>> commitTransaction
> > > >>>>>>>>>>>>>>>>>> java
> > > >>>>>>>>>>>>>>>>>>>> docs. It currently states that flush "clears the
> > last
> > > >>>>>>>> exception"
> > > >>>>>>>>>>>>>>>> and
> > > >>>>>>>>>>>>>>>>>>>> commitTransaction "will NOT throw" if called after
> > > >> flush,
> > > >>>>>> but
> > > >>>>>>>> it
> > > >>>>>>>>>>>>>>>>> really
> > > >>>>>>>>>>>>>>>>>> all
> > > >>>>>>>>>>>>>>>>>>>> depends on the config/options/method used.
> > > >>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> LM3. I find it would be helpful to include an
> > example
> > > to
> > > >>>>>> show
> > > >>>>>>>> the
> > > >>>>>>>>>>>>>>>> new
> > > >>>>>>>>>>>>>>>>>> flow
> > > >>>>>>>>>>>>>>>>>>>> that we're unblocking (I see this as the great
> gain
> > > >> here):
> > > >>>>>>>> flush
> > > >>>>>>>>>>>>>>>> with
> > > >>>>>>>>>>>>>>>>>> clear
> > > >>>>>>>>>>>>>>>>>>>> error option enabled -> catch and do whatever
> error
> > > >> handling
> > > >>>>>>>> we
> > > >>>>>>>>>>>> want
> > > >>>>>>>>>>>>>>>>> ->
> > > >>>>>>>>>>>>>>>>>>>> commitTransaction successfully
> > > >>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> Thanks!
> > > >>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> Lianet
> > > >>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> On Wed, Jun 19, 2024 at 11:26 PM Chris Egerton <
> > > >>>>>>>>>>>>>>>>> fearthecel...@gmail.com
> > > >>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>> wrote:
> > > >>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>> Hi Matthias,
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>> I like the alternatives you've listed. One more
> > that
> > > >> might
> > > >>>>>>>> help
> > > >>>>>>>>>>>> is
> > > >>>>>>>>>>>>>>>>> if,
> > > >>>>>>>>>>>>>>>>>>>>> instead of overloading flush(), we overloaded
> > > >>>>>>>> commitTransaction()
> > > >>>>>>>>>>>>>>>> to
> > > >>>>>>>>>>>>>>>>>>>>> something like commitTransaction(boolean
> > > >>>>>>>> tolerateRecordErrors).
> > > >>>>>>>>>>>>>>>> This
> > > >>>>>>>>>>>>>>>>>> seems
> > > >>>>>>>>>>>>>>>>>>>>> slightly cleaner in that it takes the behavioral
> > > >> change we
> > > >>>>>>>> want,
> > > >>>>>>>>>>>>>>>>> which
> > > >>>>>>>>>>>>>>>>>> only
> > > >>>>>>>>>>>>>>>>>>>>> applies to transactional producers, to an API
> > method
> > > >> that
> > > >>>>>> is
> > > >>>>>>>> only
> > > >>>>>>>>>>>>>>>>> used
> > > >>>>>>>>>>>>>>>>>> for
> > > >>>>>>>>>>>>>>>>>>>>> transactional producers. It would also avoid the
> > > issue
> > > >> of
> > > >>>>>>>> whether
> > > >>>>>>>>>>>>>>>> or
> > > >>>>>>>>>>>>>>>>>> not
> > > >>>>>>>>>>>>>>>>>>>>> flush() (or a new variant of it with altered
> > > semantics)
> > > >>>>>>>> should
> > > >>>>>>>>>>>>>>>> throw
> > > >>>>>>>>>>>>>>>>> or
> > > >>>>>>>>>>>>>>>>>>>>> not. Thoughts?
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>> Hi Alieh,
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>> Thanks for the KIP, I like this direction a lot
> > more
> > > >> than
> > > >>>>>> the
> > > >>>>>>>>>>>>>>>>> pluggable
> > > >>>>>>>>>>>>>>>>>>>>> handler!
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>> I share Artem's concerns that enabling this
> > behavior
> > > >> via
> > > >>>>>>>>>>>>>>>>> configuration
> > > >>>>>>>>>>>>>>>>>>>>> doesn't seem like a great fit. It's likely that
> > > >> application
> > > >>>>>>>> code
> > > >>>>>>>>>>>>>>>> will
> > > >>>>>>>>>>>>>>>>>> be
> > > >>>>>>>>>>>>>>>>>>>>> written in a style that only works with one type
> of
> > > >>>>>> behavior
> > > >>>>>>>> from
> > > >>>>>>>>>>>>>>>>>>>>> transactional producers, so requiring that
> > > application
> > > >> code
> > > >>>>>>>> to
> > > >>>>>>>>>>>>>>>>> declare
> > > >>>>>>>>>>>>>>>>>> its
> > > >>>>>>>>>>>>>>>>>>>>> expectations for the behavior of its producer
> seems
> > > >> more
> > > >>>>>>>>>>>>>>>> appropriate
> > > >>>>>>>>>>>>>>>>>> than,
> > > >>>>>>>>>>>>>>>>>>>>> e.g., allowing users deploying that application
> to
> > > >> tweak a
> > > >>>>>>>>>>>>>>>>>> configuration
> > > >>>>>>>>>>>>>>>>>>>>> file that gets fed to producers spun up inside
> it.
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>> Cheers,
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>> Chris
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>> On Wed, Jun 19, 2024 at 10:32 PM Matthias J. Sax
> <
> > > >>>>>>>>>>>> mj...@apache.org
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>> wrote:
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>> Thanks for the KIP Alieh. I actually like the
> KIP
> > > >> as-is,
> > > >>>>>> but
> > > >>>>>>>>>>>> think
> > > >>>>>>>>>>>>>>>>>>>>>> Arthem raises very good points...
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>> Seems we have four options on how to move
> forward?
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>> 1. add config to allow "silent error
> clearance"
> > as
> > > >> the
> > > >>>>>>>> KIP
> > > >>>>>>>>>>>>>>>>> proposes
> > > >>>>>>>>>>>>>>>>>>>>>> 2. change flush() to clear error and let it
> > throw
> > > >>>>>>>>>>>>>>>>>>>>>> 3. add new flushAndThrow()` (or better name)
> > which
> > > >>>>>> clears
> > > >>>>>>>>>>>> error
> > > >>>>>>>>>>>>>>>>> and
> > > >>>>>>>>>>>>>>>>>>>>>> throws
> > > >>>>>>>>>>>>>>>>>>>>>> 4. add `flush(boolean clearAndThrow)` and let
> > user
> > > >> pick
> > > >>>>>>>> (and
> > > >>>>>>>>>>>>>>>>>> deprecate
> > > >>>>>>>>>>>>>>>>>>>>>> existing `flush()`)
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>> For (2), given that it would be a behavior
> change,
> > > we
> > > >>>>>> might
> > > >>>>>>>> also
> > > >>>>>>>>>>>>>>>>> need
> > > >>>>>>>>>>>>>>>>>> a
> > > >>>>>>>>>>>>>>>>>>>>>> public "feature flag" config.
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>> It seems, both (1) and (2) have the issue Artem
> > > >> mentioned.
> > > >>>>>>>> (3)
> > > >>>>>>>>>>>> and
> > > >>>>>>>>>>>>>>>>> (4)
> > > >>>>>>>>>>>>>>>>>>>>>> would be safer to this end, however, for both we
> > > >> kinda get
> > > >>>>>>>> an
> > > >>>>>>>>>>>> ugly
> > > >>>>>>>>>>>>>>>>>> API?
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>> Not sure right now if I have any preference.
> Seems
> > > we
> > > >> need
> > > >>>>>>>> to
> > > >>>>>>>>>>>> pick
> > > >>>>>>>>>>>>>>>>>> some
> > > >>>>>>>>>>>>>>>>>>>>>> evil and that there is no clear best solution?
> > Would
> > > >> be
> > > >>>>>>>> good to
> > > >>>>>>>>>>>>>>>> her
> > > >>>>>>>>>>>>>>>>>> from
> > > >>>>>>>>>>>>>>>>>>>>>> others what they think
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>> -Matthias
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>> On 6/18/24 8:39 PM, Artem Livshits wrote:
> > > >>>>>>>>>>>>>>>>>>>>>>> Hi Alieh,
> > > >>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>> Thank you for the KIP. I have a couple of
> > > >> suggestions:
> > > >>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>> AL1. We should throw an error from flush after
> > we
> > > >> clear
> > > >>>>>>>> it.
> > > >>>>>>>>>>>>>>>> This
> > > >>>>>>>>>>>>>>>>>>>>> would
> > > >>>>>>>>>>>>>>>>>>>>>>> make it so that both "send + commit" and "send
> +
> > > >> flush +
> > > >>>>>>>>>>>> commit"
> > > >>>>>>>>>>>>>>>>> (the
> > > >>>>>>>>>>>>>>>>>>>>>>> latter looks like just a more verbose way to
> > > express
> > > >> the
> > > >>>>>>>>>>>> former,
> > > >>>>>>>>>>>>>>>>> and
> > > >>>>>>>>>>>>>>>>>> it
> > > >>>>>>>>>>>>>>>>>>>>>>> would be intuitive if it behaves the same)
> would
> > > >> throw if
> > > >>>>>>>> the
> > > >>>>>>>>>>>>>>>>>>>>> transaction
> > > >>>>>>>>>>>>>>>>>>>>>>> has an error (so if the code is written either
> > way
> > > >> it's
> > > >>>>>>>> going
> > > >>>>>>>>>>>> be
> > > >>>>>>>>>>>>>>>>>>>>>> correct).
> > > >>>>>>>>>>>>>>>>>>>>>>> At the same time, the latter could be extended
> by
> > > the
> > > >>>>>>>> caller to
> > > >>>>>>>>>>>>>>>>>>>>> intercept
> > > >>>>>>>>>>>>>>>>>>>>>>> exceptions from flush, ignore as needed, and
> > commit
> > > >> the
> > > >>>>>>>>>>>>>>>>> transaction.
> > > >>>>>>>>>>>>>>>>>>>>>> This
> > > >>>>>>>>>>>>>>>>>>>>>>> solution would keep basic things simple (if
> > someone
> > > >> has
> > > >>>>>>>> code
> > > >>>>>>>>>>>> that
> > > >>>>>>>>>>>>>>>>>>>>> doesn't
> > > >>>>>>>>>>>>>>>>>>>>>>> require advanced error handling, then basic
> > "send +
> > > >>>>>> flush +
> > > >>>>>>>>>>>>>>>> commit"
> > > >>>>>>>>>>>>>>>>>>>>> would
> > > >>>>>>>>>>>>>>>>>>>>>>> do the right thing) and advanced things
> possible,
> > > an
> > > >>>>>>>>>>>> application
> > > >>>>>>>>>>>>>>>>> can
> > > >>>>>>>>>>>>>>>>>>>>> add
> > > >>>>>>>>>>>>>>>>>>>>>>> try + catch around flush and ignore some
> errors.
> > > >>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>> AL2. I'm not sure if config is the best way to
> > > >> express
> > > >>>>>> the
> > > >>>>>>>>>>>>>>>>>>>>> modification
> > > >>>>>>>>>>>>>>>>>>>>>> of
> > > >>>>>>>>>>>>>>>>>>>>>>> the "flush" semantics -- the application logic
> > that
> > > >> calls
> > > >>>>>>>>>>>> "flush"
> > > >>>>>>>>>>>>>>>>>> needs
> > > >>>>>>>>>>>>>>>>>>>>>> to
> > > >>>>>>>>>>>>>>>>>>>>>>> match the "flush" semantics and configuring
> > > >> semantics in
> > > >>>>>> a
> > > >>>>>>>>>>>>>>>> detached
> > > >>>>>>>>>>>>>>>>>>>>> place
> > > >>>>>>>>>>>>>>>>>>>>>>> creates a room for bugs due to discrepancies.
> > This
> > > >> can
> > > >>>>>> be
> > > >>>>>>>>>>>>>>>>> especially
> > > >>>>>>>>>>>>>>>>>>>>> bad
> > > >>>>>>>>>>>>>>>>>>>>>>> if the producer loads configuration from a file
> > at
> > > >> run
> > > >>>>>>>> time, in
> > > >>>>>>>>>>>>>>>>> that
> > > >>>>>>>>>>>>>>>>>>>>>> case a
> > > >>>>>>>>>>>>>>>>>>>>>>> mistake in configuration could break the
> > > application
> > > >>>>>>>> because it
> > > >>>>>>>>>>>>>>>> was
> > > >>>>>>>>>>>>>>>>>>>>>> written
> > > >>>>>>>>>>>>>>>>>>>>>>> to expect one "flush" semantics but the
> semantics
> > > is
> > > >>>>>>>> switched.
> > > >>>>>>>>>>>>>>>>> Given
> > > >>>>>>>>>>>>>>>>>>>>>> that
> > > >>>>>>>>>>>>>>>>>>>>>>> the "flush" semantics needs to match the
> caller's
> > > >>>>>>>> expectation,
> > > >>>>>>>>>>>> a
> > > >>>>>>>>>>>>>>>>> way
> > > >>>>>>>>>>>>>>>>>> to
> > > >>>>>>>>>>>>>>>>>>>>>>> accomplish that would be to pass the caller's
> > > >> expectation
> > > >>>>>>>> to
> > > >>>>>>>>>>>> the
> > > >>>>>>>>>>>>>>>>>>>>> "flush"
> > > >>>>>>>>>>>>>>>>>>>>>>> call by either have a method with a different
> > name
> > > or
> > > >>>>>> have
> > > >>>>>>>> an
> > > >>>>>>>>>>>>>>>>>> overload
> > > >>>>>>>>>>>>>>>>>>>>>> with
> > > >>>>>>>>>>>>>>>>>>>>>>> a Boolen flag that would configure the
> semantics
> > > (the
> > > >>>>>>>> current
> > > >>>>>>>>>>>>>>>>> method
> > > >>>>>>>>>>>>>>>>>>>>>> could
> > > >>>>>>>>>>>>>>>>>>>>>>> just redirect to the new one).
> > > >>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>> -Artem
> > > >>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>> On Mon, Jun 17, 2024 at 9:09 AM Alieh Saeedi
> > > >>>>>>>>>>>>>>>>>>>>>> <asae...@confluent.io.invalid>
> > > >>>>>>>>>>>>>>>>>>>>>>> wrote:
> > > >>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>>> Hi all,
> > > >>>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>>> I'd like to kick off a discussion for KIP-1059
> > > that
> > > >>>>>>>> suggests
> > > >>>>>>>>>>>>>>>>> adding
> > > >>>>>>>>>>>>>>>>>> a
> > > >>>>>>>>>>>>>>>>>>>>>> new
> > > >>>>>>>>>>>>>>>>>>>>>>>> feature to the Producer flush() method.
> > > >>>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>>>
> > > >>>>>>>>>>>>>>
> > > >>>>>>>>>>>>
> > > >>>>>>>>
> > > >>>>>>
> > > >>
> > >
> >
> <https://cwiki.apache.org/confluence/display/KAFKA/KIP-1059%3A+Enable+the+Producer+flush%28%29+method+to+clear+the+latest+send%28%29+error>
