Hi,
Looking at your suggestions for the CommitOptions, I would be happy with either.
I definitely prefer to the Map<String, ?> in the KIP.
We also need to think about the other option where CLEAR_SEND_ERRORS hasn’t been
specified, and leave ourselves space for other options in the future.
If we use an enum, we need to give the existing behaviour a name. Maybe:
public enum CommitOptions {
/**
* Commits the ongoing transaction, flushing any unsent records before
* actually committing the transaction. If any of the records sent in
this transaction
* hit unrecoverable errors, the transaction will not be committed.
*/
NONE,
/**
* Commits the ongoing transaction, first clearing any errors from
records already sent
* in this transaction and then flushing any unsent records before
committing the transaction.
* If there are any unsent records flushed by this operation which hit
unrecoverable errors,
* these errors will not be cleared and the transaction will not be
committed.
* <p>
* To ensure there are no unsent records, you must call {@link #flush()}
before
* committing the transaction.
*/
CLEAR_SEND_ERRORS;
}
The Javadoc for commitTransaction will have to be careful crafted.
Thanks,
Andrew
> On 26 Jun 2024, at 03:23, Matthias J. Sax <mj...@apache.org> wrote:
>
>>> I was also curious about this text:
>>>> The new method clears the latest error produced by `send(ProducerRecord)`
>>> and transits the transaction back from the error state
>
> I agree. We should not say "latest" but "any".
>
>
>
>> Is it fair to say that we expect to encounter only one send response with
>> an error that we clear on commitTransaction? Is that solving the problem?
>
> Yes, I think that is an accurate description.
>
>
>> Speaking of documentation, is it confusing that the name of the KIP is no
>> longer consistent with the approach the KIP takes?
>
> Sounds like an easy fix (we should not update the subject of the discuss
> email thread thought...)
>
>
>
>> looks like we said on the KIP that the options will be of the type
>> Map<String,
>> ?> commitOptions. Would we want to define something more specific?
>
> I would also suggest to either pass an enum (as preferred by Andrew) or a
> newly define class `CommitOptions`.
>
> Something like:
>
> public enum CommitOptions {
> /**
> * Cleans any previous send errors, before committing the transaction.
> * Note, if there are pending sends, error won't be cleared.
> * To avoid pending sends, you need to call {@link #flush()} before
> committing the transaction.
> */
> CLEAR_SEND_ERRORS
> }
>
>
> Or similar for a class:
>
> public class CommitOptions {
> /**
> * Cleans any previous send errors, before committing the transaction.
> * Note, if there are pending sends, error won't be cleared.
> * To avoid pending sends, you need to call {@link #flush()} before
> committing the transaction.
> */
> public static CommitOptions clearSendErrors();
> }
>
> Plus:
>
> public class KafkaProducer {
>
> /**
> * <same JavaDocs as for commitTransaction()>
> *
> * This method should only be called if there are no pending writes,
> ie, only after calling {@link #flush()).
> * If there are any errors sending messaged to topics, these errors
> can be cleared by passing {@link CommitOptions#clearSendErrors}, allowing the
> transaction to commit even in case of data loss during.
> *
> * If this method is used while there are pending sends,
> * send error cannot be cleared.
> */
> public void commitTransaction(CommitOptions options);
> }
>
> Or something like this -- we can discuss details about JavaDocs on the PR
> IMHO.
>
>
>> 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.
>
> I would believe that users don't need to understand the full history of
> events, and the propose JavaDocs should go a long way.
>
>
>
> -Matthias
>
>
> On 6/25/24 5:00 PM, Justine Olshan wrote:
>> Hey there,
>> I had a few questions about the update.
>> Looks like we said on the KIP that the options will be of the type
>> Map<String,
>> ?> commitOptions. Would we want to define something more specific?
>> I was also curious about this text:
>>> The new method clears the latest error produced by `send(ProducerRecord)`
>> and transits the transaction back from the error state
>> Is it fair to say that we expect to encounter only one send response with
>> an error that we clear on commitTransaction? Is that solving the problem?
>> I think we also need to be really careful about the documentation of this
>> method. It should be clear that setting this option will not do anything if
>> there is any inflight record when the method is called.
>> Speaking of documentation, is it confusing that the name of the KIP is no
>> longer consistent with the approach the KIP takes?
>> Thanks,
>> Justine
>> On Tue, Jun 25, 2024 at 5:08 AM Alieh Saeedi <asae...@confluent.io.invalid>
>> wrote:
>>> Hi all,
>>>
>>> Appreciation for maintaining the momentum of our discussion.
>>>
>>>
>>> I see kinda consensus over the main points. It seems that we agreed on the
>>> following:
>>>
>>> 1) Define the `commitTnx(commitOptions)` to clear the error.
>>>
>>> 2) Make the user explicitly call `flush()` before
>>> `commitTnx(commitOptions)`, if he determines ignoring errors.
>>>
>>>
>>> I updated the KIP with the above-mentioned points. Please take a look. I am
>>> sure it is not perfect yet, and there are/will be some open questions, but
>>> if you agree, I will open voting as well. Of course, the discussion can
>>> still carry on in this thread.
>>>
>>>
>>> Cheers,
>>>
>>> Alieh
>>>
>>> On Tue, Jun 25, 2024 at 11:36 AM Chris Egerton <fearthecel...@gmail.com>
>>> wrote:
>>>
>>>> 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
>>>>>
>>>>
>>>
