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
>>>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>>>>>>>> Cheers,
>>>>>>>>>>>>>>>>>>>>>> Alieh
