Hi Greg,
I appreciate your concerns and comprehensive answer.
I am not sure whether I fully understood what you meant or not. You mean,
at the end, the user can go for one of the following scenarios: Either
1) `beginTxn()` and `send(record)` and `commitTxn()` or
2) `beginTxn()` and `prepare(record)` and `send(prepared_record)` and
`commitTxn()` ?
Of course, the `send` in scenario 1 is different from the one in scenario
2, since a part of the second one 's job has been done during
`prepare()`ing.
Cheers,
Alieh
On Sat, Jul 20, 2024 at 1:20 AM Greg Harris <greg.har...@aiven.io.invalid>
wrote:
> Hi Artem and Matthias,
>
> > On the other hand, the effort to prove that
> > keeping all records in memory won't break some scenarios (and generally
> > breaking one is enough to cause a lot of pain) seems to be significantly
> > higher than to prove that setting some flag in some API has pretty much 0
> > chance of regression
>
> > in the end, why buffer records twice?
>
> > This way we don't
> > ignore the error, we're just changing the method they are delivered.
>
> > Very clean semantics
> > which should also address the concern of "non-atomic tx"
>
> I feel like my concerns are being minimized instead of being addressed
> in this discussion, and if that's because I'm not expressing them clearly,
> I apologize.
>
> Many users come to Kafka with prior expectations, especially when we use
> industry-standard terminology like 'Exactly Once Semantics",
> "Transactions", "Commit", "Abort". Of course Kafka isn't an ACID-compliant
> database, but users will evaluate, discuss, and develop applications with
> Kafka through the lens of the ACID principles, because that is the
> framework most commonly applied to transactional semantics.
> The original design of KIP-98 [1] explicitly mentions atomic commits (with
> the same meaning as the A in ACID) as the primary abstraction being added
> (reproduced here):
>
> > At the core, transactional guarantees enable applications to produce to
> multiple TopicPartitions atomically, ie. all writes to these
> TopicPartitions will succeed or fail as a unit.
> > Further, since consumer progress is recorded as a write to the offsets
> topic, the above capability is leveraged to enable applications to batch
> consumed and produced messages into a single atomic unit, ie. a set of
> messages may be considered consumed only if the entire
> ‘consume-transform-produce’ executed in its entirety.
>
> I think it's important to say that to a user, "writes" really means "send()
> and commitOffsets() calls", not literal produce requests to Kafka brokers,
> and "consume-transform-produce" really means "poll(), transform, send()".
> This is because to a user, the implementation within poll() and send() and
> the broker are none of their concern, and are intended to be within the
> abstraction.
> When I say that this feature is a non-atomic commit, I mean that this
> feature does not fit the above description, and breaks the transaction
> abstraction in a meaningful way. No longer do all writes succeed or fail as
> a unit, some failures are permitted to drop data. No longer must a
> consume-transform-produce cycle be executed in its entirety, some parts may
> be left incomplete.
> This means that this method will be difficult to define ("which exceptions
> are covered?"), difficult to document ("how do we explain
> 'not-really-atomic commits' clearly and unambiguously to a potential
> user?"), and difficult to compose ("if someone turns this option on, how
> does that affect delivery guarantees and opportunities for bugs in
> upper layers?").
> Users currently rely heavily on analogies to other database systems to make
> sense of Kafka's transactions, and we need to use that to our benefit,
> rather than designing in spite of it being true.
>
> However this atomicity guarantee isn't always desirable, as evidenced by
> the original bug report [2]. If you're interacting with a website form for
> example, and a database transaction fails because one of your strings is
> oversize, you don't need to re-input all of your form responses from
> scratch, as there is an application layer/browser in-between to preserve
> the state and retry the transaction.
> And while you could make a convenience/performance/etc argument in that
> situation ("The database should truncate/null-out the oversize string") and
> modern databases often have very expressive DML that would permit such a
> behavior (try-catch, etc), the End-to-End arguments [3] make me believe
> that is a bad design and should be discouraged.
> To that end, I was suggesting ways to push this farther and farther up the
> stack, such as performing record size estimation. This doesn't mean that it
> can't be added at a low level of abstraction, just that we need to make
> sure to exhaust all other alternatives, and justify it with a performance
> benefit.
>
> I was holding off on discussing the literal design until you provided
> concrete performance justification, but to progress the discussion while
> i'm waiting for that, I can give my thoughts:
>
> I don't think an overloaded send() method is appropriate given that this
> appears to be a niche use-case, and the send() method itself is probably
> the single most important method in the Clients library. The KIP-98 design
> was a much more substantial change to the Producer than this KIP, and it
> found a way to preserve the original type signature (but added an
> exception).
> Users picking up the Producer for the first time may see this additional
> method, and may spend time trying to understand whether it is something
> suitable for their use-case. In the best case, they ignore it and use the
> other two signatures. But it is also possible that they will use it without
> understanding it, and have unexpected data loss. Overall, this feels like a
> net negative to the producer user-base.
> Also boolean, enum, vararg, flags, etc don't follow the current trend of
> the AdminClient passing Options DTOs for modifying individual API calls,
> which is a much more extensible design.
>
> If there was a way of preparing a record before calling send() that did
> some or all of the pre-send validation (serialization, size checking, maybe
> topic-partition existence existence, authorization, etc) that could be a
> reasonable way to emit these sorts of errors in a way that makes it clear
> that sending hasn't started and the record is not part of the transaction
> yet. I'm imagining something like:
>
> ```
> public abstract class PreparedRecord<K, V> extends ProducerRecord<K, V> { }
> // Actual instantiated class and methods could be an implementation detail.
> interface Producer {
> Optional<PreparedRecord<K, V>> prepare(ProducerRecord<K, V> record,
> PrepareOptions options) throws SerializationException,
> RecordTooLargeException;
> }
>
> // Application code:
> Optional<PreparedRecord<byte[], byte[]>> prepared = Optional.empty();
> try {
> prepared = producer.prepare(record, null);
> } catch (Exception e) {
> if (critical(e)) {
> producer.abortTransaction();
> return;
> }
> }
> if (prepared.isPresent()) {
> producer.send(prepared.get()); // errors will be propagated via
> commitTransaction() and also abort the transaction.
> }
> ```
>
> Semantically then, it would make sense that the data which has been
> "prepared to send" but has not been "sent" is intentionally left out of the
> batch by the application control-flow, not an option passed into the
> producer changing the control-flow within the Producer. If passed a
> prepared record, the send method can then be responsible only for waiting
> for sufficient buffer capacity, reusing the work done by the prepare()
> method.
> This prepare method could also enable other use-cases, like parallel
> serialization, exception-less handling, or more advanced buffering/balking
> behavior ("refuse to prepare records that can't be sent without blocking").
> And because the synchronous processing is made explicit, it's much easier
> to communicate to users which exceptions are covered and can be prevented
> from causing transaction aborts.
> If someone uses the method, or doesn't use it, they aren't put at risk of
> compromising their delivery guarantees unexpectedly.
>
> Thanks,
> Greg
>
> [1]
>
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-98+-+Exactly+Once+Delivery+and+Transactional+Messaging
> [2] https://issues.apache.org/jira/browse/KAFKA-15259
> [3] https://web.mit.edu/Saltzer/www/publications/endtoend/endtoend.pdf
>
> On Fri, Jul 19, 2024 at 12:39 PM Matthias J. Sax <mj...@apache.org> wrote:
>
> > For catching client side errors this would work IMHO. I am ok with this.
> >
> > We throw before we add the record to the batch. Very clean semantics
> > which should also address the concern of "non-atomic tx"... The
> > exception clearly indicates that the record was not added to the TX, and
> > users can react to it accordingly.
> >
> > We did discuss this idea previously, but did not have a good proposal to
> > make it backward compatible. The newly proposed overload would address
> > this issue of backward compatibility.
> >
> > Of course, it might not make it easily extensible in the future for
> > broker side errors, but it's unclear anyway right now, if we would even
> > get to a solution for broker side errors or not -- so maybe it's ok to
> > accept this and drop/ignore the broker side error question for now.
> >
> >
> >
> > A small follow up thought/question: instead of using a boolean, would we
> > actually want to make it a var-arg enum to allow users to enable this
> > for certain errors explicitly and individually? Beside the added
> > flexibility and fine grain control, a var-arg enum would also make the
> > API nicer/cleaner IMHO compare to a boolean.
> >
> > For convenience, this enum could have an additional `ALL` option (and we
> > would call out that if `ALL` is used, new error types might be added in
> > future release making the code less safe/robust -- ie, use at your own
> > risk only)
> >
> > This way, we also explicitly document what exception might be thrown in
> > the KIP, as we would add an enum for each error type explicitly, and
> > also make if future proof for new error types we want to cover -- each
> > addition would require a KIP to extend the enum.
> >
> >
> >
> > -Matthias
> >
> >
> > On 7/18/24 10:33 PM, Artem Livshits wrote:
> > > Hey folks,
> > >
> > > Hopefully not to make this KIP go for another spin :-), but I thought
> of
> > a
> > > modification that might actually address safety concerns over using
> flags
> > > to ignore a vaguely specified class of errors.
> > >
> > > What if we had the following overload of .send method:
> > >
> > > void send(ProducerRecord record, Callback callback, boolean
> > > throwImmediately)
> > >
> > > and if throwImmediately=false, then we behave the same way as now
> (return
> > > errors via Future and poison transaction) and if throwImmediately=true
> > then
> > > we just throw errors immediately from the send function. This way we
> > don't
> > > ignore the error, we're just changing the method they are delivered.
> > Then
> > > KStreams can catch the error for send(record, callback, true) and do
> > > whatever it needs to do.
> > >
> > > -Artem
> > >
> > >
> > > On Mon, Jul 15, 2024 at 4:30 PM Greg Harris
> <greg.har...@aiven.io.invalid
> > >
> > > wrote:
> > >
> > >> Matthias,
> > >>
> > >> Thank you for rejecting my suggested alternatives. Your responses are
> > the
> > >> sorts of things I expected to see summarized in the text of the KIP.
> > >>
> > >> I agree with most of your rejections, except this one:
> > >>
> > >>> "Estimation" is not sufficient, but we would need to know it exactly.
> > >>> And that's an impl detail, given that the message format could change
> > >>> and we could add new internal fields increasing the message size.
> > >>
> > >> An estimate is certainly going to have an error. But an estimate
> > shouldn't
> > >> be treated as exact anyway, there should be an error bound, or "safety
> > >> factor" used when interpreting it. For example, if the broker side
> > limit is
> > >> 1MB, and an estimate could be wrong by ~10%, then computing an
> estimate
> > and
> > >> dropping records >900kb should be sufficient to prevent RTLEs.
> > >> This is the sort of estimation that I would expect application
> > developers
> > >> could implement, without knowing the exact serialization and protocol
> > >> overhead. This could prevent user-originated oversize records from
> > making
> > >> it to the producer.
> > >>
> > >> Thanks,
> > >> Greg
> > >>
> > >>
> > >> On Mon, Jul 15, 2024 at 4:08 PM Matthias J. Sax <mj...@apache.org>
> > wrote:
> > >>
> > >>> I agree with Alieh and Artem -- in the end, why buffer records twice?
> > We
> > >>> effectively want to allow to push some error handling (which I btw
> > >>> consider "business logic") into the producer. IMHO, there is nothing
> > >>> wrong with it. Dropping a poison pill record is no really a violation
> > of
> > >>> atomicity from my POV, but a business logic decision to not include a
> > >>> record in a transaction -- the proposed API just makes it much
> simpler
> > >>> to achieve this business logic goal.
> > >>>
> > >>>
> > >>>
> > >>> For memory size estimation, throughput or message size is actually
> not
> > >>> relevant, right? We would need to look at producer buffer size, ie,
> > >>> `batch.size`, `max.in.flight.request.per.connection` and guesstimate
> > the
> > >>> number of connections there might be? At least for KS, we don't need
> to
> > >>> buffer everything until commit, but only until we get a successful
> > "ack"
> > >>> back.
> > >>>
> > >>> Note that KS application not only need to write to (a single) user
> > >>> result topic, but multiple output topics, as well as repartition and
> > >>> changelog topics, across all tasks assigned to a thread (ie,
> producer),
> > >>> which can easily be 10 tasks or more. If we assume topics with 30
> > >>> partitions (topics with 50 or more partitions are not uncommon
> either),
> > >>> and a producer who must write to 10 different topics, the number of
> > >>> required connections is very quickly very high, and thus the required
> > >>> "application buffer space" would be significant.
> > >>>
> > >>>
> > >>>
> > >>> Your others ideas seems not to be viable alternatives:
> > >>>
> > >>>> Streams users that specifically want to drop oversize records can
> > >>>> estimate the size of their data and drop records which are too
> > >>>> large, enforcing their own limits that are lower than the Kafka
> > limits.
> > >>>
> > >>> "Estimation" is not sufficient, but we would need to know it exactly.
> > >>> And that's an impl detail, given that the message format could change
> > >>> and we could add new internal fields increasing the message size. The
> > >>> idea to add some `producer.serializedRecordSize()` helper method was
> > >>> discussed, but it's a very ugly API and clumsy to use -- also, the
> user
> > >>> code would need to know the producer config which it might not have
> > >>> access to (as it might get passed in from some config file; and it
> > might
> > >>> also be changed).
> > >>>
> > >>> Some other alternative we also discussed was, to let `send()` throw
> an
> > >>> exception for a "record too large" case directly. However, this
> > solution
> > >>> raises backward compatibly concerns, and it might also not help us to
> > >>> extend the solution in the future (eg, tackle broker side errors). So
> > we
> > >>> discarded this idea.
> > >>>
> > >>>
> > >>>
> > >>>> Streams users that want CONTINUE semantics can use at_least_once
> > >>>> semantics
> > >>>
> > >>> Not really. EOS is mainly about not having duplicates in the result,
> > but
> > >>> at-least-once cannot provide this guarantee. (Even if I repeat my
> self:
> > >>> but dropping a poison pill record based on a business logic decision
> is
> > >>> not data loss, but effectively a business logic filter...)
> > >>>
> > >>>
> > >>>
> > >>>> Streams itself can store record hashes/coordinates and fast rewind
> to
> > >>>> the end of the last transaction, recomputing data rather than
> storing
> > >> it.
> > >>>
> > >>> Given the very complex nature of topologies, with joins,
> aggregations,
> > >>> flatmaps etc, this is a 100x more complex solution and not viable in
> > >>> practice.
> > >>>
> > >>>
> > >>>
> > >>>> Streams can define exactly_once + CONTINUE semantics to permit the
> > >> whole
> > >>>> transaction to be dropped, because it would allow the next batch to
> be
> > >>>> started processing.
> > >>>
> > >>> Would this not be much worse? I have a single poison pill record and
> > >>> would need to drop a full tx (this could be tens of thousands of
> > >>> records...). Also, given that KS write into changelog topic in the
> same
> > >>> TX, this could break the whole application.
> > >>>
> > >>>
> > >>>
> > >>>> Streams can emit records with both a transactional and
> > >> non-transactional
> > >>>> producer if some records are not critical-path
> > >>>
> > >>> We (1) already have a "too many connections" problem with KS so using
> > >>> move clients is something we try to avoid (and we actually hope to
> > >>> reduce the number of client and connection mid to long term), (2)
> this
> > >>> would be very hard to express at the API level to the user, and (3)
> it
> > >>> would provide very weird semantics.
> > >>>
> > >>>
> > >>>
> > >>>> they should optimize for smaller transactions,
> > >>>
> > >>> IMHO, this would not work in practice because transaction have a high
> > >>> overhead and commit-interval is used to tradeoff throughput vs
> > >>> end-to-end latency. Given certain throughput requirement, it would
> not
> > >>> be possible to just use a lower commit interval to reduce memory
> > >>> requirements.
> > >>>
> > >>>
> > >>>
> > >>> -Matthias
> > >>>
> > >>>
> > >>>
> > >>>
> > >>> On 7/15/24 2:25 PM, Artem Livshits wrote:
> > >>>> Hi Greg,
> > >>>>
> > >>>>> This makes me think that this IGNORE_SEND_ERRORS covers an
> arbitrary
> > >> set
> > >>>> of error conditions that may be expanded in the future, possibly to
> > >> cover
> > >>>> the broker side RecordTooLargeException.
> > >>>>
> > >>>> I don't think it contradicts what I said (the keyword here is "in
> the
> > >>>> future") -- with the current functionality, the correct way to
> handle
> > >>> RTLE
> > >>>> is by only letting the client ignore client-originated RTLE (this
> can
> > >> be
> > >>>> easily implemented on the client side). In the future, we can
> improve
> > >> on
> > >>>> that by making the broker return a different exception for
> > >>> batch-too-large
> > >>>> case, then the producer would be able to return broker side
> exceptions
> > >> as
> > >>>> well (and if the application chooses to ignore it -- it will be able
> > >> to,
> > >>>> but it would be an explicit choice rather than ignoring it by
> > mistake),
> > >>> in
> > >>>> this case the producer client would encapsulate backward
> compatibility
> > >>>> logic when it connects to older brokers to make sure the the
> > >> application
> > >>>> doesn't accidentally gets RTLE originated by the old broker. This
> > >>>> functionality is obviously more involved and we'll need to see if
> > going
> > >>> all
> > >>>> the way is justified, but the partial client-only solution doesn't
> > >> close
> > >>>> the door.
> > >>>>
> > >>>> So one way to look at the current situation is the following:
> > >>>>
> > >>>> 1. We can do a low effort partial solution to solve a real existing
> > >>>> problem. We can easily prove that it would do exactly what it needs
> > to
> > >>> do
> > >>>> with minimal risk of regression.
> > >>>> 2. We have a path to a more comprehensive solution, so if we justify
> > >> the
> > >>>> effort required for that, we can get there.
> > >>>>
> > >>>> BTW, as a side note (I think a saw a question in the thread), we do
> > try
> > >>> to
> > >>>> introduce error categories here
> > >>>>
> > >>>
> > >>
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1050%3A+Consistent+error+handling+for+Transactions
> > >>>> so eventually we may have a better classification for the errors.
> > >>>>
> > >>>>> "if a streams producer is producing 1MB/s, and the commit interval
> is
> > >> 1
> > >>>> hour, I expect 3600MB of additional heap needed ...
> > >>>>
> > >>>> Agree, that would be ideal. On the other hand, the effort to prove
> > >> that
> > >>>> keeping all records in memory won't break some scenarios (and
> > generally
> > >>>> breaking one is enough to cause a lot of pain) seems to be
> > >> significantly
> > >>>> higher than to prove that setting some flag in some API has pretty
> > >> much 0
> > >>>> chance of regression (we basically have a flag to say "unfix
> > >> KAFKA-9279"
> > >>> so
> > >>>> we're getting to fairly "known good" state). I'll let KStream folks
> > >>>> comment on this one (and we still need to solve the problem of
> > >> accidental
> > >>>> handling of RTLE originated from broker, so some KIP would be
> required
> > >> to
> > >>>> somehow help to differentiate those).
> > >>>>
> > >>>> -Artem
> > >>>>
> > >>>> On Mon, Jul 15, 2024 at 1:31 PM Greg Harris
> > >> <greg.har...@aiven.io.invalid
> > >>>>
> > >>>> wrote:
> > >>>>
> > >>>>> Hi Artem,
> > >>>>>
> > >>>>> Thank you for clarifying as I'm joining the conversation late and
> may
> > >>> have
> > >>>>> some misconceptions.
> > >>>>>
> > >>>>>> Because of this, a more "complete" solution that
> > >>>>>> allows ignoring RecordTooLargeException regardless of its origin
> is
> > >>>>>> actually incorrect, while a "partial" solution that allows
> ignoring
> > >>>>>> RecordTooLargeException only originating in client code
> accomplishes
> > >>> the
> > >>>>>> required functionality.
> > >>>>>
> > >>>>> This is not how I understood this feature. Above Matthias said the
> > >>>>> following:
> > >>>>>
> > >>>>>> We can do
> > >>>>>> follow up KIP for other errors on an on-demand basis and
> fix-forward
> > >> /
> > >>>>>> enlarge the scope successively.
> > >>>>>
> > >>>>> This makes me think that this IGNORE_SEND_ERRORS covers an
> arbitrary
> > >>> set of
> > >>>>> error conditions that may be expanded in the future, possibly to
> > cover
> > >>> the
> > >>>>> broker side RecordTooLargeException.
> > >>>>>
> > >>>>>> Obviously, we could solve this problem by changing logic in the
> > >>>>>> broker to return a different error when the batch is too large,
> but
> > >>> right
> > >>>>>> now this is not the case
> > >>>>>
> > >>>>> If the broker/wire protocol isn't ready for these errors to be
> > >>> propagated,
> > >>>>> then I don't think we're ready to add this API. It's going to be
> > >>>>> under-generalized, and there's a decent chance that we're going to
> > >>> regret
> > >>>>> the design choices in the future. And users that expect it to be
> > fully
> > >>>>> generalized are going to be disappointed when they don't read the
> > fine
> > >>>>> print and still get faulted by non-covered errors.
> > >>>>>
> > >>>>>> AL2. In a high performance system, "just an optimization" can be
> a
> > >>>>>> functional requirement ...
> > >>>>>> I just wanted to make the point that we shouldn't necessarily
> > >> dismiss
> > >>>>>> API changes that allow for optimizations.
> > >>>>>
> > >>>>> My earlier statement didn't dismiss this feature as "just an
> > >>> optimization",
> > >>>>> actually the opposite. I said that performance could be a
> > >> justification,
> > >>>>> but only if it is quantified and stated explicitly. We shouldn't be
> > >>> voting
> > >>>>> on hand-wavy optimizations, we should be voting on things that are
> > >>>>> quantifiable.
> > >>>>> For example an analysis like the following would facilitate further
> > >>>>> discussion: "if a streams producer is producing 1MB/s, and the
> commit
> > >>>>> interval is 1 hour, I expect 3600MB of additional heap needed per
> > >>>>> producer". We can then discuss whether we expect higher or lower
> > >>>>> throughput, commit intervals, or heap usage to determine what the
> > >>> operating
> > >>>>> envelope of this feature could be.
> > >>>>> If there are a substantial number of users that have high
> throughput,
> > >>> long
> > >>>>> commit intervals, _and_ RTLEs, then this feature could make sense.
> If
> > >>> not,
> > >>>>> then the downsides of this feature (complication of the API,
> > >>>>> under-specification of the error coverage, etc) look unjustified.
> In
> > >>> fact,
> > >>>>> if the number of users regularly encountering RTLEs is sufficiently
> > >>> small,
> > >>>>> I would strongly advocate for an application-specific workaround
> > >>> instead of
> > >>>>> trying to fix this in Streams, or make memory buffering an optional
> > >>> feature
> > >>>>> in streams.
> > >>>>>
> > >>>>> Thanks,
> > >>>>> Greg
> > >>>>>
> > >>>>> On Mon, Jul 15, 2024 at 1:29 PM Greg Harris <greg.har...@aiven.io>
> > >>> wrote:
> > >>>>>
> > >>>>>> Hi Alieh,
> > >>>>>>
> > >>>>>> Thanks for your response.
> > >>>>>>
> > >>>>>>> what does a user do
> > >>>>>>> after a transaction is failed due to a `too-large-record
> > `exception?
> > >>>>> They
> > >>>>>>> will submit the same batch without the problematic record again.
> > >>>>>>
> > >>>>>> If they re-submit the same record, they are indicating that this
> > >> record
> > >>>>> is
> > >>>>>> an integral part of the transaction, and the transaction should
> only
> > >> be
> > >>>>>> committed with it present. If the record isn't integral to the
> > >>>>> transaction,
> > >>>>>> they shouldn't submit it as part of the transaction.
> > >>>>>>
> > >>>>>>> Regarding your solution to solve the issue application-side: I
> am
> > a
> > >>>>>>> bit hesitant to keep all sent records in memory since I think
> > >>> buffering
> > >>>>>>> records twice (both in Streams and Producer) would not be an
> > >> efficient
> > >>>>>>> solution.
> > >>>>>>
> > >>>>>> I understand your hesitation, and this touches on the
> "performance"
> > >>>>> caveat
> > >>>>>> of the end-to-end arguments in system design. There are no perfect
> > >>>>> designs,
> > >>>>>> and some API cleanliness may be sacrificed in favor of more
> > >> performant
> > >>>>>> solutions. You would need to make a concrete and convincing
> argument
> > >>> that
> > >>>>>> the performance of this solution would be better than every
> > >>> alternative.
> > >>>>> To
> > >>>>>> that end, I would recommend that you add more to the "Rejected
> > >>>>>> Alternatives" section, as that is going to carry this proposal.
> > >>>>>> Some alternatives that I can think of, but which aren't
> necessarily
> > >>>>> better:
> > >>>>>> 1. Streams users that specifically want to drop oversize records
> can
> > >>>>>> estimate the size of their data and drop records which are too
> > >>>>>> large, enforcing their own limits that are lower than the Kafka
> > >> limits.
> > >>>>>> 2. Streams users that want CONTINUE semantics can use
> at_least_once
> > >>>>>> semantics
> > >>>>>> 3. Streams itself can store record hashes/coordinates and fast
> > rewind
> > >>> to
> > >>>>>> the end of the last transaction, recomputing data rather than
> > storing
> > >>> it.
> > >>>>>> 4. Streams can define exactly_once + CONTINUE semantics to permit
> > the
> > >>>>>> whole transaction to be dropped, because it would allow the next
> > >> batch
> > >>> to
> > >>>>>> be started processing.
> > >>>>>> 5. Streams can emit records with both a transactional and
> > >>>>>> non-transactional producer if some records are not critical-path
> > >>>>>>
> > >>>>>> To generalize this point: Suppose an application tries to minimize
> > >>>>> storage
> > >>>>>> costs by having only one party responsible for a piece of data at
> a
> > >>> time.
> > >>>>>> They initially have the data, call send(), and want to know the
> > >>> earliest
> > >>>>>> time they can forget the data and transfer the responsibility to
> > >> Kafka.
> > >>>>>> With a non-transactional producer, they are responsible for the
> data
> > >>>>> until
> > >>>>>> the send() callback has succeeded. With a transactional producer,
> > >> they
> > >>>>> are
> > >>>>>> responsible for the data until commitTransaction() has succeeded.
> > >>>>>> With this proposed change that makes the producer tolerate
> > >>>>>> too-large-exceptions, applications are still responsible for
> storing
> > >>>>> their
> > >>>>>> data until commitTransaction() has succeeded, because
> > >>> abortTransaction()
> > >>>>>> could have also been called, or the producer could have been
> fenced,
> > >> or
> > >>>>> any
> > >>>>>> number of other failures could have occurred. This feature does
> not
> > >>>>> enable
> > >>>>>> Streams to drop responsibility earlier, it carves out a specific
> > >>>>> situation
> > >>>>>> in which it doesn't have to rewind processing, which is a
> > performance
> > >>>>>> concern.
> > >>>>>>
> > >>>>>> For Streams and the general case, if an application is trying to
> > >>> optimize
> > >>>>>> storage costs, they should optimize for smaller transactions,
> > because
> > >>>>> this
> > >>>>>> both lowers the bound on record re-delivery and lowers the
> > likelihood
> > >>> of
> > >>>>> a
> > >>>>>> bad record being included in any individual transaction.
> > >>>>>>
> > >>>>>> Thanks,
> > >>>>>> Greg
> > >>>>>>
> > >>>>>> On Mon, Jul 15, 2024 at 12:35 PM Artem Livshits
> > >>>>>> <alivsh...@confluent.io.invalid> wrote:
> > >>>>>>
> > >>>>>>> Hi Greg,
> > >>>>>>>
> > >>>>>>> What you say makes a lot of sense. I just wanted to clarify a
> > >> couple
> > >>> of
> > >>>>>>> subtle points.
> > >>>>>>>
> > >>>>>>> AL1. There is a functional reason to handle errors that happen on
> > >> send
> > >>>>>>> (oginate in the producer logic in the client) vs. errors that are
> > >>>>> returned
> > >>>>>>> from the broker. The problem is that RecordTooLargeException is
> > >>>>> returned
> > >>>>>>> in two cases: (1) the producer logic on the client checks that
> > >> record
> > >>> is
> > >>>>>>> too large and throws the exception before doing anything with
> this
> > >> --
> > >>>>> this
> > >>>>>>> is very "clean" situation with one specific record being marked
> as
> > >>>>> "poison
> > >>>>>>> pill" and rejected; (2) the broker throws the same error if the
> > >> batch
> > >>> is
> > >>>>>>> too large -- the batch may include multiple records and none of
> > them
> > >>>>> would
> > >>>>>>> necessarily be a "poison pill" record, it's just a random
> > >>>>> misconfiguration
> > >>>>>>> of client vs. broker. Because of this, a more "complete"
> solution
> > >>> that
> > >>>>>>> allows ignoring RecordTooLargeException regardless of its origin
> is
> > >>>>>>> actually incorrect, while a "partial" solution that allows
> ignoring
> > >>>>>>> RecordTooLargeException only originating in client code
> > accomplishes
> > >>> the
> > >>>>>>> required functionality. This is an important nuance and should
> be
> > >>> added
> > >>>>>>> to
> > >>>>>>> the KIP. Obviously, we could solve this problem by changing
> logic
> > >> in
> > >>>>> the
> > >>>>>>> broker to return a different error when the batch is too large,
> but
> > >>>>> right
> > >>>>>>> now this is not the case (and to have the correct error handling
> > >> we'd
> > >>>>> need
> > >>>>>>> to know the version of the broker so we can only drop the records
> > if
> > >>> the
> > >>>>>>> error is returned from a broker that knows to return a different
> > >>> error).
> > >>>>>>>
> > >>>>>>> AL2. In a high performance system, "just an optimization" can
> be a
> > >>>>>>> functional requirement -- if a solution impacts memory or
> > >>> computational
> > >>>>>>> complexity (in the sense of bigO notation) on the main code path
> I
> > >> can
> > >>>>>>> justify changing APIs to avoid such an impact. I'll let KStream
> > >> folks
> > >>>>>>> comment on whether an implementation that requires storing
> records
> > >> in
> > >>>>>>> memory actually violates the computational complexity on the main
> > >> code
> > >>>>>>> path, I just wanted to make the point that we shouldn't
> necessarily
> > >>>>>>> dismiss
> > >>>>>>> API changes that allow for optimizations.
> > >>>>>>>
> > >>>>>>> -Artem
> > >>>>>>>
> > >>>>>>> On Fri, Jul 12, 2024 at 1:07 PM Greg Harris
> > >>>>> <greg.har...@aiven.io.invalid
> > >>>>>>>>
> > >>>>>>> wrote:
> > >>>>>>>
> > >>>>>>>> Hi all,
> > >>>>>>>>
> > >>>>>>>> Alieh, thanks for the KIP! And everyone else, thanks for the
> > robust
> > >>>>>>>> discussion.
> > >>>>>>>>
> > >>>>>>>> I understand that there are situations in which users desire
> that
> > >> the
> > >>>>>>>> pipeline "just keep working" and skip errors. However, I
> question
> > >>>>>>> whether
> > >>>>>>>> it is appropriate to support/encourage this behavior via
> inclusion
> > >> in
> > >>>>>>> the
> > >>>>>>>> Producer API.
> > >>>>>>>> This feature is essentially a "non-atomic transaction", as it
> > >> allows
> > >>>>>>>> commits in which not all records passed to send() ultimately get
> > >>>>>>> committed.
> > >>>>>>>> As atomicity is one of the most important semantics associated
> > with
> > >>>>>>>> transactions, I question whether there are users other than
> > Streams
> > >>>>> that
> > >>>>>>>> would choose non-atomic transactions over a
> traditional/idempotent
> > >>>>>>>> producer.
> > >>>>>>>> Some cursory research shows that non-atomic transactions may be
> > >>>>> present
> > >>>>>>> in
> > >>>>>>>> other databases, but is actively discouraged due to the
> complexity
> > >>>>> they
> > >>>>>>> add
> > >>>>>>>> to error-handling. [1]
> > >>>>>>>>
> > >>>>>>>> I'd like to invoke the End-to-End Arguments in System Design [2]
> > >>> here,
> > >>>>>>> and
> > >>>>>>>> recommend that this behavior may be present in Streams, but
> should
> > >>> not
> > >>>>>>> be
> > >>>>>>>> in the Producer.
> > >>>>>>>> 1. Dropping records that cause errors is already expressible via
> > >> the
> > >>>>>>>> current Producer API. You can store the records in-memory after
> > >>>>> calling
> > >>>>>>>> send(), wait for a successful no-error flush() before calling
> > >>>>>>>> commitTransaction() and allowing the record to be garbage
> > >> collected.
> > >>>>> If
> > >>>>>>>> errors occur, abortTransaction() and re-submit the records.
> > >>>>>>>> 2. Implementing this inside the Producer API is complex and
> > >> difficult
> > >>>>> to
> > >>>>>>>> holistically define in a way that we won't regret or need to
> > change
> > >>>>>>> later.
> > >>>>>>>> I think some of the disagreement in this thread originates from
> > >> this,
> > >>>>>>> and I
> > >>>>>>>> don't find the proposed API satisfactory.
> > >>>>>>>> 3. The performance improvement of including this change in the
> > >> lower
> > >>>>>>> level
> > >>>>>>>> needs to be quantified in order to be a justification, and I
> don't
> > >>> see
> > >>>>>>> any
> > >>>>>>>> analysis about this.
> > >>>>>>>>
> > >>>>>>>> I imagine that the alternative implementation I suggested in (1)
> > >>> would
> > >>>>>>> also
> > >>>>>>>> enable more expressive error handlers in Streams, if such a
> thing
> > >> was
> > >>>>>>>> desired. Keeping the record around until after the transaction
> is
> > >>>>>>> committed
> > >>>>>>>> would enable a DLQ or passing the erroneous record to the error
> > >>>>> handler.
> > >>>>>>>>
> > >>>>>>>> I think that the current pattern of the application being
> > >> responsible
> > >>>>>>> for
> > >>>>>>>> providing good data to the producer is very reasonable; Having
> the
> > >>>>>>> producer
> > >>>>>>>> responsible for implementing the application's error handling of
> > >> bad
> > >>>>>>> data
> > >>>>>>>> is not something I can support.
> > >>>>>>>>
> > >>>>>>>> Thanks,
> > >>>>>>>> Greg
> > >>>>>>>>
> > >>>>>>>> [1] https://www.sommarskog.se/error_handling/Part1.html
> > >>>>>>>> [2]
> > >>>>> https://web.mit.edu/Saltzer/www/publications/endtoend/endtoend.pdf
> > >>>>>>>>
> > >>>>>>>> On Fri, Jul 12, 2024 at 8:52 AM Justine Olshan
> > >>>>>>>> <jols...@confluent.io.invalid>
> > >>>>>>>> wrote:
> > >>>>>>>>
> > >>>>>>>>> Can we update the KIP to clearly document these decisions?
> > >>>>>>>>>
> > >>>>>>>>> Thanks,
> > >>>>>>>>>
> > >>>>>>>>> Justine
> > >>>>>>>>>
> > >>>>>>>>> On Tue, Jul 9, 2024 at 9:25 AM Andrew Schofield <
> > >>>>>>>> andrew_schofi...@live.com
> > >>>>>>>>>>
> > >>>>>>>>> wrote:
> > >>>>>>>>>
> > >>>>>>>>>> Hi Chris,
> > >>>>>>>>>> As it stands, the error handling for transactions in
> > >> KafkaProducer
> > >>>>>>> is
> > >>>>>>>> not
> > >>>>>>>>>> ideal. There’s no reason why a failed operation should fail a
> > >>>>>>>> transaction
> > >>>>>>>>>> provided that the application can tell that the operation was
> > not
> > >>>>>>>>> included
> > >>>>>>>>>> in the transaction and then make its own decision whether to
> > >>>>>>> continue
> > >>>>>>>> or
> > >>>>>>>>>> back out. So, I think I disagree with the original premise of
> a
> > >>>>>>>>> client-side
> > >>>>>>>>>> error state for a transaction, but we are where we are.
> > >>>>>>>>>>
> > >>>>>>>>>> When I voted, I did not expect the KIP to handle ALL errors
> > which
> > >>>>>>> could
> > >>>>>>>>>> conceivably be handled. I did expect it to handle client-side
> > >> send
> > >>>>>>>> errors
> > >>>>>>>>>> that would cause a record to be rejected from a batch before
> > >>>>> sending
> > >>>>>>>> to a
> > >>>>>>>>>> broker. I think that it does make the KafkaProducer interface
> > >> very
> > >>>>>>>>> slightly
> > >>>>>>>>>> more complicated, but the new option is a clear improvement
> and
> > I
> > >>>>>>>>>> don’t see anyone getting into a mess using it.
> > >>>>>>>>>>
> > >>>>>>>>>> I think broker-side errors are more tricky and I don’t think
> an
> > >>>>>>>> overload
> > >>>>>>>>>> on the send() method is going to do the job. I don’t see that
> as
> > >> a
> > >>>>>>>>> problem
> > >>>>>>>>>> with the KIP, just that the underlying RPCs and behaviour is
> not
> > >>>>>>> very
> > >>>>>>>>>> amenable to record-specific error handling. The Produce RPC
> is a
> > >>>>>>>>>> complicated beast which can include a set of records for
> mutiple
> > >>>>>>>>>> topic-partitions. Although ProduceResponse v10 does include
> > >> record
> > >>>>>>>>>> errors, I don’t believe this is surfaced in the client. Let’s
> > >>>>>>> imagine
> > >>>>>>>>>> something
> > >>>>>>>>>> like broker-side record validation which barfs on one record.
> > >>>>>>> Failing
> > >>>>>>>> an
> > >>>>>>>>>> entire batch is easier, but less useful if the problem is
> > related
> > >>>>> to
> > >>>>>>>> one
> > >>>>>>>>>> record.
> > >>>>>>>>>>
> > >>>>>>>>>> In summary, I’m happy that my vote stands, and I am happy with
> > >> the
> > >>>>>>> KIP
> > >>>>>>>>>> only supporting client-side record errors.
> > >>>>>>>>>>
> > >>>>>>>>>> Thanks,
> > >>>>>>>>>> Andrew
> > >>>>>>>>>>
> > >>>>>>>>>>> On 8 Jul 2024, at 16:37, Chris Egerton
> <chr...@aiven.io.INVALID
> > >>>>>>
> > >>>>>>>>> wrote:
> > >>>>>>>>>>>
> > >>>>>>>>>>> Hi Alieh,
> > >>>>>>>>>>>
> > >>>>>>>>>>> Can you clarify why broker-side errors shouldn't be covered?
> > The
> > >>>>>>> only
> > >>>>>>>>>> real
> > >>>>>>>>>>> rationale I can come up with is that it's easier to
> implement.
> > >>>>>>>>>>>
> > >>>>>>>>>>> "Things were better for Kafka Streams before KAFKA-9279 was
> > >>>>> fixed"
> > >>>>>>>>> isn't
> > >>>>>>>>>>> very convincing, because Kafka Streams is not the only user
> of
> > >>>>> the
> > >>>>>>>> Java
> > >>>>>>>>>>> producer client. And for others, especially new users, I
> doubt
> > >>>>>>> that
> > >>>>>>>>> this
> > >>>>>>>>>>> new API we're proposing would make sense without having to
> > >>>>>>> consult a
> > >>>>>>>>> lot
> > >>>>>>>>>> of
> > >>>>>>>>>>> historical context.
> > >>>>>>>>>>>
> > >>>>>>>>>>> I also don't think that most users will know or even care
> about
> > >>>>>>> the
> > >>>>>>>>>>> distinction between errors that cause a record to fail before
> > >>>>> it's
> > >>>>>>>>> added
> > >>>>>>>>>> to
> > >>>>>>>>>>> a batch vs. after. If you were writing a producer application
> > of
> > >>>>>>> your
> > >>>>>>>>>> own,
> > >>>>>>>>>>> and you wanted to handle RecordTooLargeException instances by
> > >>>>>>>> dropping
> > >>>>>>>>> a
> > >>>>>>>>>>> record without aborting a transaction, would you care about
> > >>>>>>> whether
> > >>>>>>>> it
> > >>>>>>>>>> was
> > >>>>>>>>>>> your client or your broker that balked? Would you be happy if
> > >>>>> you
> > >>>>>>>> wrote
> > >>>>>>>>>>> logic expecting that that problem was solved once and for
> all,
> > >>>>>>> only
> > >>>>>>>> to
> > >>>>>>>>>>> learn that it could still affect you in other circumstances?
> > Or,
> > >>>>>>>>>>> alternatively, would you be happy if you wanted to solve that
> > >>>>>>> problem
> > >>>>>>>>> and
> > >>>>>>>>>>> found an API that seemed to do exactly what you wanted, but
> > >>>>> after
> > >>>>>>>>> reading
> > >>>>>>>>>>> the fine print, realized you'd have to do it yourself
> instead?
> > >>>>>>>>>>>
> > >>>>>>>>>>> Ultimately, the more I think about this, the more I believe
> > that
> > >>>>>>>> we're
> > >>>>>>>>>>> adding noise to the API (with the new overloaded variant of
> > >>>>> send)
> > >>>>>>>> for a
> > >>>>>>>>>>> feature that will likely bring confusion and even frustration
> > to
> > >>>>>>>> anyone
> > >>>>>>>>>>> besides maintainers of Kafka Streams who tries to use it.
> > >>>>>>>>>>>
> > >>>>>>>>>>> If the only concern about covering broker-side errors is that
> > it
> > >>>>>>>> would
> > >>>>>>>>> be
> > >>>>>>>>>>> more difficult to implement, I believe we should strongly
> > >>>>>>> reconsider
> > >>>>>>>>> that
> > >>>>>>>>>>> alternative. That said, if there is a straightforward way to
> > >>>>>>> explain
> > >>>>>>>>> this
> > >>>>>>>>>>> feature to new users that won't mislead them or require them
> to
> > >>>>> do
> > >>>>>>>>>> research
> > >>>>>>>>>>> on producer internals, then I can still live with it.
> > >>>>>>>>>>>
> > >>>>>>>>>>> Regarding a list of recoverable vs. irrecoverable errors,
> this
> > >>>>> is
> > >>>>>>>>>> actually
> > >>>>>>>>>>> the subject of another recently-introduced KIP:
> > >>>>>>>>>>>
> > >>>>>>>>>>
> > >>>>>>>>>
> > >>>>>>>>
> > >>>>>>>
> > >>>>>
> > >>>
> > >>
> >
> https://cwiki.apache.org/confluence/display/KAFKA/KIP-1050%3A+Consistent+error+handling+for+Transactions
> > >>>>>>>>>>>
> > >>>>>>>>>>> Finally, I'd also like to ask the people who have already
> voted
> > >>>>>>>>> (Andrew,
> > >>>>>>>>>>> Matthias) if, at the time they voted, they believed that the
> > API
> > >>>>>>>> would
> > >>>>>>>>>>> handle all errors, or only the subset of errors that would
> > >>>>> cause a
> > >>>>>>>>> record
> > >>>>>>>>>>> to be rejected from a batch before it can be sent to a
> broker.
> > >>>>>>>>>>>
> > >>>>>>>>>>> Best,
> > >>>>>>>>>>>
> > >>>>>>>>>>> Chris
> > >>>>>>>>>>>
> > >>>>>>>>>>> On Thu, Jul 4, 2024 at 12:43 PM Alieh Saeedi
> > >>>>>>>>>> <asae...@confluent.io.invalid>
> > >>>>>>>>>>> wrote:
> > >>>>>>>>>>>
> > >>>>>>>>>>>> Salut from the KIP’s author
> > >>>>>>>>>>>>
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> Clarifying two points:
> > >>>>>>>>>>>>
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> 1) broker side errors:
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> As far as I remember we are not going to cover the errors
> > >>>>>>>> originating
> > >>>>>>>>>> from
> > >>>>>>>>>>>> the broker!
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> A historical fact: One of the debate points in KIP-1038 was
> > >>>>> that
> > >>>>>>> by
> > >>>>>>>>>>>> defining a producer custom handler, the user may assume that
> > >>>>>>>>> broker-side
> > >>>>>>>>>>>> errors must be covered as well. They may define a handler
> for
> > >>>>>>>> handling
> > >>>>>>>>>>>> `RecordTooLargeException` and still see such errors not
> being
> > >>>>>>>> handled
> > >>>>>>>>> as
> > >>>>>>>>>>>> they wish.
> > >>>>>>>>>>>>
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> 2) Regarding irrecoverable/recoverable errors:
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> Before the fix of `KAFKA-9279`, errors such as
> > >>>>>>>>>> `RecordTooLargeException`
> > >>>>>>>>>>>> or errors related to missing meta data (both originating
> from
> > >>>>>>>> Producer
> > >>>>>>>>>>>> `send()`) were considered as recoverable but after that they
> > >>>>>>> turned
> > >>>>>>>>> into
> > >>>>>>>>>>>> being irrecoverable without changing any Javadocs or having
> > any
> > >>>>>>> KIP.
> > >>>>>>>>>> All
> > >>>>>>>>>>>> the effort made in this KIP and the former one have been
> > >>>>> towards
> > >>>>>>>>>> returning
> > >>>>>>>>>>>> to the former state.
> > >>>>>>>>>>>>
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> I am sure that it is clear for you that which sort of errors
> > we
> > >>>>>>> are
> > >>>>>>>>>> going
> > >>>>>>>>>>>> to cover: A single record may happen to NOT get added to the
> > >>>>>>> batch
> > >>>>>>>> due
> > >>>>>>>>>> to
> > >>>>>>>>>>>> the issues with the record or its corresponding topic. The
> > >>>>> point
> > >>>>>>> was
> > >>>>>>>>>> that
> > >>>>>>>>>>>> if the record is not added to the batch let ’s don’t fail
> the
> > >>>>>>> whole
> > >>>>>>>>>> batch
> > >>>>>>>>>>>> because of that non-existing record. We never intended to do
> > >>>>> sth
> > >>>>>>> in
> > >>>>>>>>>> broker
> > >>>>>>>>>>>> side or ignore more important errors. But I agree with you
> > >>>>>>> Chris.
> > >>>>>>>> If
> > >>>>>>>>> we
> > >>>>>>>>>>>> are adding a new API, we must have good documentation for
> > that.
> > >>>>>>> The
> > >>>>>>>>>>>> sentence `all irrecoverable transactional errors will still
> be
> > >>>>>>>> fatal`
> > >>>>>>>>> as
> > >>>>>>>>>>>> you suggested is good. What do you think? I am totally
> against
> > >>>>>>>>>> enumerating
> > >>>>>>>>>>>> errors in Javadocs since these sort of errors can be
> changing
> > >>>>>>> during
> > >>>>>>>>>>>> time. More
> > >>>>>>>>>>>> over, have you ever seen any list of recoverable or
> > >>>>> irrecoverable
> > >>>>>>>>> errors
> > >>>>>>>>>>>> somewhere so far?
> > >>>>>>>>>>>>
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> Bests,
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> Alieh
> > >>>>>>>>>>>>
> > >>>>>>>>>>>> On Wed, Jul 3, 2024 at 6:07 PM Chris Egerton
> > >>>>>>>> <chr...@aiven.io.invalid
> > >>>>>>>>>>
> > >>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>
> > >>>>>>>>>>>>> Hi Justine,
> > >>>>>>>>>>>>>
> > >>>>>>>>>>>>> I agree that enumerating a list of errors that should be
> > >>>>>>> covered by
> > >>>>>>>>> the
> > >>>>>>>>>>>> KIP
> > >>>>>>>>>>>>> is difficult; I was thinking it might be easier if we list
> > the
> > >>>>>>>> errors
> > >>>>>>>>>>>> that
> > >>>>>>>>>>>>> should _not_ be covered by the KIP, and only if we can't
> > >>>>> define
> > >>>>>>> a
> > >>>>>>>>>>>>> reasonable heuristic that would cover them without having
> to
> > >>>>>>>>> explicitly
> > >>>>>>>>>>>>> list them. Could it be enough to say "all irrecoverable
> > >>>>>>>> transactional
> > >>>>>>>>>>>>> errors will still be fatal", or even just "all
> transactional
> > >>>>>>> errors
> > >>>>>>>>> (as
> > >>>>>>>>>>>>> opposed to errors related to this specific record) will
> still
> > >>>>> be
> > >>>>>>>>>> fatal"?
> > >>>>>>>>>>>>>
> > >>>>>>>>>>>>> Cheers,
> > >>>>>>>>>>>>>
> > >>>>>>>>>>>>> Chris
> > >>>>>>>>>>>>>
> > >>>>>>>>>>>>> On Wed, Jul 3, 2024 at 11:56 AM Justine Olshan
> > >>>>>>>>>>>>> <jols...@confluent.io.invalid>
> > >>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>
> > >>>>>>>>>>>>>> Hey Chris,
> > >>>>>>>>>>>>>>
> > >>>>>>>>>>>>>> I think what you say makes sense. I agree that defining
> the
> > >>>>>>>> behavior
> > >>>>>>>>>>>>> based
> > >>>>>>>>>>>>>> on code that can possibly change is not a good idea, and I
> > >>>>> was
> > >>>>>>>>> trying
> > >>>>>>>>>>>> to
> > >>>>>>>>>>>>>> get a clearer definition from the KIP's author :)
> > >>>>>>>>>>>>>>
> > >>>>>>>>>>>>>> I think it can always be hard to ensure that only specific
> > >>>>>>> errors
> > >>>>>>>>> are
> > >>>>>>>>>>>>>> handled unless they are explicitly enumerated in code as
> the
> > >>>>>>> code
> > >>>>>>>>> can
> > >>>>>>>>>>>>>> change and can be changed by folks who are not aware of
> this
> > >>>>>>> KIP
> > >>>>>>>> or
> > >>>>>>>>>>>>>> conversation.
> > >>>>>>>>>>>>>> I personally don't have the bandwidth to do this
> > >>>>>>>>>> definition/enumeration
> > >>>>>>>>>>>>> of
> > >>>>>>>>>>>>>> errors, so hopefully Alieh can expand upon this.
> > >>>>>>>>>>>>>>
> > >>>>>>>>>>>>>> Justine
> > >>>>>>>>>>>>>>
> > >>>>>>>>>>>>>> On Wed, Jul 3, 2024 at 8:28 AM Chris Egerton
> > >>>>>>>>> <chr...@aiven.io.invalid
> > >>>>>>>>>>>
> > >>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> Hi Alieh,
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> I don't love defining the changes for this KIP in terms
> of
> > a
> > >>>>>>>> catch
> > >>>>>>>>>>>>> clause
> > >>>>>>>>>>>>>>> in the KafkaProducer class, for two reasons. First, the
> set
> > >>>>> of
> > >>>>>>>>> errors
> > >>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>> are handled by that clause may shift over time as the
> code
> > >>>>>>> base
> > >>>>>>>> is
> > >>>>>>>>>>>>>>> modified, and second, it would be fairly opaque to users
> > who
> > >>>>>>> want
> > >>>>>>>>> to
> > >>>>>>>>>>>>>>> understand whether an error would be affected by using
> this
> > >>>>>>> API
> > >>>>>>>> or
> > >>>>>>>>>>>> not.
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> It also seems strange that we'd handle some types of
> > >>>>>>>>>>>>>>> RecordTooLargeException (i.e., ones reported client-side)
> > >>>>> with
> > >>>>>>>> this
> > >>>>>>>>>>>>> API,
> > >>>>>>>>>>>>>>> but not others (i.e., ones reported by a broker).
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> I think this kind of API would be most powerful, most
> > >>>>>>> intuitive
> > >>>>>>>> to
> > >>>>>>>>>>>>> users,
> > >>>>>>>>>>>>>>> and easiest to document if we expanded the scope to all
> > >>>>>>>>>>>>>> record-send-related
> > >>>>>>>>>>>>>>> errors, except anything indicating issues with
> exactly-once
> > >>>>>>>>>>>> semantics.
> > >>>>>>>>>>>>>> That
> > >>>>>>>>>>>>>>> would include records that are too large (when caught
> both
> > >>>>>>>> client-
> > >>>>>>>>>>>> and
> > >>>>>>>>>>>>>>> server-side), records that can't be sent due to
> > >>>>> authorization
> > >>>>>>>>>>>> failures,
> > >>>>>>>>>>>>>>> records sent to nonexistent topics/topic partitions, and
> > >>>>>>> keyless
> > >>>>>>>>>>>>> records
> > >>>>>>>>>>>>>>> sent to compacted topics. It would not include
> > >>>>>>>>>>>>>>> ProducerFencedException, InvalidProducerEpochException,
> > >>>>>>>>>>>>>>> UnsupportedVersionException,
> > >>>>>>>>>>>>>>> and possibly others.
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> @Justine -- do you think it would be possible to develop
> > >>>>>>> either a
> > >>>>>>>>>>>>> better
> > >>>>>>>>>>>>>>> definition for the kinds of "excluded" errors that should
> > >>>>> not
> > >>>>>>> be
> > >>>>>>>>>>>>> covered
> > >>>>>>>>>>>>>> by
> > >>>>>>>>>>>>>>> this API, or, barring that, a comprehensive list of exact
> > >>>>>>> error
> > >>>>>>>>>>>> types?
> > >>>>>>>>>>>>>> And
> > >>>>>>>>>>>>>>> do you think this would be acceptable in terms of risk
> and
> > >>>>>>>>>>>> complexity?
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> Cheers,
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> Chris
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> On Tue, Jul 2, 2024 at 5:05 PM Alieh Saeedi
> > >>>>>>>>>>>>> <asae...@confluent.io.invalid
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>> Hey Justine,
> > >>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>> About the consequences: the consequences will be like
> when
> > >>>>> we
> > >>>>>>>> did
> > >>>>>>>>>>>> not
> > >>>>>>>>>>>>>>> have
> > >>>>>>>>>>>>>>>> the fix made in `KAFKA-9279`: silent loss of data!
> > >>>>> Obviously,
> > >>>>>>>> when
> > >>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>> user
> > >>>>>>>>>>>>>>>> intentionally chose to ignore errors, that would not be
> > >>>>>>> silent
> > >>>>>>>> any
> > >>>>>>>>>>>>>> more.
> > >>>>>>>>>>>>>>>> Right?
> > >>>>>>>>>>>>>>>> Of course, considering all types of `ApiException`s
> would
> > >>>>> be
> > >>>>>>> too
> > >>>>>>>>>>>>> broad.
> > >>>>>>>>>>>>>>> But
> > >>>>>>>>>>>>>>>> are the exceptions caught in `catch(ApiException e)` of
> > the
> > >>>>>>>>>>>>> `doSend()`
> > >>>>>>>>>>>>>>>> method also too broad?
> > >>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>> -Alieh
> > >>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>> On Tue, Jul 2, 2024 at 9:45 PM Justine Olshan
> > >>>>>>>>>>>>>>> <jols...@confluent.io.invalid
> > >>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>> Hey Alieh,
> > >>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>> If we want to allow any error to be ignored we should
> > >>>>>>> probably
> > >>>>>>>>>>>> run
> > >>>>>>>>>>>>>>>> through
> > >>>>>>>>>>>>>>>>> all the errors to make sure they make sense.
> > >>>>>>>>>>>>>>>>> I just want to feel confident that we aren't just
> making
> > a
> > >>>>>>>>>>>> decision
> > >>>>>>>>>>>>>>>> without
> > >>>>>>>>>>>>>>>>> considering the consequences carefully.
> > >>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>> Justine
> > >>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>> On Tue, Jul 2, 2024 at 12:30 PM Alieh Saeedi
> > >>>>>>>>>>>>>>>> <asae...@confluent.io.invalid
> > >>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>> Hey Justine,
> > >>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>> yes we talked about `RecordTooLargeException` as an
> > >>>>>>> example,
> > >>>>>>>>>>>> but
> > >>>>>>>>>>>>>> did
> > >>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>> ever limit ourselves to only this specific exception?
> I
> > >>>>>>> think
> > >>>>>>>>>>>>>> neither
> > >>>>>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>> the KIP nor in the PR. As Chris mentioned, this KIP
> is
> > >>>>>>> going
> > >>>>>>>>>>>> to
> > >>>>>>>>>>>>>> undo
> > >>>>>>>>>>>>>>>>> what
> > >>>>>>>>>>>>>>>>>> we have done in `KAFKA-9279` in case 1) the user is
> in a
> > >>>>>>>>>>>>>> transaction
> > >>>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>>> 2)
> > >>>>>>>>>>>>>>>>>> he decides to ignore the errors in which the record
> was
> > >>>>> not
> > >>>>>>>>>>>> even
> > >>>>>>>>>>>>>>> added
> > >>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>> the batch. Yes, and we suggested some methods for
> > undoing
> > >>>>>>> or,
> > >>>>>>>>>>>> in
> > >>>>>>>>>>>>>>> fact,
> > >>>>>>>>>>>>>>>>>> moving back the transaction from the error state in
> > >>>>>>> `flush` or
> > >>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>> `commitTnx` and we finally came to the idea of not
> even
> > >>>>>>> doing
> > >>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>> changes
> > >>>>>>>>>>>>>>>>>> (better than undoing) in `send`.
> > >>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>> Bests,
> > >>>>>>>>>>>>>>>>>> Alieh
> > >>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>> On Tue, Jul 2, 2024 at 8:03 PM Justine Olshan
> > >>>>>>>>>>>>>>>>> <jols...@confluent.io.invalid
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>> Hey folks,
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>> I understand where you are coming from by asking for
> > >>>>>>> specific
> > >>>>>>>>>>>>> use
> > >>>>>>>>>>>>>>>>> cases.
> > >>>>>>>>>>>>>>>>>> My
> > >>>>>>>>>>>>>>>>>>> understanding based on previous conversations was
> that
> > >>>>>>> there
> > >>>>>>>>>>>>>> were a
> > >>>>>>>>>>>>>>>> few
> > >>>>>>>>>>>>>>>>>>> different errors that have been seen.
> > >>>>>>>>>>>>>>>>>>> One example I heard some information about was when
> the
> > >>>>>>>>>>>> record
> > >>>>>>>>>>>>>> was
> > >>>>>>>>>>>>>>>> too
> > >>>>>>>>>>>>>>>>>>> large and it fails the batch. Besides that, I'm not
> > >>>>> really
> > >>>>>>>>>>>> sure
> > >>>>>>>>>>>>>> if
> > >>>>>>>>>>>>>>>>> there
> > >>>>>>>>>>>>>>>>>>> are cases in mind, though it is fair to ask on those
> > and
> > >>>>>>>>>>>> bring
> > >>>>>>>>>>>>>> them
> > >>>>>>>>>>>>>>>> up.
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>> Does a record qualify as a poison pill if it
> targets a
> > >>>>>>>>>>>> topic
> > >>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>> doesn't exist? Or if it targets a topic that the
> > >>>>> producer
> > >>>>>>>>>>>>>> principal
> > >>>>>>>>>>>>>>>>> lacks
> > >>>>>>>>>>>>>>>>>>> ACLs for? What if it fails broker-side validation
> > (e.g.,
> > >>>>>>> has
> > >>>>>>>>>>>> a
> > >>>>>>>>>>>>>> null
> > >>>>>>>>>>>>>>>> key
> > >>>>>>>>>>>>>>>>>> for
> > >>>>>>>>>>>>>>>>>>> a compacted topic)?
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>> I think there was some parallel work with addressing
> > the
> > >>>>>>>>>>>>>>>>>>> UnknownTopicOrPartitionError in another way. As for
> the
> > >>>>>>> other
> > >>>>>>>>>>>>>>> checks,
> > >>>>>>>>>>>>>>>>>> acls,
> > >>>>>>>>>>>>>>>>>>> validation etc. I am not aware of that being in
> Alieh's
> > >>>>>>>>>>>> scope,
> > >>>>>>>>>>>>>> but
> > >>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>> should be clear about exactly what we are doing.
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>> All errors that fall into ApiException seems too
> broad
> > >>>>> to
> > >>>>>>> me.
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>> Justine
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>> On Tue, Jul 2, 2024 at 10:51 AM Alieh Saeedi
> > >>>>>>>>>>>>>>>>>> <asae...@confluent.io.invalid
> > >>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>> Hey Chris,
> > >>>>>>>>>>>>>>>>>>>> thanks for sharing your concerns.
> > >>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>> 1) About the language of KIP (or maybe later in
> > >>>>>>> Javadocs):
> > >>>>>>>>>>>> Is
> > >>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>> alright
> > >>>>>>>>>>>>>>>>>>>> if I write all errors that fall into the
> > `ApiException`
> > >>>>>>>>>>>>>> category
> > >>>>>>>>>>>>>>>>> thrown
> > >>>>>>>>>>>>>>>>>>>> (actually returned) by Producer?
> > >>>>>>>>>>>>>>>>>>>> 2) About future expansion: do you have any better
> > >>>>>>>>>>>> suggestions
> > >>>>>>>>>>>>>> for
> > >>>>>>>>>>>>>>>>> enum
> > >>>>>>>>>>>>>>>>>>>> names? Do you think `IGNORE_API_EXEPTIONS` or
> > something
> > >>>>>>>>>>>> like
> > >>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>>>>>> "better/more accurate" one?
> > >>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>> Bests,
> > >>>>>>>>>>>>>>>>>>>> Alieh
> > >>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>> On Tue, Jul 2, 2024 at 7:29 PM Chris Egerton
> > >>>>>>>>>>>>>>>> <chr...@aiven.io.invalid
> > >>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> Hi Alieh and Justine,
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> I'm concerned that we're settling on a definition
> of
> > >>>>>>>>>>>>> "poison
> > >>>>>>>>>>>>>>>> pill"
> > >>>>>>>>>>>>>>>>>>> that's
> > >>>>>>>>>>>>>>>>>>>>> easiest to tackle right now but may lead to
> > >>>>> shortcomings
> > >>>>>>>>>>>>> down
> > >>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>> road. I
> > >>>>>>>>>>>>>>>>>>>>> understand the relationship between this KIP and
> > >>>>>>>>>>>>> KAFKA-9279,
> > >>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>> I
> > >>>>>>>>>>>>>>>>>> can
> > >>>>>>>>>>>>>>>>>>>>> totally get behind the desire to keep things small,
> > >>>>>>>>>>>>> focused,
> > >>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>>>> simple
> > >>>>>>>>>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>>>>> the name of avoiding bugs. However, what I don't
> > think
> > >>>>>>> is
> > >>>>>>>>>>>>>> clear
> > >>>>>>>>>>>>>>>> at
> > >>>>>>>>>>>>>>>>>> all
> > >>>>>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>>>> what the "specific circumstances" are that Justine
> > >>>>>>>>>>>>>> mentioned. I
> > >>>>>>>>>>>>>>>>> had a
> > >>>>>>>>>>>>>>>>>>>>> drastically different idea of what the intended
> > >>>>>>>>>>>> behavioral
> > >>>>>>>>>>>>>>> change
> > >>>>>>>>>>>>>>>>>> would
> > >>>>>>>>>>>>>>>>>>>> be
> > >>>>>>>>>>>>>>>>>>>>> before looking at the draft PR.
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> I would like 1) for us to be clearer about the
> > >>>>>>> categories
> > >>>>>>>>>>>>> of
> > >>>>>>>>>>>>>>>> errors
> > >>>>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>>>> want to cover with this new API (especially since
> > >>>>> we'll
> > >>>>>>>>>>>>> have
> > >>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>> find
> > >>>>>>>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>>>>>>> clear, succinct way to document this for users),
> and
> > >>>>> 2)
> > >>>>>>>>>>>> to
> > >>>>>>>>>>>>>> make
> > >>>>>>>>>>>>>>>>> sure
> > >>>>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>>> if we do try to expand this API in the future, that
> > we
> > >>>>>>>>>>>>> won't
> > >>>>>>>>>>>>>> be
> > >>>>>>>>>>>>>>>>>> painted
> > >>>>>>>>>>>>>>>>>>>>> into a corner.
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> For item 1, hopefully we can agree that the
> language
> > >>>>> in
> > >>>>>>>>>>>> the
> > >>>>>>>>>>>>>> KIP
> > >>>>>>>>>>>>>>>>>>>>> for IGNORE_SEND_ERRORS ("The records causing
> > >>>>>>>>>>>> irrecoverable
> > >>>>>>>>>>>>>>> errors
> > >>>>>>>>>>>>>>>>> are
> > >>>>>>>>>>>>>>>>>>>>> excluded from the batch and the transaction is
> > >>>>> committed
> > >>>>>>>>>>>>>>>>>>> successfully.")
> > >>>>>>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>>>> pretty vague. If we start using the phrase "poison
> > >>>>> pill
> > >>>>>>>>>>>>>> record"
> > >>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>> could
> > >>>>>>>>>>>>>>>>>>>>> help, but IMO more detail would still be needed. We
> > >>>>> know
> > >>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>> want
> > >>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>> include records that are so large that they can be
> > >>>>>>>>>>>>>> immediately
> > >>>>>>>>>>>>>>>>>> rejected
> > >>>>>>>>>>>>>>>>>>>> by
> > >>>>>>>>>>>>>>>>>>>>> the producer. But there are other cases that users
> > >>>>> might
> > >>>>>>>>>>>>>> expect
> > >>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>> be
> > >>>>>>>>>>>>>>>>>>>>> handled. Does a record qualify as a poison pill if
> it
> > >>>>>>>>>>>>>> targets a
> > >>>>>>>>>>>>>>>>> topic
> > >>>>>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>>> doesn't exist? Or if it targets a topic that the
> > >>>>>>> producer
> > >>>>>>>>>>>>>>>> principal
> > >>>>>>>>>>>>>>>>>>> lacks
> > >>>>>>>>>>>>>>>>>>>>> ACLs for? What if it fails broker-side validation
> > >>>>> (e.g.,
> > >>>>>>>>>>>>> has
> > >>>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>> null
> > >>>>>>>>>>>>>>>>>> key
> > >>>>>>>>>>>>>>>>>>>> for
> > >>>>>>>>>>>>>>>>>>>>> a compacted topic)?
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> For item 2, this really depends on how narrow the
> > >>>>> scope
> > >>>>>>>>>>>> of
> > >>>>>>>>>>>>>> what
> > >>>>>>>>>>>>>>>>> we're
> > >>>>>>>>>>>>>>>>>>>> doing
> > >>>>>>>>>>>>>>>>>>>>> right now is. If we only handle a subset of the
> > >>>>> examples
> > >>>>>>>>>>>> I
> > >>>>>>>>>>>>>> laid
> > >>>>>>>>>>>>>>>> out
> > >>>>>>>>>>>>>>>>>>> above
> > >>>>>>>>>>>>>>>>>>>>> that could possibly be considered poison pills with
> > >>>>> this
> > >>>>>>>>>>>>> KIP,
> > >>>>>>>>>>>>>>> do
> > >>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>> want
> > >>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>> lock ourselves in to never addressing more in the
> > >>>>>>> future,
> > >>>>>>>>>>>>> or
> > >>>>>>>>>>>>>>> can
> > >>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>>> choose
> > >>>>>>>>>>>>>>>>>>>>> an API (probably just enum names would be the only
> > >>>>>>>>>>>>> important
> > >>>>>>>>>>>>>>>>> decision
> > >>>>>>>>>>>>>>>>>>>> here)
> > >>>>>>>>>>>>>>>>>>>>> that leaves room for more later?
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> Best,
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> Chris
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> On Tue, Jul 2, 2024 at 12:28 PM Justine Olshan
> > >>>>>>>>>>>>>>>>>>>>> <jols...@confluent.io.invalid>
> > >>>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>> Chris and Alieh,
> > >>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>> My understanding is that this KIP is really only
> > >>>>> trying
> > >>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>> solve
> > >>>>>>>>>>>>>>>>> an
> > >>>>>>>>>>>>>>>>>>>> issue
> > >>>>>>>>>>>>>>>>>>>>>> of a "poison pill" record that fails send().
> > >>>>>>>>>>>>>>>>>>>>>> We've talked a lot about having a generic
> framework
> > >>>>> for
> > >>>>>>>>>>>>> all
> > >>>>>>>>>>>>>>>>> errors,
> > >>>>>>>>>>>>>>>>>>>> but I
> > >>>>>>>>>>>>>>>>>>>>>> don't think that is what this KIP is trying to do.
> > >>>>>>>>>>>>>>> Essentially
> > >>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>> request
> > >>>>>>>>>>>>>>>>>>>>>> is to undo the change from KAFKA-9279
> > >>>>>>>>>>>>>>>>>>>>>> <https://issues.apache.org/jira/browse/KAFKA-9279
> >
> > >>>>> but
> > >>>>>>>>>>>>>> under
> > >>>>>>>>>>>>>>>>>>> specific
> > >>>>>>>>>>>>>>>>>>>>>> circumstances that are controlled. I really am
> > >>>>>>>>>>>> concerned
> > >>>>>>>>>>>>>>> about
> > >>>>>>>>>>>>>>>>>>> opening
> > >>>>>>>>>>>>>>>>>>>>> new
> > >>>>>>>>>>>>>>>>>>>>>> avenues for bugs with EOS and hesitate to handle
> any
> > >>>>>>>>>>>>> other
> > >>>>>>>>>>>>>>>> types
> > >>>>>>>>>>>>>>>>> of
> > >>>>>>>>>>>>>>>>>>>>> errors.
> > >>>>>>>>>>>>>>>>>>>>>> I think if we all agree on the problem that we are
> > >>>>>>>>>>>> trying
> > >>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>> solve,
> > >>>>>>>>>>>>>>>>>>> it
> > >>>>>>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>>>>> easier to agree on solutions.
> > >>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>> Justine
> > >>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>> On Mon, Jul 1, 2024 at 2:20 AM Alieh Saeedi
> > >>>>>>>>>>>>>>>>>>>> <asae...@confluent.io.invalid
> > >>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>> Hi Matthias,
> > >>>>>>>>>>>>>>>>>>>>>>> Thanks for the valid points you mentioned. I
> > updated
> > >>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>> KIP
> > >>>>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>> PR
> > >>>>>>>>>>>>>>>>>>>>>>> with:
> > >>>>>>>>>>>>>>>>>>>>>>> 1) mentioning that the new overloaded `send`
> throws
> > >>>>>>>>>>>>>>>>>>>>>> `IllegalStateException`
> > >>>>>>>>>>>>>>>>>>>>>>> if the user tries to ignore `send()` errors
> outside
> > >>>>>>>>>>>> of
> > >>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>>>>> transaction.
> > >>>>>>>>>>>>>>>>>>>>>>> 2) the default implementation in `Producer`
> > >>>>> interface
> > >>>>>>>>>>>>>>> throws
> > >>>>>>>>>>>>>>>> an
> > >>>>>>>>>>>>>>>>>>>>>>> `UnsupportedOperationException`
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>> Hi Chris,
> > >>>>>>>>>>>>>>>>>>>>>>> Thanks for the feedback. I tried to clarify the
> > >>>>>>>>>>>> points
> > >>>>>>>>>>>>>> you
> > >>>>>>>>>>>>>>>>>> listed:
> > >>>>>>>>>>>>>>>>>>>>>>> -------> we've narrowed the scope from any error
> > >>>>> that
> > >>>>>>>>>>>>>> might
> > >>>>>>>>>>>>>>>>> take
> > >>>>>>>>>>>>>>>>>>>> place
> > >>>>>>>>>>>>>>>>>>>>>> with
> > >>>>>>>>>>>>>>>>>>>>>>> producing a record to Kafka, to only the ones
> that
> > >>>>>>>>>>>> are
> > >>>>>>>>>>>>>>> thrown
> > >>>>>>>>>>>>>>>>>>>> directly
> > >>>>>>>>>>>>>>>>>>>>>> from
> > >>>>>>>>>>>>>>>>>>>>>>> Producer::send;
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>> From the very beginning and even since
> KIP-1038,
> > >> the
> > >>>>>>>>>>>>> main
> > >>>>>>>>>>>>>>>>> purpose
> > >>>>>>>>>>>>>>>>>>> was
> > >>>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>>> have "more flexibility," or, in other words,
> > "giving
> > >>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>> user
> > >>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>> authority" to handle some specific exceptions
> > thrown
> > >>>>>>>>>>>>> from
> > >>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>> `Producer`.
> > >>>>>>>>>>>>>>>>>>>>>>> Due to the specific cases we had in mind,
> KIP-1038
> > >>>>>>>>>>>> was
> > >>>>>>>>>>>>>>>>> discarded
> > >>>>>>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>>>>>> decided to not define a `CustomExceptionHandler`
> > for
> > >>>>>>>>>>>>>>>> `Producer`
> > >>>>>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>>>>>>>> instead
> > >>>>>>>>>>>>>>>>>>>>>>> treat the `send` failures in a different way. The
> > >>>>>>>>>>>> main
> > >>>>>>>>>>>>>>> issue
> > >>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>>>> `send`
> > >>>>>>>>>>>>>>>>>>>>>>> makes a transition to error state, which is
> > >>>>> undoable.
> > >>>>>>>>>>>>> In
> > >>>>>>>>>>>>>>>> fact,
> > >>>>>>>>>>>>>>>>>> one
> > >>>>>>>>>>>>>>>>>>>>> single
> > >>>>>>>>>>>>>>>>>>>>>>> poison pill record makes the whole batch fail.
> The
> > >>>>>>>>>>>>> former
> > >>>>>>>>>>>>>>>>>>> suggestions
> > >>>>>>>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>>>>> you agreed with have been all about un-doing this
> > >>>>>>>>>>>>>>> transition
> > >>>>>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>>>> `flush`
> > >>>>>>>>>>>>>>>>>>>>>> or
> > >>>>>>>>>>>>>>>>>>>>>>> `commit`. The new suggestion is to un-do (or
> > better,
> > >>>>>>>>>>>>> NOT
> > >>>>>>>>>>>>>>> do)
> > >>>>>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>>>> `send`
> > >>>>>>>>>>>>>>>>>>>>>> due
> > >>>>>>>>>>>>>>>>>>>>>>> to the reasons listed in the discussions above.
> > >>>>>>>>>>>>>>>>>>>>>>> Moreover, I would say that having such a large
> > scope
> > >>>>>>>>>>>> as
> > >>>>>>>>>>>>>> you
> > >>>>>>>>>>>>>>>>>>> mentioned
> > >>>>>>>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>>>>>> impossible. In the best case, we may have control
> > >>>>>>>>>>>> over
> > >>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>> `Producer`.
> > >>>>>>>>>>>>>>>>>>>>>> What
> > >>>>>>>>>>>>>>>>>>>>>>> shall we do with the broker? The `any error that
> > >>>>>>>>>>>> might
> > >>>>>>>>>>>>>> take
> > >>>>>>>>>>>>>>>>> place
> > >>>>>>>>>>>>>>>>>>>> with
> > >>>>>>>>>>>>>>>>>>>>>>> producing a record to Kafka` is too much, I
> think.
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>> -------> is this all we want to handle, and will
> it
> > >>>>>>>>>>>>>> prevent
> > >>>>>>>>>>>>>>>> us
> > >>>>>>>>>>>>>>>>>> from
> > >>>>>>>>>>>>>>>>>>>>>>> handling more in the future in an intuitive way?
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>> I think yes. This is all we want. Other sorts of
> > >>>>>>>>>>>> errors
> > >>>>>>>>>>>>>>> such
> > >>>>>>>>>>>>>>>> as
> > >>>>>>>>>>>>>>>>>>>> having
> > >>>>>>>>>>>>>>>>>>>>>>> problem with partition addition, producer fenced
> > >>>>>>>>>>>>>> exception,
> > >>>>>>>>>>>>>>>> etc
> > >>>>>>>>>>>>>>>>>>> seem
> > >>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>> be
> > >>>>>>>>>>>>>>>>>>>>>>> more serious issues. The intention was to handle
> > >>>>>>>>>>>>> problems
> > >>>>>>>>>>>>>>>>> created
> > >>>>>>>>>>>>>>>>>>> by
> > >>>>>>>>>>>>>>>>>>>>>>> (maybe) a single poison pill record. BTW, I do
> not
> > >>>>>>>>>>>> see
> > >>>>>>>>>>>>>> any
> > >>>>>>>>>>>>>>>>>>> obstacles
> > >>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>>> future changes.
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>> Bests,
> > >>>>>>>>>>>>>>>>>>>>>>> Alieh
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>> On Sat, Jun 29, 2024 at 3:03 AM Chris Egerton
> > >>>>>>>>>>>>>>>>>>>> <chr...@aiven.io.invalid
> > >>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>> Ah, sorry--spoke too soon. The PR doesn't show
> > that
> > >>>>>>>>>>>>>>> errors
> > >>>>>>>>>>>>>>>>>> thrown
> > >>>>>>>>>>>>>>>>>>>>> from
> > >>>>>>>>>>>>>>>>>>>>>>>> Producer::send are handled, but instead,
> > >>>>>>>>>>>> ApiException
> > >>>>>>>>>>>>>>>>> instances
> > >>>>>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>>>> are
> > >>>>>>>>>>>>>>>>>>>>>>>> caught inside KafkaProducer::doSend and are
> > handled
> > >>>>>>>>>>>>> by
> > >>>>>>>>>>>>>>>>>> returning
> > >>>>>>>>>>>>>>>>>>> an
> > >>>>>>>>>>>>>>>>>>>>>>>> already-failed future are. I think the same
> > >>>>>>>>>>>> question
> > >>>>>>>>>>>>>>> still
> > >>>>>>>>>>>>>>>>>>> applies
> > >>>>>>>>>>>>>>>>>>>>> (is
> > >>>>>>>>>>>>>>>>>>>>>>> this
> > >>>>>>>>>>>>>>>>>>>>>>>> all we want to handle, and will it prevent us
> from
> > >>>>>>>>>>>>>>> handling
> > >>>>>>>>>>>>>>>>>> more
> > >>>>>>>>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>> future in an intuitive way), though.
> > >>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>> On Fri, Jun 28, 2024 at 8:57 PM Chris Egerton <
> > >>>>>>>>>>>>>>>>> chr...@aiven.io
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>> Hi Alieh,
> > >>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>> This KIP has evolved a lot since I last looked
> at
> > >>>>>>>>>>>>> it,
> > >>>>>>>>>>>>>>> but
> > >>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>> changes
> > >>>>>>>>>>>>>>>>>>>>>>>> seem
> > >>>>>>>>>>>>>>>>>>>>>>>>> well thought-out both in semantics and API. One
> > >>>>>>>>>>>>>>>> clarifying
> > >>>>>>>>>>>>>>>>>>>>> question I
> > >>>>>>>>>>>>>>>>>>>>>>>> have
> > >>>>>>>>>>>>>>>>>>>>>>>>> is that it looks based on the draft PR that
> we've
> > >>>>>>>>>>>>>>>> narrowed
> > >>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>> scope
> > >>>>>>>>>>>>>>>>>>>>>>> from
> > >>>>>>>>>>>>>>>>>>>>>>>>> any error that might take place with producing
> a
> > >>>>>>>>>>>>>> record
> > >>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>> Kafka,
> > >>>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>>> only
> > >>>>>>>>>>>>>>>>>>>>>>>>> the ones that are thrown directly from
> > >>>>>>>>>>>>>> Producer::send;
> > >>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>> intended
> > >>>>>>>>>>>>>>>>>>>>>>>>> behavior here? And if so, do you have thoughts
> on
> > >>>>>>>>>>>>> how
> > >>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>> might
> > >>>>>>>>>>>>>>>>>>>>>> design a
> > >>>>>>>>>>>>>>>>>>>>>>>>> follow-up KIP that would catch all errors
> > >>>>>>>>>>>>> (including
> > >>>>>>>>>>>>>>> ones
> > >>>>>>>>>>>>>>>>>>>> reported
> > >>>>>>>>>>>>>>>>>>>>>>>>> asynchronously instead of synchronously)? I'd
> > >>>>>>>>>>>> like
> > >>>>>>>>>>>>> it
> > >>>>>>>>>>>>>>> if
> > >>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>> could
> > >>>>>>>>>>>>>>>>>>>>>> leave
> > >>>>>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>>> door open for that without painting ourselves
> > >>>>>>>>>>>> into
> > >>>>>>>>>>>>>> too
> > >>>>>>>>>>>>>>>> much
> > >>>>>>>>>>>>>>>>>> of
> > >>>>>>>>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>>>>>>>> corner
> > >>>>>>>>>>>>>>>>>>>>>>>>> with the API design for this KIP.
> > >>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>> Cheers,
> > >>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>> Chris
> > >>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>> On Fri, Jun 28, 2024 at 6:31 PM Matthias J.
> Sax <
> > >>>>>>>>>>>>>>>>>>>> mj...@apache.org>
> > >>>>>>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>> Thanks Alieh,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>> it seems this KIP can just pick between a
> couple
> > >>>>>>>>>>>>> of
> > >>>>>>>>>>>>>>>>>> tradeoffs.
> > >>>>>>>>>>>>>>>>>>>>>> Adding
> > >>>>>>>>>>>>>>>>>>>>>>> an
> > >>>>>>>>>>>>>>>>>>>>>>>>>> overloaded `send()` as the KIP propose makes
> > >>>>>>>>>>>> sense
> > >>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>> me
> > >>>>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>>>>>> seems
> > >>>>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>>>>>> provides the cleanest solution compare to
> there
> > >>>>>>>>>>>>>>> options
> > >>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>>>>> discussed.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>> Given the explicit name of the passed-in
> option
> > >>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>> highlights
> > >>>>>>>>>>>>>>>>>>>>> that
> > >>>>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>>>> option is for TX only make is pretty clear and
> > >>>>>>>>>>>>>> avoids
> > >>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>> issue
> > >>>>>>>>>>>>>>>>>>>> of
> > >>>>>>>>>>>>>>>>>>>>>>>>>> `flush()` ambiguity.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>> Nit: We should make clear on the KIP though,
> > >>>>>>>>>>>> that
> > >>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>> new
> > >>>>>>>>>>>>>>>>>>>> `send()`
> > >>>>>>>>>>>>>>>>>>>>>>>>>> overload would throw an
> `IllegalStateException`
> > >>>>>>>>>>>> if
> > >>>>>>>>>>>>>> TX
> > >>>>>>>>>>>>>>>> are
> > >>>>>>>>>>>>>>>>>> not
> > >>>>>>>>>>>>>>>>>>>> used
> > >>>>>>>>>>>>>>>>>>>>>>>>>> (similar to other TX methods like initTx(),
> etc)
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>> About the `Producer` interface, I am not sure
> > >>>>>>>>>>>> how
> > >>>>>>>>>>>>>> this
> > >>>>>>>>>>>>>>>> was
> > >>>>>>>>>>>>>>>>>>> done
> > >>>>>>>>>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>>>> past (eg, KIP-266 added
> > >>>>>>>>>>>> `Consumer.poll(Duration)`
> > >>>>>>>>>>>>>> w/o
> > >>>>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>>>>> default
> > >>>>>>>>>>>>>>>>>>>>>>>>>> implementation), if we need a default
> > >>>>>>>>>>>>> implementation
> > >>>>>>>>>>>>>>> for
> > >>>>>>>>>>>>>>>>>>>> backward
> > >>>>>>>>>>>>>>>>>>>>>>>>>> compatibility or not? If we do want to add
> one,
> > >>>>>>>>>>>> I
> > >>>>>>>>>>>>>>> think
> > >>>>>>>>>>>>>>>> it
> > >>>>>>>>>>>>>>>>>>> would
> > >>>>>>>>>>>>>>>>>>>>> be
> > >>>>>>>>>>>>>>>>>>>>>>>>>> appropriate to throw an
> > >>>>>>>>>>>>>>> `UnsupportedOperationException`
> > >>>>>>>>>>>>>>>> by
> > >>>>>>>>>>>>>>>>>>>>> default,
> > >>>>>>>>>>>>>>>>>>>>>>>>>> instead of just keeping the default impl
> empty?
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>> My points are rather minor, and should not
> block
> > >>>>>>>>>>>>>> this
> > >>>>>>>>>>>>>>>> KIP
> > >>>>>>>>>>>>>>>>>>>> though.
> > >>>>>>>>>>>>>>>>>>>>>>>>>> Overall LGTM.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>> -Matthias
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>> On 6/27/24 1:28 PM, Alieh Saeedi wrote:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> Hi Justine,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> Thanks for the suggestion.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> Making applications to validate every single
> > >>>>>>>>>>>>>> record
> > >>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>> not
> > >>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>> best
> > >>>>>>>>>>>>>>>>>>>>>>>> way,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> from an efficiency point of view.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> Moreover, between changing the behavior of
> the
> > >>>>>>>>>>>>>>>> Producer
> > >>>>>>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>>>>> `send`
> > >>>>>>>>>>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> `commitTnx`, the former seems more reasonable
> > >>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>> clean.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> Bests,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> Alieh
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> On Thu, Jun 27, 2024 at 8:14 PM Justine
> Olshan
> > >>>>>>>>>>>>>>>>>>>>>>>>>> <jols...@confluent.io.invalid>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> Hey Alieh,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> I see there are two options now. So folks
> > >>>>>>>>>>>> will
> > >>>>>>>>>>>>> be
> > >>>>>>>>>>>>>>>>>>> discussing
> > >>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>>>> approaches
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> and deciding the best way forward before we
> > >>>>>>>>>>>>> vote?
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> I do think there could be a problem with the
> > >>>>>>>>>>>>>>> approach
> > >>>>>>>>>>>>>>>>> on
> > >>>>>>>>>>>>>>>>>>>> commit
> > >>>>>>>>>>>>>>>>>>>>>> if
> > >>>>>>>>>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>>>>>>>>> get
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> stuck on an earlier error and have more
> > >>>>>>>>>>>> records
> > >>>>>>>>>>>>>>>>>>> (potentially
> > >>>>>>>>>>>>>>>>>>>> on
> > >>>>>>>>>>>>>>>>>>>>>> new
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> partitions) to commit as the current PR is
> > >>>>>>>>>>>>>>>> implemented.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> I guess this takes us back to the question
> of
> > >>>>>>>>>>>>>>> whether
> > >>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>> error
> > >>>>>>>>>>>>>>>>>>>>>>>> should
> > >>>>>>>>>>>>>>>>>>>>>>>>>> be
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> cleared on send.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> (And I guess at the back of my mind, I'm
> > >>>>>>>>>>>>>> wondering
> > >>>>>>>>>>>>>>> if
> > >>>>>>>>>>>>>>>>>> there
> > >>>>>>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>>>>>>>>> way
> > >>>>>>>>>>>>>>>>>>>>>>>>>> we can
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> validate the "posion pill" records
> > >>>>>>>>>>>> application
> > >>>>>>>>>>>>>> side
> > >>>>>>>>>>>>>>>>>> before
> > >>>>>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>>>>> even
> > >>>>>>>>>>>>>>>>>>>>>>>> try
> > >>>>>>>>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> send them)
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> Justine
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> On Wed, Jun 26, 2024 at 4:38 PM Alieh Saeedi
> > >>>>>>>>>>>>>>>>>>>>>>>>>> <asae...@confluent.io.invalid
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Hi Justine,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> I did not update the KIP with
> > >>>>>>>>>>>> `TxnSendOption`
> > >>>>>>>>>>>>>>> since
> > >>>>>>>>>>>>>>>> I
> > >>>>>>>>>>>>>>>>>>>> thought
> > >>>>>>>>>>>>>>>>>>>>>> it'd
> > >>>>>>>>>>>>>>>>>>>>>>>> be
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> better discussed here beforehand.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> right now, there are 2 PRs:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> - the PR that implements the current
> version
> > >>>>>>>>>>>>> of
> > >>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>> KIP:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> https://github.com/apache/kafka/pull/16332
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> - the POC PR that clarifies the
> > >>>>>>>>>>>>> `TxnSendOption`:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> https://github.com/apache/kafka/pull/16465
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Bests,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> Alieh
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> On Thu, Jun 27, 2024 at 12:42 AM Justine
> > >>>>>>>>>>>>> Olshan
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> <jols...@confluent.io.invalid> wrote:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Hey Alieh,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> I think I am a little confused. Are the 3
> > >>>>>>>>>>>>>> points
> > >>>>>>>>>>>>>>>>> above
> > >>>>>>>>>>>>>>>>>>>>>> addressed
> > >>>>>>>>>>>>>>>>>>>>>>> by
> > >>>>>>>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> KIP
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> or did something change? The PR seems to
> > >>>>>>>>>>>> not
> > >>>>>>>>>>>>>>>> include
> > >>>>>>>>>>>>>>>>>> this
> > >>>>>>>>>>>>>>>>>>>>>> change
> > >>>>>>>>>>>>>>>>>>>>>>>> and
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> still
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> has the CommitOption as well.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Thanks,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Justine
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> On Wed, Jun 26, 2024 at 2:15 PM Alieh
> > >>>>>>>>>>>> Saeedi
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> <asae...@confluent.io.invalid
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> wrote:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Hi all,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Looking at the PR <
> > >>>>>>>>>>>>>>>>>>>>> https://github.com/apache/kafka/pull/16332
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> corresponding to the KIP, there are some
> > >>>>>>>>>>>>>> points
> > >>>>>>>>>>>>>>>>> worthy
> > >>>>>>>>>>>>>>>>>>> of
> > >>>>>>>>>>>>>>>>>>>>>>> mention:
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 1) clearing the error ends up dirty/messy
> > >>>>>>>>>>>>> code
> > >>>>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> `TransactionManager`.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 2) By clearing the error, we are actually
> > >>>>>>>>>>>>>> doing
> > >>>>>>>>>>>>>>> an
> > >>>>>>>>>>>>>>>>>>> illegal
> > >>>>>>>>>>>>>>>>>>>>>>>>>> transition
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> from
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> `ABORTABLE_ERROR` to `IN_TRANSACTION`
> > >>>>>>>>>>>> which
> > >>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>>> conceptually
> > >>>>>>>>>>>>>>>>>>>>>> not
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> acceptable.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> This can be the root cause of some
> issues,
> > >>>>>>>>>>>>>> with
> > >>>>>>>>>>>>>>>>>> perhaps
> > >>>>>>>>>>>>>>>>>>>>>> further
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> future
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> changes by others.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> 3) If the poison pill record `r1` causes
> a
> > >>>>>>>>>>>>>>>>> transition
> > >>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>> error
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> state
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> and then the next record `r2` requires
> > >>>>>>>>>>>>> adding
> > >>>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>>>>> partition
> > >>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> transaction, the action fails due to
> being
> > >>>>>>>>>>>>> in
> > >>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>> error
> > >>>>>>>>>>>>>>>>>>>>> state.
> > >>>>>>>>>>>>>>>>>>>>>>> In
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> this
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> case, clearing errors during
> > >>>>>>>>>>>>>>>>>>> `commitTnx(CLEAR_SEND_ERROR)`
> > >>>>>>>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>>>>>> too
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> late.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> However, this case can NOT be the main
> > >>>>>>>>>>>>> concern
> > >>>>>>>>>>>>>>> as
> > >>>>>>>>>>>>>>>>> soon
> > >>>>>>>>>>>>>>>>>>> as
> > >>>>>>>>>>>>>>>>>>>>>>> KIP-890
> > >>>>>>>>>>>>>>>>>>>>>>>> is
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> fully
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> implemented.
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> My suggestion is to solve the problem
> > >>>>>>>>>>>> where
> > >>>>>>>>>>>>> it
> > >>>>>>>>>>>>>>>>> arises.
> > >>>>>>>>>>>>>>>>>>> If
> > >>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> transition
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> the error state does not happen during
> > >>>>>>>>>>>>>> `send()`,
> > >>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>> do
> > >>>>>>>>>>>>>>>>>>> not
> > >>>>>>>>>>>>>>>>>>>>>> need
> > >>>>>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>> clear
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> the error later. Therefore, instead of
> > >>>>>>>>>>>>>>>>> `CommitOption`,
> > >>>>>>>>>>>>>>>>>>> we
> > >>>>>>>>>>>>>>>>>>>>> can
> > >>>>>>>>>>>>>>>>>>>>>>>> define
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> a
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> `TxnSendOption` and prevent the `send()`
> > >>>>>>>>>>>>>> method
> > >>>>>>>>>>>>>>>> from
> > >>>>>>>>>>>>>>>>>>> going
> > >>>>>>>>>>>>>>>>>>>>> to
> > >>>>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> error
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> state in case 1) we're in a transaction
> > >>>>>>>>>>>> and
> > >>>>>>>>>>>>> 2)
> > >>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>> user
> > >>>>>>>>>>>>>>>>>>>>> asked
> > >>>>>>>>>>>>>>>>>>>>>>> for
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> IGONRE_SEND_ERRORS. For more clarity, you
> > >>>>>>>>>>>>> can
> > >>>>>>>>>>>>>>>> take a
> > >>>>>>>>>>>>>>>>>>> look
> > >>>>>>>>>>>>>>>>>>>> at
> > >>>>>>>>>>>>>>>>>>>>>> the
> > >>>>>>>>>>>>>>>>>>>>>>>> POC
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>> PR
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> <
> > >>>>>>>>>>>> https://github.com/apache/kafka/pull/16465
> > >>>>>>>>>>>>>> .
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Cheers,
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>> Alieh
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>>
> > >>>>>>>>>>>>>>
> > >>>>>>>>>>>>>
> > >>>>>>>>>>>>
> > >>>>>>>>>>
> > >>>>>>>>>>
> > >>>>>>>>>
> > >>>>>>>>
> > >>>>>>>
> > >>>>>>
> > >>>>>
> > >>>>
> > >>>
> > >>
> > >
> >
>
