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
